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 | import { configError, transportError } from './errors.js';
import { encode, requestUrl } from './http.js';
import * as p2p from './p2p.js';
/**
* Mint a mailbox token for APP-ID and return the credentials directly —
* `{ token, ttl, expires_at, endpoint, room? }` — ready to pass to
* `deposit` and `replay`. BODY may scope the token to one `room` and set
* `ttl_sec` (default 3600, capped at 86400). Mint short and reuse one
* across calls: every mint is a charged request.
*/
export const credentials = async (client, appId, body) => (await p2p.mintMailboxCredentials(client, appId, body)).credentials;
/**
* 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 unescapes it back
* into one room.
*/
const encodeRoom = (room) => {
if (!room?.trim())
throw configError('Mailbox room must be a non-empty string', { room });
return encode(room);
};
const messagesUrl = (creds, room) => {
if (!creds.endpoint?.trim()) {
throw configError('Mailbox credentials must carry an endpoint (mint them with `credentials`)', {
endpoint: creds.endpoint,
});
}
return `${creds.endpoint.replace(/\/+$/, '')}/rooms/${encodeRoom(room)}/messages`;
};
/** A string is its own UTF-8 bytes — payloads are data here, never file paths. */
const toBody = (payload) => {
if (typeof payload === 'string')
return new TextEncoder().encode(payload);
if (payload instanceof Uint8Array)
return payload;
if (typeof Blob !== 'undefined' && payload instanceof Blob)
return payload;
if (typeof ReadableStream !== 'undefined' && payload instanceof ReadableStream)
return payload;
throw transportError('Unsupported mailbox payload; expected bytes, a string, a Blob, or a ReadableStream', {
payload: typeof payload,
});
};
/**
* POST /mailbox/v1/rooms/{room}/messages — Deposit PAYLOAD into ROOM for
* peers that arrive later. Identical payloads deduplicate per room within
* the retention window (`deduped: true`), so retries are safe and free.
* Throws an api error whose `code` is the node's, notably
* "message_too_large" and "node_storage_full".
*/
export const deposit = (creds, room, payload, options = {}) => requestUrl('POST', messagesUrl(creds, room), {
token: creds.token,
headers: { 'Content-Type': 'application/octet-stream' },
body: toBody(payload),
fetch: options.fetch,
timeoutMs: options.timeoutMs,
});
/**
* GET /mailbox/v1/rooms/{room}/messages — Replay ROOM, oldest first. Every
* `seq` is a string and must stay one: it packs a millisecond clock into
* its high bits, far past what a JSON number survives. There is no
* server-side ack: the cursor is yours to keep; check it against
* `oldest_seq` with `isHistoryGap` on every replay.
*/
export const replay = (creds, room, options = {}) => requestUrl('GET', messagesUrl(creds, room), {
token: creds.token,
query: { after: options.after, before: options.before, limit: options.limit },
fetch: options.fetch,
timeoutMs: options.timeoutMs,
});
/** Decode one replayed message's base64 payload into bytes. */
export const payloadBytes = (message) => Uint8Array.from(atob(message.payload), (c) => c.charCodeAt(0));
/** Decode one replayed message's payload as UTF-8 text (only meaningful for text deposits). */
export const payloadString = (message) => new TextDecoder().decode(payloadBytes(message));
/**
* 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. An absent/blank cursor (a first visit) is never a gap.
*/
export const isHistoryGap = (cursor, page) => {
if (!cursor?.trim())
return false;
const oldest = page.oldest_seq;
if (oldest === undefined || oldest === null)
return false;
return BigInt(cursor) < BigInt(oldest);
};
/**
* Replay ROOM forward from `after` until `has_more` is exhausted,
* collecting every message. Every page is a charged request, so pass the
* largest `limit` you can use.
*/
export const replayAll = async (creds, room, options = {}) => {
const { after, limit, fetch, timeoutMs } = options;
const messages = [];
let cursor = after;
let gap;
for (;;) {
const page = await replay(creds, room, { after: cursor, limit, fetch, timeoutMs });
gap ??= isHistoryGap(after, page);
messages.push(...page.messages);
cursor = page.messages.at(-1)?.seq ?? cursor;
if (!(page.has_more && page.messages.length > 0)) {
return { messages, cursor: messages.at(-1)?.seq ?? after, gap: gap ?? false };
}
}
};
|