1/**
2 * Message envelopes shared by unstable RPC clients, servers, serializers, and
3 * transports.
4 *
5 * `RpcMessage` is the protocol vocabulary below `RpcClient` and `RpcServer`.
6 * It defines decoded messages for in-process channels and encoded messages for
7 * transport boundaries, so custom protocols can move the same request,
8 * streaming, acknowledgement, interrupt, keepalive, and defect signals as the
9 * built-in HTTP, socket, worker, and test transports.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import type { NonEmptyReadonlyArray } from "../Array.ts"
15import type { Branded } from "../Brand.ts"
16import type { Headers } from "../http/Headers.ts"
17import * as Schema from "../Schema.ts"
18import type * as Rpc from "./Rpc.ts"
19import type { RpcClientError } from "./RpcClientError.ts"
20
21/**
22 * Decoded messages that can be sent from an RPC client to a server.
23 *
24 * @stability unstable
25 * @category models
26 * @since 4.0.0
27 */
28export type FromClient<A extends Rpc.Any> = Request<A> | Ack | Interrupt | Eof
29
30/**
31 * Transport-encoded messages that can be sent from an RPC client to a server.
32 *
33 * @stability unstable
34 * @category models
35 * @since 4.0.0
36 */
37export type FromClientEncoded = RequestEncoded | AckEncoded | InterruptEncoded | Ping | Eof
38
39/**
40 * A branded request identifier used to correlate RPC requests, responses,
41 * chunks, acknowledgements, and interrupts.
42 *
43 * @stability unstable
44 * @category models
45 * @since 4.0.0
46 */
47export type RequestId = Branded<string | number, "~effect/rpc/RpcMessage/RequestId">
48
49/**
50 * Converts a bigint or string request id into the branded `RequestId` type.
51 *
52 * @stability unstable
53 * @category constructors
54 * @since 4.0.0
55 */
56export const RequestId = (id: string | number): RequestId => id as RequestId
57
58/**
59 * The transport-encoded RPC request envelope, including the string request id,
60 * RPC tag, encoded payload, headers, and optional trace context.
61 *
62 * **Details**
63 *
64 * Requests flow in both directions: servers use them for server-originated
65 * requests and, with `isNotification` set, for server notifications.
66 *
67 * @stability unstable
68 * @category models
69 * @since 4.0.0
70 */
71export interface RequestEncoded {
72 readonly _tag: "Request"
73 readonly id: string | number
74 readonly tag: string
75 readonly payload: unknown
76 readonly headers: ReadonlyArray<[string, string]>
77 readonly isNotification?: true
78 readonly traceId?: string
79 readonly spanId?: string
80 readonly sampled?: boolean
81}
82
83/**
84 * The decoded RPC request envelope for an RPC union, carrying a branded request
85 * id, typed RPC tag, decoded payload, headers, and optional trace context.
86 *
87 * @stability unstable
88 * @category models
89 * @since 4.0.0
90 */
91export interface Request<A extends Rpc.Any> {
92 readonly _tag: "Request"
93 readonly id: RequestId
94 readonly tag: Rpc.Tag<A>
95 readonly payload: Rpc.Payload<A>
96 readonly headers: Headers
97 readonly traceId?: string
98 readonly spanId?: string
99 readonly sampled?: boolean
100}
101
102/**
103 * A decoded acknowledgement for a streamed RPC response chunk.
104 *
105 * @stability unstable
106 * @category models
107 * @since 4.0.0
108 */
109export interface Ack {
110 readonly _tag: "Ack"
111 readonly requestId: RequestId
112}
113
114/**
115 * A decoded request to interrupt an in-flight RPC, carrying the request id and
116 * interrupting fiber ids.
117 *
118 * @stability unstable
119 * @category models
120 * @since 4.0.0
121 */
122export interface Interrupt {
123 readonly _tag: "Interrupt"
124 readonly requestId: RequestId
125 readonly interruptors: ReadonlyArray<number>
126}
127
128/**
129 * The transport-encoded acknowledgement for a streamed RPC response chunk.
130 *
131 * @stability unstable
132 * @category models
133 * @since 4.0.0
134 */
135export interface AckEncoded {
136 readonly _tag: "Ack"
137 readonly requestId: string | number
138}
139
140/**
141 * The transport-encoded request to interrupt an in-flight RPC.
142 *
143 * @stability unstable
144 * @category models
145 * @since 4.0.0
146 */
147export interface InterruptEncoded {
148 readonly _tag: "Interrupt"
149 readonly requestId: string | number
150}
151
152/**
153 * A client-to-server message indicating that the client has finished sending
154 * input for the current connection or request batch.
155 *
156 * @stability unstable
157 * @category models
158 * @since 4.0.0
159 */
160export interface Eof {
161 readonly _tag: "Eof"
162}
163
164/**
165 * A client-to-server keepalive message used by protocols that monitor
166 * connection liveness.
167 *
168 * @stability unstable
169 * @category models
170 * @since 4.0.0
171 */
172export interface Ping {
173 readonly _tag: "Ping"
174}
175
176/**
177 * Represents the reusable `Eof` message value.
178 *
179 * @stability unstable
180 * @category constants
181 * @since 4.0.0
182 */
183export const constEof: Eof = { _tag: "Eof" }
184
185/**
186 * Represents the reusable `Ping` message value.
187 *
188 * @stability unstable
189 * @category constants
190 * @since 4.0.0
191 */
192export const constPing: Ping = { _tag: "Ping" }
193
194/**
195 * Decoded messages that can be sent from an RPC server to a client.
196 *
197 * @stability unstable
198 * @category models
199 * @since 4.0.0
200 */
201export type FromServer<A extends Rpc.Any> =
202 | ResponseChunk<A>
203 | ResponseExit<A>
204 | ResponseDefect
205 | ClientEnd
206
207/**
208 * Transport-encoded messages that can be sent from an RPC server to a client.
209 *
210 * @stability unstable
211 * @category models
212 * @since 4.0.0
213 */
214export type FromServerEncoded =
215 | ResponseChunkEncoded
216 | ResponseExitEncoded
217 | ResponseDefectEncoded
218 | Pong
219 | ClientProtocolError
220 | RequestEncoded
221
222/**
223 * The brand identifier used by the `ResponseId` type.
224 *
225 * @stability unstable
226 * @category type IDs
227 * @since 4.0.0
228 */
229export const ResponseIdTypeId = "~effect/rpc/RpcMessage/ResponseId"
230
231/**
232 * The literal type of the `ResponseId` brand identifier.
233 *
234 * @stability unstable
235 * @category type IDs
236 * @since 4.0.0
237 */
238export type ResponseIdTypeId = typeof ResponseIdTypeId
239
240/**
241 * A branded numeric identifier for server responses.
242 *
243 * @stability unstable
244 * @category models
245 * @since 4.0.0
246 */
247export type ResponseId = Branded<number, ResponseIdTypeId>
248
249/**
250 * The transport-encoded response message containing a non-empty batch of stream
251 * chunk values for a request.
252 *
253 * @stability unstable
254 * @category models
255 * @since 4.0.0
256 */
257export interface ResponseChunkEncoded {
258 readonly _tag: "Chunk"
259 readonly requestId: string | number
260 readonly values: NonEmptyReadonlyArray<unknown>
261}
262
263/**
264 * The decoded response message containing a non-empty batch of stream chunk
265 * values for a specific client and request.
266 *
267 * @stability unstable
268 * @category models
269 * @since 4.0.0
270 */
271export interface ResponseChunk<A extends Rpc.Any> {
272 readonly _tag: "Chunk"
273 readonly clientId: number
274 readonly requestId: RequestId
275 readonly values: NonEmptyReadonlyArray<Rpc.SuccessChunk<A>>
276}
277
278/**
279 * The transport representation of an RPC `Exit`, encoding success values or a
280 * failure cause made of failures, defects, and interrupts.
281 *
282 * @stability unstable
283 * @category models
284 * @since 4.0.0
285 */
286export type ExitEncoded<A, E> = {
287 readonly _tag: "Success"
288 readonly value: A
289} | {
290 readonly _tag: "Failure"
291 readonly cause: ReadonlyArray<
292 {
293 readonly _tag: "Fail"
294 readonly error: E
295 } | {
296 readonly _tag: "Die"
297 readonly defect: unknown
298 } | {
299 readonly _tag: "Interrupt"
300 readonly fiberId: number | null | undefined
301 }
302 >
303}
304
305/**
306 * The transport-encoded terminal response for a request, carrying the encoded
307 * `Exit`.
308 *
309 * @stability unstable
310 * @category models
311 * @since 4.0.0
312 */
313export interface ResponseExitEncoded {
314 readonly _tag: "Exit"
315 readonly requestId: string | number
316 readonly exit: ExitEncoded<unknown, unknown>
317}
318
319/**
320 * A server-to-client protocol message reporting a client protocol error to all
321 * affected in-flight requests.
322 *
323 * @stability unstable
324 * @category models
325 * @since 4.0.0
326 */
327export interface ClientProtocolError {
328 readonly _tag: "ClientProtocolError"
329 readonly error: RpcClientError
330}
331
332/**
333 * The decoded terminal response for a request, carrying the typed `Rpc.Exit`
334 * for the RPC.
335 *
336 * @stability unstable
337 * @category models
338 * @since 4.0.0
339 */
340export interface ResponseExit<A extends Rpc.Any> {
341 readonly _tag: "Exit"
342 readonly clientId: number
343 readonly requestId: RequestId
344 readonly exit: Rpc.Exit<A>
345}
346
347/**
348 * The transport-encoded server defect message used for protocol-level defects
349 * that affect the client connection.
350 *
351 * @stability unstable
352 * @category models
353 * @since 4.0.0
354 */
355export interface ResponseDefectEncoded {
356 readonly _tag: "Defect"
357 readonly defect: unknown
358}
359
360/**
361 * Creates an encoded terminal response for a request whose exit is a defect.
362 *
363 * **Details**
364 *
365 * The defect must already be encoded, because the codec that fills the defect
366 * hole belongs to the protocol. Encode it with
367 * `protocol.codecFor(Schema.Defect())` first.
368 *
369 * This constructor produces the structured exit used by JSON-compatible
370 * protocols. Serializations whose `codecFor` returns bytes must encode the
371 * complete RPC exit before placing it in the response envelope instead.
372 *
373 * @stability unstable
374 * @category constructors
375 * @since 4.0.0
376 */
377export const ResponseExitDieEncoded = (options: {
378 readonly requestId: RequestId
379 readonly encodedDefect: unknown
380}): ResponseExitEncoded => ({
381 _tag: "Exit",
382 requestId: options.requestId,
383 exit: {
384 _tag: "Failure",
385 cause: [{
386 _tag: "Die",
387 defect: options.encodedDefect
388 }]
389 }
390})
391
392/**
393 * Creates a transport-encoded defect response around an already-encoded defect.
394 *
395 * **Details**
396 *
397 * Encode the defect with `protocol.codecFor(Schema.Defect())` before wrapping
398 * it, because the codec that fills the defect hole belongs to the protocol.
399 *
400 * @stability unstable
401 * @category constructors
402 * @since 4.0.0
403 */
404export const ResponseDefectEncoded = (encodedDefect: unknown): ResponseDefectEncoded => ({
405 _tag: "Defect",
406 defect: encodedDefect
407})
408
409/**
410 * The decoded server defect message for a client connection.
411 *
412 * @stability unstable
413 * @category models
414 * @since 4.0.0
415 */
416export interface ResponseDefect {
417 readonly _tag: "Defect"
418 readonly clientId: number
419 readonly defect: unknown
420}
421
422/**
423 * A server message indicating that the client connection has ended.
424 *
425 * @stability unstable
426 * @category models
427 * @since 4.0.0
428 */
429export interface ClientEnd {
430 readonly _tag: "ClientEnd"
431 readonly clientId: number
432}
433
434/**
435 * A server-to-client keepalive response to a `Ping` message.
436 *
437 * @stability unstable
438 * @category models
439 * @since 4.0.0
440 */
441export interface Pong {
442 readonly _tag: "Pong"
443}
444
445const RequestIdSchema = Schema.Union([Schema.String, Schema.Number])
446
447// Schema for transport-encoded RPC requests whose payload hole has already
448// been filled by the active serialization.
449const RequestEncodedSchema = Schema.Struct({
450 _tag: Schema.tag("Request"),
451 id: RequestIdSchema,
452 tag: Schema.String,
453 payload: Schema.Uint8Array,
454 headers: Schema.Array(Schema.Tuple([Schema.String, Schema.String])),
455 isNotification: Schema.optional(Schema.Literal(true)),
456 traceId: Schema.optional(Schema.String),
457 spanId: Schema.optional(Schema.String),
458 sampled: Schema.optional(Schema.Boolean)
459})
460
461/**
462 * Schema for every transport-encoded RPC envelope that crosses the wire.
463 * Binary serializers use it only after each schema-dependent hole has been
464 * encoded as bytes. `ClientProtocolError` is excluded because clients create
465 * it locally rather than receiving it from a server.
466 *
467 * @stability unstable
468 * @category schemas
469 * @since 4.0.0
470 */
471export const EncodedSchema = Schema.Union([
472 RequestEncodedSchema,
473 Schema.Struct({
474 _tag: Schema.tag("Ack"),
475 requestId: RequestIdSchema
476 }),
477 Schema.Struct({
478 _tag: Schema.tag("Interrupt"),
479 requestId: RequestIdSchema
480 }),
481 Schema.Struct({ _tag: Schema.tag("Ping") }),
482 Schema.Struct({ _tag: Schema.tag("Eof") }),
483 Schema.Struct({
484 _tag: Schema.tag("Chunk"),
485 requestId: RequestIdSchema,
486 values: Schema.Uint8Array
487 }),
488 Schema.Struct({
489 _tag: Schema.tag("Exit"),
490 requestId: RequestIdSchema,
491 exit: Schema.Uint8Array
492 }),
493 Schema.Struct({
494 _tag: Schema.tag("Defect"),
495 defect: Schema.Uint8Array
496 }),
497 Schema.Struct({ _tag: Schema.tag("Pong") })
498])
499
500/**
501 * Represents the reusable `Pong` message value.
502 *
503 * @stability unstable
504 * @category constants
505 * @since 4.0.0
506 */
507export const constPong: Pong = { _tag: "Pong" }
508
509/**
510 * Checks if the response type is terminal.
511 *
512 * @stability unstable
513 * @category guards
514 * @since 4.0.0
515 */
516export const isTerminalResponse = (response: FromServerEncoded): boolean => {
517 switch (response._tag) {
518 case "Exit":
519 case "Defect":
520 case "ClientProtocolError": {
521 return true
522 }
523 default: {
524 return false
525 }
526 }
527}