1/**
2 * Serializes RPC protocol messages for transports.
3 *
4 * `RpcSerialization` is the boundary between `RpcMessage` envelopes and the
5 * bytes or strings carried by a transport. This module provides built-in
6 * serializers for JSON, newline-delimited JSON, JSON-RPC 2.0, and SchemaBinary,
7 * including framed formats that can decode multiple messages
8 * from streaming chunks.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Context from "../Context.ts"
14import * as Data from "../Data.ts"
15import * as SchemaBinary from "../encoding/SchemaBinary.ts"
16import * as Layer from "../Layer.ts"
17import * as Predicate from "../Predicate.ts"
18import { hasProperty } from "../Predicate.ts"
19import * as Schema from "../Schema.ts"
20import * as RpcMessage from "./RpcMessage.ts"
21
22/**
23 * Builds the codec used to fill the `unknown` holes of RPC protocol messages,
24 * such as request payloads, stream chunks, exits, and defects.
25 *
26 * **Details**
27 *
28 * The envelope around the hole is framed by {@link Parser}, which never sees the
29 * schema. JSON based serializations return `Schema.toCodecJson`, so the encoded
30 * hole is JSON. Other serializations may encode the hole as bytes; both fit the
31 * `unknown` hole, so `RpcSerialization` has no type parameter.
32 *
33 * @stability unstable
34 * @category serialization
35 * @since 4.0.0
36 */
37export type CodecFor = <S extends Schema.Top>(
38 schema: S
39) => Schema.Codec<S["Type"], unknown, S["DecodingServices"], S["EncodingServices"]>
40
41const codecForJson = Schema.toCodecJson as CodecFor
42
43// shared by every parser that decodes whole frames, so creating a parser does
44// not pay for a native TextDecoder
45let sharedTextDecoder: TextDecoder | undefined
46const decodeText = (bytes: Uint8Array): string => (sharedTextDecoder ??= new TextDecoder()).decode(bytes)
47
48/**
49 * Service that describes how RPC protocol messages are encoded and decoded,
50 * including the content type and whether the serialization format provides
51 * message framing.
52 *
53 * **When to use**
54 *
55 * Use to provide the serialization boundary shared by RPC clients and servers
56 * for a chosen wire format.
57 *
58 * @stability unstable
59 * @category services
60 * @since 4.0.0
61 */
62export class RpcSerialization extends Context.Service<RpcSerialization, {
63 makeUnsafe(): Parser
64 readonly contentType: string
65 readonly includesFraming: boolean
66 readonly codecFor: CodecFor
67}>()("effect/rpc/RpcSerialization") {}
68
69/**
70 * A stateful parser for an RPC serialization format, able to decode input
71 * chunks into protocol messages and encode messages for transport.
72 *
73 * @stability unstable
74 * @category serialization
75 * @since 4.0.0
76 */
77export interface Parser {
78 readonly decode: (data: Uint8Array | string) => ReadonlyArray<unknown>
79 readonly encode: (response: unknown) => Uint8Array | string | undefined
80}
81
82/**
83 * Error raised when a streaming parser retains more data than its configured
84 * buffer limit.
85 *
86 * @stability unstable
87 * @category errors
88 * @since 4.0.0
89 */
90export class MaxBufferSizeExceeded extends Data.TaggedError("MaxBufferSizeExceeded")<{
91 readonly maxBufferSize: number
92}> {
93 override get message() {
94 return `RPC serialization buffer exceeded the maximum size of ${this.maxBufferSize}`
95 }
96}
97
98/**
99 * Options shared by streaming RPC serialization formats.
100 *
101 * @stability unstable
102 * @category serialization
103 * @since 4.0.0
104 */
105export interface StreamOptions {
106 /**
107 * Maximum number of bytes or string code units retained for an incomplete frame.
108 * The default is 16 MiB. Use `"unbounded"` to disable the limit.
109 */
110 readonly maxBufferSize?: number | "unbounded" | undefined
111}
112
113const defaultMaxBufferSize = 16 * 1024 * 1024
114
115const isBufferSizeExceeded = (
116 bufferSize: number,
117 maxBufferSize: number | "unbounded"
118): maxBufferSize is number => maxBufferSize !== "unbounded" && bufferSize > maxBufferSize
119
120/**
121 * JSON RPC serialization for whole message payloads. It does not include
122 * message framing, so it is intended for transports that frame responses
123 * themselves.
124 *
125 * @stability unstable
126 * @category serialization
127 * @since 4.0.0
128 */
129export const json: RpcSerialization["Service"] = RpcSerialization.of({
130 contentType: "application/json",
131 includesFraming: false,
132 codecFor: codecForJson,
133 makeUnsafe: () => ({
134 decode: (bytes) => {
135 const decoded = JSON.parse(typeof bytes === "string" ? bytes : decodeText(bytes))
136 return Array.isArray(decoded) ? decoded : [decoded]
137 },
138 encode: (response) => JSON.stringify(response)
139 })
140})
141
142/**
143 * Serializes RPC protocol messages as newline-delimited JSON, framing each message
144 * with a trailing newline.
145 *
146 * @stability unstable
147 * @category serialization
148 * @since 4.0.0
149 */
150export const makeNdjson = (options?: StreamOptions): RpcSerialization["Service"] => {
151 const maxBufferSize = options?.maxBufferSize ?? defaultMaxBufferSize
152 return RpcSerialization.of({
153 contentType: "application/ndjson",
154 includesFraming: true,
155 codecFor: codecForJson,
156 makeUnsafe: () => {
157 // lazily created: string transports never need byte decoding
158 let decoder: TextDecoder | undefined
159 let buffer = ""
160 const failMaxBufferSize = (maxBufferSize: number): never => {
161 buffer = ""
162 throw new MaxBufferSizeExceeded({ maxBufferSize })
163 }
164 return ({
165 decode: (bytes) => {
166 buffer += typeof bytes === "string" ? bytes : (decoder ??= new TextDecoder()).decode(bytes, { stream: true })
167 let position = 0
168 let nlIndex = buffer.indexOf("\n", position)
169 const items: Array<unknown> = []
170 while (nlIndex !== -1) {
171 if (isBufferSizeExceeded(nlIndex - position, maxBufferSize)) {
172 failMaxBufferSize(maxBufferSize)
173 }
174 const line = buffer.slice(position, nlIndex)
175 position = nlIndex + 1
176 try {
177 items.push(JSON.parse(line))
178 } catch {
179 }
180 nlIndex = buffer.indexOf("\n", position)
181 }
182 buffer = buffer.slice(position)
183 if (isBufferSizeExceeded(buffer.length, maxBufferSize)) {
184 failMaxBufferSize(maxBufferSize)
185 }
186 return items
187 },
188 encode: (response) => {
189 if (Array.isArray(response)) {
190 if (response.length === 0) return undefined
191 let data = ""
192 for (let i = 0; i < response.length; i++) {
193 data += JSON.stringify(response[i]) + "\n"
194 }
195 return data
196 }
197 return JSON.stringify(response) + "\n"
198 }
199 })
200 }
201 })
202}
203
204/**
205 * Default newline-delimited JSON RPC serialization.
206 *
207 * @stability unstable
208 * @category serialization
209 * @since 4.0.0
210 */
211export const ndjson: RpcSerialization["Service"] = makeNdjson()
212
213/**
214 * Creates a JSON-RPC 2.0 serialization for RPC protocol messages without
215 * additional message framing.
216 *
217 * @stability unstable
218 * @category serialization
219 * @since 4.0.0
220 */
221export const jsonRpc = (options?: {
222 readonly contentType?: string | undefined
223}): RpcSerialization["Service"] =>
224 RpcSerialization.of({
225 contentType: options?.contentType ?? "application/json",
226 includesFraming: false,
227 codecFor: codecForJson,
228 makeUnsafe: () => {
229 const batches = new Map<string | number, {
230 readonly size: number
231 readonly responses: Map<string | number, RpcMessage.FromServerEncoded>
232 }>()
233 return {
234 decode: (bytes) => {
235 const decoded: unknown = JSON.parse(
236 typeof bytes === "string" ? bytes : decodeText(bytes)
237 )
238 return decodeJsonRpcRaw(decoded, batches)
239 },
240 encode: (response) => {
241 const encoded = encodeJsonRpcResponse(response as any, batches)
242 return encoded && JSON.stringify(encoded)
243 }
244 }
245 }
246 })
247
248/**
249 * Creates a newline-delimited JSON-RPC 2.0 serialization for RPC protocol
250 * messages.
251 *
252 * @stability unstable
253 * @category serialization
254 * @since 4.0.0
255 */
256export const ndJsonRpc = (options?: {
257 readonly contentType?: string | undefined
258 readonly maxBufferSize?: number | "unbounded" | undefined
259}): RpcSerialization["Service"] =>
260 RpcSerialization.of({
261 contentType: options?.contentType ?? "application/json-rpc",
262 includesFraming: true,
263 codecFor: codecForJson,
264 makeUnsafe: () => {
265 const parser = makeNdjson({ maxBufferSize: options?.maxBufferSize }).makeUnsafe()
266 const batches = new Map<string, {
267 readonly size: number
268 readonly responses: Map<string, RpcMessage.FromServerEncoded>
269 }>()
270 return ({
271 decode: (bytes) => {
272 const frames = parser.decode(bytes)
273 if (frames.length === 0) return []
274 const messages: Array<RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded> = []
275 for (let i = 0; i < frames.length; i++) {
276 messages.push(...decodeJsonRpcRaw(frames[i], batches))
277 }
278 return messages
279 },
280 encode: (response) => {
281 const encoded = encodeJsonRpcResponse(response as any, batches)
282 return encoded && parser.encode(encoded)
283 }
284 })
285 }
286 })
287
288function decodeJsonRpcRaw(
289 decoded: unknown,
290 batches: Map<string | number, {
291 readonly size: number
292 readonly responses: Map<string | number, RpcMessage.FromServerEncoded>
293 }>
294) {
295 if (Array.isArray(decoded)) {
296 const batch = {
297 size: 0,
298 responses: new Map<string, RpcMessage.FromServerEncoded>()
299 }
300 const messages: Array<RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded> = []
301 for (let i = 0; i < decoded.length; i++) {
302 const item = decoded[i]
303 if (!Predicate.isObject(item)) continue
304 const message = decodeJsonRpcMessage(item as unknown as JsonRpcMessage)
305 messages.push(message)
306 if (message._tag === "Request" && !message.isNotification) {
307 batch.size++
308 batches.set(message.id, batch)
309 }
310 }
311 return messages
312 }
313 return Predicate.isObject(decoded) ? [decodeJsonRpcMessage(decoded as unknown as JsonRpcMessage)] : []
314}
315
316function decodeJsonRpcMessage(decoded: JsonRpcMessage): RpcMessage.FromClientEncoded | RpcMessage.FromServerEncoded {
317 if (Object.hasOwn(decoded, "method")) {
318 const request = decoded as JsonRpcRequest
319 if (
320 Predicate.isNullish(request.id) && Predicate.isString(request.method) && request.method.startsWith("@effect/rpc/")
321 ) {
322 const tag = request.method.slice("@effect/rpc/".length) as
323 | RpcMessage.FromServerEncoded["_tag"]
324 | Exclude<RpcMessage.FromClientEncoded["_tag"], "Request">
325 const requestId = (request as any).params?.requestId
326 return requestId !== undefined ?
327 {
328 _tag: tag,
329 requestId
330 } as any :
331 { _tag: tag } as any
332 }
333 return {
334 _tag: "Request",
335 id: request.id ?? "",
336 tag: request.method,
337 payload: request.params ?? null,
338 headers: request.headers ?? [],
339 ...(Predicate.hasProperty(request, "id") ? {} : { isNotification: true as const }),
340 ...(request.spanId ?
341 {
342 traceId: request.traceId,
343 spanId: request.spanId!,
344 sampled: request.sampled!
345 } :
346 {})
347 }
348 }
349 const response = decoded as JsonRpcResponse
350 const hasError = Object.hasOwn(response, "error")
351 if (hasError && response.error && response.error._tag === "Defect") {
352 return {
353 _tag: "Defect",
354 defect: response.error.data
355 }
356 } else if (Object.hasOwn(response, "chunk") && response.chunk === true) {
357 return {
358 _tag: "Chunk",
359 requestId: response.id ?? "",
360 values: response.result as any
361 }
362 }
363 return {
364 _tag: "Exit",
365 requestId: response.id ?? "",
366 exit: hasError && response.error != null ?
367 {
368 _tag: "Failure",
369 cause: response.error._tag === "Cause" ?
370 response.error.data as any :
371 [{
372 _tag: "Fail",
373 error: response.error
374 }]
375 } :
376 {
377 _tag: "Success",
378 value: response.result
379 }
380 }
381}
382
383function encodeJsonRpcRaw(
384 response: RpcMessage.FromServerEncoded | RpcMessage.FromClientEncoded,
385 batches: Map<string | number, {
386 readonly size: number
387 readonly responses: Map<string | number, RpcMessage.FromServerEncoded>
388 }>
389) {
390 if (!("requestId" in response)) {
391 return encodeJsonRpcMessage(response)
392 }
393 const batch = batches.get(response.requestId)
394 if (batch) {
395 batches.delete(response.requestId)
396 batch.responses.set(response.requestId, response as any)
397 if (batch.size === batch.responses.size) {
398 return Array.from(batch.responses.values(), encodeJsonRpcMessage)
399 }
400 return undefined
401 }
402 return encodeJsonRpcMessage(response)
403}
404
405function encodeJsonRpcResponse(
406 response:
407 | RpcMessage.FromServerEncoded
408 | RpcMessage.FromClientEncoded
409 | Array<RpcMessage.FromServerEncoded | RpcMessage.FromClientEncoded>,
410 batches: Map<string | number, {
411 readonly size: number
412 readonly responses: Map<string | number, RpcMessage.FromServerEncoded>
413 }>
414) {
415 if (Array.isArray(response) === false) {
416 return encodeJsonRpcRaw(response, batches)
417 }
418 if (response.length === 0) {
419 return undefined
420 }
421 const encoded: Array<JsonRpcMessage | Array<JsonRpcMessage>> = []
422 for (let i = 0; i < response.length; i++) {
423 const current = encodeJsonRpcRaw(response[i], batches)
424 if (current !== undefined) {
425 encoded.push(current)
426 }
427 }
428 if (encoded.length === 0) {
429 return undefined
430 }
431 if (encoded.length === 1) {
432 return encoded[0]
433 }
434 const messages: Array<JsonRpcMessage> = []
435 for (let i = 0; i < encoded.length; i++) {
436 const current = encoded[i]
437 if (Array.isArray(current)) {
438 messages.push(...current)
439 } else {
440 messages.push(current)
441 }
442 }
443 return messages
444}
445
446function encodeJsonRpcMessage(response: RpcMessage.FromServerEncoded | RpcMessage.FromClientEncoded): JsonRpcMessage {
447 switch (response._tag) {
448 case "Request":
449 return {
450 jsonrpc: "2.0",
451 method: response.tag,
452 params: response.payload,
453 // a JSON-RPC notification is a request without an id
454 ...(response.isNotification ? {} : { id: response.id }),
455 ...(response.headers?.length > 0 ? { headers: response.headers } : {}),
456 traceId: response.traceId,
457 spanId: response.spanId,
458 sampled: response.sampled
459 }
460 case "Ping":
461 case "Pong":
462 case "Interrupt":
463 case "Ack":
464 case "Eof":
465 return {
466 jsonrpc: "2.0",
467 method: `@effect/rpc/${response._tag}`,
468 params: "requestId" in response ? { requestId: response.requestId } : undefined
469 }
470 case "Chunk":
471 return {
472 jsonrpc: "2.0",
473 chunk: true,
474 id: response.requestId,
475 result: response.values
476 }
477 case "Exit": {
478 if (response.exit._tag === "Success") {
479 return {
480 jsonrpc: "2.0",
481 id: response.requestId ?? undefined,
482 result: response.exit.value
483 } as any
484 }
485 const failure = response.exit.cause.find((failure) => failure._tag === "Fail")
486 const error = failure?._tag === "Fail" ? failure.error : undefined
487 return {
488 jsonrpc: "2.0",
489 id: response.requestId ?? undefined,
490 error: response.exit._tag === "Failure" ?
491 {
492 _tag: "Cause",
493 code: error && Predicate.hasProperty(error, "code") ? Number(error.code) : 0,
494 message: error && hasProperty(error, "message")
495 ? error.message
496 : JSON.stringify(response.exit.cause),
497 data: response.exit.cause
498 } :
499 undefined
500 } as any
501 }
502 case "Defect":
503 return {
504 jsonrpc: "2.0",
505 id: jsonRpcInternalError,
506 error: {
507 _tag: "Defect",
508 code: 1,
509 message: "A defect occurred",
510 data: response.defect
511 }
512 }
513 case "ClientProtocolError":
514 return {} as never
515 }
516}
517
518const jsonRpcInternalError = -32603
519
520interface JsonRpcRequest {
521 readonly jsonrpc: "2.0"
522 readonly id?: number | string | null
523 readonly method: string
524 readonly params?: unknown
525 readonly headers?: ReadonlyArray<[string, string]>
526 readonly traceId?: string
527 readonly spanId?: string
528 readonly sampled?: boolean
529}
530
531interface JsonRpcResponse {
532 readonly jsonrpc: "2.0"
533 readonly id?: number | string | null
534 readonly result?: unknown
535 readonly chunk?: boolean
536 readonly error?: {
537 readonly code: number
538 readonly message: string
539 readonly data?: unknown
540 readonly _tag?: "Cause" | "Defect"
541 }
542}
543
544type JsonRpcMessage = JsonRpcRequest | JsonRpcResponse
545
546const defaultSchemaBinaryMaxFrameSize = 16 * 1024 * 1024
547
548const schemaBinaryTextEncoder = new TextEncoder()
549
550const makeSchemaBinary = (options?: {
551 readonly maxFrameSize?: number | "unbounded" | undefined
552 readonly fingerprintPayloads?: boolean | undefined
553}): RpcSerialization["Service"] => {
554 const maxFrameSize = options?.maxFrameSize === "unbounded"
555 ? undefined
556 : options?.maxFrameSize ?? defaultSchemaBinaryMaxFrameSize
557 const codecFor: CodecFor = options?.fingerprintPayloads === true
558 ? (schema) => SchemaBinary.toCodecDirect(schema, { fingerprint: true })
559 : SchemaBinary.toCodecDirect
560 const envelopeOptions = { fingerprint: true } as const
561 return RpcSerialization.of({
562 contentType: "application/vnd.effect.rpc+schema-binary",
563 includesFraming: true,
564 codecFor,
565 makeUnsafe: () => {
566 const parser = SchemaBinary.parser(RpcMessage.EncodedSchema, { ...envelopeOptions, maxFrameSize })
567 const encoder = SchemaBinary.encoder(RpcMessage.EncodedSchema, envelopeOptions)
568 return {
569 decode: (data) => parser.feedSync(typeof data === "string" ? schemaBinaryTextEncoder.encode(data) : data),
570 encode: (response) => {
571 if (!Array.isArray(response)) {
572 return encoder.encode(response)
573 }
574 if (response.length === 0) return undefined
575 return encoder.encodeMany(response)
576 }
577 }
578 }
579 })
580}
581
582/**
583 * RPC serialization layer that uses JSON for serialization.
584 *
585 * **When to use**
586 *
587 * Use when you have a transport protocol that already provides message framing.
588 *
589 * @see {@link layerNdjson} for transports that need newline-delimited framing
590 *
591 * @stability unstable
592 * @category layers
593 * @since 4.0.0
594 */
595export const layerJson: Layer.Layer<RpcSerialization> = Layer.succeed(RpcSerialization)(json)
596
597/**
598 * RPC serialization layer that uses NDJSON for serialization.
599 *
600 * **When to use**
601 *
602 * Use when you have a transport protocol that does not provide message framing.
603 *
604 * @see {@link layerJson} for transports that already provide message framing
605 *
606 * @stability unstable
607 * @category layers
608 * @since 4.0.0
609 */
610export const layerNdjson: Layer.Layer<RpcSerialization> = Layer.succeed(RpcSerialization)(ndjson)
611
612/**
613 * RPC serialization layer that uses NDJSON with custom streaming options.
614 *
615 * @stability unstable
616 * @category layers
617 * @since 4.0.0
618 */
619export const layerNdjsonWith = (options?: StreamOptions): Layer.Layer<RpcSerialization> =>
620 Layer.succeed(RpcSerialization)(makeNdjson(options))
621
622/**
623 * RPC serialization layer that uses JSON-RPC for serialization.
624 *
625 * @stability unstable
626 * @category layers
627 * @since 4.0.0
628 */
629export const layerJsonRpc = (options?: {
630 readonly contentType?: string | undefined
631}): Layer.Layer<RpcSerialization> => Layer.succeed(RpcSerialization)(jsonRpc(options))
632
633/**
634 * RPC serialization layer that uses newline-delimited JSON-RPC for
635 * serialization.
636 *
637 * @stability unstable
638 * @category layers
639 * @since 4.0.0
640 */
641export const layerNdJsonRpc = (options?: {
642 readonly contentType?: string | undefined
643 readonly maxBufferSize?: number | "unbounded" | undefined
644}): Layer.Layer<RpcSerialization> => Layer.succeed(RpcSerialization)(ndJsonRpc(options))
645
646/**
647 * RPC serialization layer that uses SchemaBinary with fingerprinted RPC
648 * envelopes. Payload fingerprints are disabled by default to support compatible
649 * schema evolution. Frames default to a 16 MiB maximum size. Use `"unbounded"`
650 * to disable the frame-size limit.
651 *
652 * @stability unstable
653 * @category layers
654 * @since 4.0.0
655 */
656export const layerSchemaBinary = (options?: {
657 readonly maxFrameSize?: number | "unbounded" | undefined
658 readonly fingerprintPayloads?: boolean | undefined
659}): Layer.Layer<RpcSerialization> => Layer.sync(RpcSerialization)(() => makeSchemaBinary(options))