1/**
2 * Initial messages for worker-backed RPC protocols.
3 *
4 * A worker-backed RPC client can send one schema-encoded value before normal RPC
5 * requests are handled. This module defines the `InitialMessage` service, a
6 * helper for encoding that value while collecting transferables, a layer for
7 * providing it to the client protocol, and a decoder for reading it from the
8 * worker server protocol.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import type { NoSuchElementError } from "../Cause.ts"
14import * as Context from "../Context.ts"
15import * as Effect from "../Effect.ts"
16import * as Layer from "../Layer.ts"
17import * as Schema from "../Schema.ts"
18import * as Transferable from "../workers/Transferable.ts"
19import type { Protocol } from "./RpcServer.ts"
20
21/**
22 * Context service that supplies the initial RPC worker message as encoded data
23 * paired with any transferables that should be posted with it.
24 *
25 * @stability unstable
26 * @category services
27 * @since 4.0.0
28 */
29export class InitialMessage extends Context.Service<
30 InitialMessage,
31 Effect.Effect<
32 readonly [
33 data: unknown,
34 transfers: ReadonlyArray<Transferable>
35 ]
36 >
37>()("effect/rpc/RpcWorker/InitialMessage") {}
38
39/**
40 * Types related to the encoded initial message exchanged with an RPC worker.
41 *
42 * @stability unstable
43 * @since 4.0.0
44 */
45export declare namespace InitialMessage {
46 /**
47 * Tagged wire representation of an RPC worker initial message after schema
48 * encoding.
49 *
50 * @stability unstable
51 * @category models
52 * @since 4.0.0
53 */
54 export interface Encoded {
55 readonly _tag: "InitialMessage"
56 readonly value: unknown
57 }
58}
59
60const ProtocolTag = Context.Service<Protocol, Protocol["Service"]>(
61 "effect/rpc/RpcServer/Protocol" satisfies Protocol["key"]
62)
63
64/**
65 * Runs an effect, encodes its result with the schema's JSON codec, and returns
66 * the encoded value together with collected transferables.
67 *
68 * @stability unstable
69 * @category encoding
70 * @since 4.0.0
71 */
72export const makeInitialMessage = <S extends Schema.Constraint, E, R2>(
73 schema: S,
74 effect: Effect.Effect<S["Type"], E, R2>
75): Effect.Effect<
76 readonly [data: unknown, transferables: ReadonlyArray<globalThis.Transferable>],
77 E | Schema.SchemaError,
78 S["EncodingServices"] | R2
79> => {
80 const schemaJson = Schema.toCodecJson(schema)
81 return Effect.flatMap(effect, (value) => {
82 const collector = Transferable.makeCollectorUnsafe()
83 return Schema.encodeEffect(schemaJson)(value).pipe(
84 Effect.provideService(Transferable.Collector, collector),
85 Effect.map((encoded) => [encoded, collector.clearUnsafe()] as const)
86 )
87 })
88}
89
90/**
91 * Provides the `InitialMessage` service from a schema and build effect,
92 * capturing the layer context and dying if schema encoding fails.
93 *
94 * @stability unstable
95 * @category layers
96 * @since 4.0.0
97 */
98export const layerInitialMessage = <S extends Schema.Constraint, R2>(
99 schema: S,
100 build: Effect.Effect<S["Type"], never, R2>
101): Layer.Layer<InitialMessage, never, S["EncodingServices"] | R2> =>
102 Layer.effect(InitialMessage)(
103 Effect.contextWith((context: Context.Context<S["EncodingServices"] | R2>) =>
104 Effect.succeed(
105 Effect.provideContext(Effect.orDie(makeInitialMessage(schema, build)), context)
106 )
107 )
108 )
109
110/**
111 * Reads the protocol initial message and decodes it with the supplied schema,
112 * failing if no initial message is available or decoding fails.
113 *
114 * @stability unstable
115 * @category decoding
116 * @since 4.0.0
117 */
118export const initialMessage = <S extends Schema.Constraint>(
119 schema: S
120): Effect.Effect<S["Type"], NoSuchElementError | Schema.SchemaError, Protocol | S["DecodingServices"]> =>
121 ProtocolTag.pipe(
122 Effect.flatMap((protocol) => protocol.initialMessage),
123 Effect.flatMap(Effect.fromOption),
124 Effect.flatMap(Schema.decodeUnknownEffect(Schema.toCodecJson(schema)))
125 )