1/**
2 * Defines protocol messages for event-log remote clients and servers.
3 *
4 * This module is the shared boundary between `EventLogRemote` clients and
5 * event-log servers. It defines store ids, protocol errors, the
6 * hello/authenticate session handshake, remote calls for writes and changes,
7 * and message formats for encrypted or plaintext journal entries.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import type { NonEmptyArray, NonEmptyReadonlyArray } from "../Array.ts"
13import type { Brand } from "../Brand.ts"
14import * as SchemaBinary from "../encoding/SchemaBinary.ts"
15import * as Predicate from "../Predicate.ts"
16import * as Rpc from "../rpc/Rpc.ts"
17import * as RpcGroup from "../rpc/RpcGroup.ts"
18import * as RpcMiddleware from "../rpc/RpcMiddleware.ts"
19import * as Schema from "../Schema.ts"
20import * as Transferable from "../workers/Transferable.ts"
21import { Entry, RemoteEntry, RemoteId } from "./EventJournal.ts"
22import type { Identity } from "./EventLog.ts"
23import { EncryptedEntry, EncryptedRemoteEntry } from "./EventLogEncryption.ts"
24
25/**
26 * Type-level identifier used to brand event-log store ids.
27 *
28 * @stability unstable
29 * @category type IDs
30 * @since 4.0.0
31 */
32export type StoreIdTypeId = "effect/eventlog/EventLog/StoreId"
33
34/**
35 * Runtime brand identifier for event-log store ids.
36 *
37 * @stability unstable
38 * @category type IDs
39 * @since 4.0.0
40 */
41export const StoreIdTypeId: StoreIdTypeId = "effect/eventlog/EventLog/StoreId"
42
43/**
44 * Branded string identifying a logical event-log store.
45 *
46 * @stability unstable
47 * @category models
48 * @since 4.0.0
49 */
50export type StoreId = string & Brand<StoreIdTypeId>
51
52/**
53 * Schema for branded event-log store ids.
54 *
55 * @stability unstable
56 * @category schemas
57 * @since 4.0.0
58 */
59export const StoreId = Schema.String.pipe(Schema.brand(StoreIdTypeId))
60
61/**
62 * Error returned by event-log remote RPCs.
63 *
64 * **Details**
65 *
66 * It records the request tag, optional identity and store information, a protocol
67 * error code, and a human-readable message.
68 *
69 * @stability unstable
70 * @category protocols
71 * @since 4.0.0
72 */
73export class EventLogProtocolError extends Schema.TaggedError<EventLogProtocolError>(
74 "effect/eventlog/EventLogRemote/ProtocolError"
75)("EventLogProtocolError", {
76 requestTag: Schema.String,
77 publicKey: Schema.optional(Schema.String),
78 storeId: Schema.optional(StoreId),
79 code: Schema.Literals(["Unauthorized", "Forbidden", "NotFound", "InvalidRequest", "InternalServerError"]),
80 message: Schema.String
81}) {
82 /**
83 * Returns `true` when the value is an `EventLogProtocolError`.
84 *
85 * @stability unstable
86 * @since 4.0.0
87 */
88 static is(u: unknown): u is EventLogProtocolError {
89 return Predicate.isTagged(u, "EventLogProtocolError")
90 }
91}
92
93/**
94 * RPC middleware that authenticates event-log requests and provides the client
95 * `Identity` to authenticated handlers.
96 *
97 * @stability unstable
98 * @category middleware
99 * @since 4.0.0
100 */
101export class EventLogAuthentication extends RpcMiddleware.Service<EventLogAuthentication, {
102 provides: Identity
103}>()("effect/eventlog/EventLogMessage/EventLogAuthentication", {
104 error: EventLogProtocolError
105}) {}
106
107/**
108 * Response sent by the remote server during the authentication handshake.
109 *
110 * **Details**
111 *
112 * It contains the server remote id and a challenge that must be signed by the
113 * client.
114 *
115 * @stability unstable
116 * @category protocols
117 * @since 4.0.0
118 */
119export class HelloResponse extends Schema.Class<HelloResponse>("effect/eventlog/EventLogRemote/HelloResponse")({
120 remoteId: RemoteId,
121 challenge: Transferable.Uint8Array
122}) {}
123
124/**
125 * RPC used to start an event-log remote session and receive a `HelloResponse`.
126 *
127 * @stability unstable
128 * @category protocols
129 * @since 4.0.0
130 */
131export class HelloRpc extends Rpc.make("EventLog.Hello", {
132 success: HelloResponse
133}) {}
134
135/**
136 * Schema for an authentication request containing the client public key,
137 * Ed25519 signing public key, signature over the session challenge payload, and
138 * algorithm name.
139 *
140 * @stability unstable
141 * @category protocols
142 * @since 4.0.0
143 */
144export class Authenticate extends Schema.Class<Authenticate>("effect/eventlog/EventLogRemote/Authenticate")({
145 publicKey: Schema.String,
146 signingPublicKey: Transferable.Uint8Array,
147 signature: Transferable.Uint8Array,
148 algorithm: Schema.Literal("Ed25519")
149}) {}
150
151/**
152 * RPC used to authenticate a remote event-log session after `HelloRpc`.
153 *
154 * @stability unstable
155 * @category protocols
156 * @since 4.0.0
157 */
158export class AuthenticateRpc extends Rpc.make("EventLog.Authenticate", {
159 payload: Authenticate,
160 error: EventLogProtocolError
161}) {}
162
163/**
164 * Represents an entire encoded event-log payload in one transport frame.
165 *
166 * @stability unstable
167 * @category protocols
168 * @since 4.0.0
169 */
170export class SingleMessage
171 extends Schema.TaggedClass<SingleMessage>("effect/eventlog/EventLogRemote/SingleMessage")("Single", {
172 data: Transferable.Uint8Array
173 })
174{}
175
176/**
177 * Represents one part of a large encoded event-log payload.
178 *
179 * **When to use**
180 *
181 * Use to divide data into chunks and `join` to reassemble all chunks with
182 * the same id once every part has arrived.
183 *
184 * @stability unstable
185 * @category protocols
186 * @since 4.0.0
187 */
188export class ChunkedMessage
189 extends Schema.TaggedClass<ChunkedMessage>("effect/eventlog/EventLogRemote/ChunkedMessage")("Chunked", {
190 id: Schema.Int,
191 part: Schema.Tuple([Schema.Natural, Schema.Natural]).check(
192 Schema.makeFilter(([index, total]) => index < total, {
193 expected: "a chunk part with an index less than its total"
194 })
195 ),
196 data: Transferable.Uint8Array
197 })
198{
199 static chunkSize = 512_000
200
201 static initialJoinState() {
202 return new Map<number, {
203 readonly parts: Array<Uint8Array>
204 count: number
205 bytes: number
206 }>()
207 }
208
209 /**
210 * Splits binary event-log message data into numbered chunks.
211 *
212 * @stability unstable
213 * @since 4.0.0
214 */
215 static split(id: number, data: Uint8Array): NonEmptyReadonlyArray<ChunkedMessage> {
216 const parts = Math.max(1, Math.ceil(data.byteLength / ChunkedMessage.chunkSize))
217 const result: NonEmptyArray<ChunkedMessage> = new Array(parts) as any
218 for (let i = 0; i < parts; i++) {
219 const start = i * ChunkedMessage.chunkSize
220 const end = Math.min((i + 1) * ChunkedMessage.chunkSize, data.byteLength)
221 result[i] = new ChunkedMessage({
222 id,
223 part: [i, parts],
224 data: data.subarray(start, end) as any
225 })
226 }
227 return result
228 }
229
230 /**
231 * Reassembles all chunks for a message id into the original binary payload.
232 *
233 * @stability unstable
234 * @since 4.0.0
235 */
236 static join(
237 map: Map<number, {
238 readonly parts: Array<Uint8Array>
239 count: number
240 bytes: number
241 }>,
242 part: ChunkedMessage
243 ): Uint8Array<ArrayBuffer> | undefined {
244 const [index, total] = part.part
245 let entry = map.get(part.id)
246 if (!entry) {
247 entry = {
248 parts: new Array(total),
249 count: 0,
250 bytes: 0
251 }
252 map.set(part.id, entry)
253 }
254 if (entry.parts[index] !== undefined) {
255 return
256 }
257 entry.parts[index] = part.data
258 entry.count++
259 entry.bytes += part.data.byteLength
260 if (entry.count !== total) {
261 return
262 }
263 const data = new Uint8Array(entry.bytes)
264 let offset = 0
265 for (const part of entry.parts) {
266 data.set(part, offset)
267 offset += part.byteLength
268 }
269 map.delete(part.id)
270 return data
271 }
272}
273
274/**
275 * RPC used to send one chunk of a large encoded write payload.
276 *
277 * @stability unstable
278 * @category protocols
279 * @since 4.0.0
280 */
281export class WriteChunkedRpc extends Rpc.make("EventLog.WriteChunked", {
282 payload: ChunkedMessage,
283 error: EventLogProtocolError
284}).middleware(EventLogAuthentication) {}
285
286/**
287 * Schema for encrypted event-log write payloads sent to a remote store.
288 *
289 * **Details**
290 *
291 * It includes the client public key, target store id, and encrypted entries
292 * with their AES-GCM initialization vectors.
293 *
294 * @stability unstable
295 * @category protocols
296 * @since 4.0.0
297 */
298export class WriteEntries extends Schema.Class<WriteEntries>("effect/eventlog/EventLogRemote/WriteEntries")({
299 publicKey: Schema.String,
300 storeId: StoreId,
301 encryptedEntries: Schema.Array(EncryptedEntry)
302}) {
303 static FromSchemaBinary = SchemaBinary.toCodec(WriteEntries)
304 static encode = Schema.encodeEffect(this.FromSchemaBinary)
305 static decode = Schema.decodeEffect(this.FromSchemaBinary)
306 get encoded() {
307 return WriteEntries.encode(this)
308 }
309}
310
311/**
312 * Schema for plaintext event-log write payloads sent to a remote store.
313 *
314 * @stability unstable
315 * @category protocols
316 * @since 4.0.0
317 */
318export class WriteEntriesUnencrypted
319 extends Schema.Class<WriteEntriesUnencrypted>("effect/eventlog/EventLogRemote/WriteEntriesUnencrypted")({
320 publicKey: Schema.String,
321 storeId: StoreId,
322 entries: Schema.Array(Entry)
323 })
324{
325 static FromSchemaBinary = SchemaBinary.toCodec(WriteEntriesUnencrypted)
326 static encode = Schema.encodeEffect(this.FromSchemaBinary)
327 static decode = Schema.decodeEffect(this.FromSchemaBinary)
328 get encoded() {
329 return WriteEntriesUnencrypted.encode(this)
330 }
331}
332
333/**
334 * RPC used to send an encoded write payload that fits in one message.
335 *
336 * @stability unstable
337 * @category protocols
338 * @since 4.0.0
339 */
340export class WriteSingleRpc extends Rpc.make("EventLog.WriteSingle", {
341 payload: {
342 data: Transferable.Uint8Array
343 },
344 error: EventLogProtocolError
345}).middleware(EventLogAuthentication) {}
346
347/**
348 * RPC used to stream remote event-log changes for a public key and store id
349 * starting at a sequence number.
350 *
351 * **Details**
352 *
353 * Responses are encoded as either `SingleMessage` values or `ChunkedMessage`
354 * parts.
355 *
356 * @stability unstable
357 * @category protocols
358 * @since 4.0.0
359 */
360export class ChangesRpc extends Rpc.make("EventLog.Changes", {
361 payload: {
362 publicKey: Schema.String,
363 storeId: StoreId,
364 startSequence: Schema.Natural
365 },
366 success: Schema.Union([SingleMessage, ChunkedMessage]),
367 error: EventLogProtocolError,
368 stream: true
369}).middleware(EventLogAuthentication) {
370 static EncryptedFromSchemaBinary = SchemaBinary.toCodec(Schema.NonEmptyArray(EncryptedRemoteEntry))
371 static UnencryptedFromSchemaBinary = SchemaBinary.toCodec(Schema.NonEmptyArray(RemoteEntry))
372 static encodeEncrypted = Schema.encodeEffect(ChangesRpc.EncryptedFromSchemaBinary)
373 static decodeEncrypted = Schema.decodeEffect(ChangesRpc.EncryptedFromSchemaBinary)
374 static encodeUnencrypted = Schema.encodeEffect(ChangesRpc.UnencryptedFromSchemaBinary)
375 static decodeUnencrypted = Schema.decodeEffect(ChangesRpc.UnencryptedFromSchemaBinary)
376}
377
378/**
379 * RPC group containing the event-log remote handshake, authentication, write, and
380 * changes endpoints.
381 *
382 * @stability unstable
383 * @category protocols
384 * @since 4.0.0
385 */
386export class EventLogRemoteRpcs extends RpcGroup.make(
387 HelloRpc,
388 AuthenticateRpc,
389 WriteChunkedRpc,
390 WriteSingleRpc,
391 ChangesRpc
392) {}