1/**
2 * Defines the message shapes moved through Effect Cluster.
3 *
4 * Messages carry entity requests and control envelopes between callers, durable
5 * storage, transports, and runner handlers. This module includes incoming and
6 * outgoing variants for encoded stored requests, decoded local requests,
7 * acknowledgements, and interrupts. It also provides helpers for local delivery
8 * and for encoding or decoding request payloads with matching RPC schemas.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Context from "../Context.ts"
14import * as Data from "../Data.ts"
15import * as Effect from "../Effect.ts"
16import * as Option from "../Option.ts"
17import * as Rpc from "../rpc/Rpc.ts"
18import type * as RpcSerialization from "../rpc/RpcSerialization.ts"
19import * as Schema from "../Schema.ts"
20import type * as Scope from "../Scope.ts"
21import type { PersistenceError } from "./ClusterError.ts"
22import { MalformedMessage } from "./ClusterError.ts"
23import * as ClusterSchema from "./ClusterSchema.ts"
24import type { EntityAddress } from "./EntityAddress.ts"
25import * as Envelope from "./Envelope.ts"
26import type * as Reply from "./Reply.ts"
27import type { Snowflake } from "./Snowflake.ts"
28
29/**
30 * Message read by a runner from storage or transport.
31 *
32 * **Details**
33 *
34 * An incoming message is either a persisted request with an encoded payload or an
35 * incoming control envelope.
36 *
37 * @stability unstable
38 * @category models
39 * @since 4.0.0
40 */
41export type Incoming<R extends Rpc.Any> = IncomingRequest<R> | IncomingEnvelope
42
43/**
44 * Locally decoded incoming message for in-process delivery.
45 *
46 * **Details**
47 *
48 * It is either a request with a decoded payload or an incoming control envelope.
49 *
50 * @stability unstable
51 * @category models
52 * @since 4.0.0
53 */
54export type IncomingLocal<R extends Rpc.Any> = IncomingRequestLocal<R> | IncomingEnvelope
55
56/**
57 * Converts an outgoing message into a locally deliverable incoming message.
58 *
59 * **Details**
60 *
61 * Request messages keep their decoded payload and response callback, while
62 * control envelopes are wrapped as incoming envelopes.
63 *
64 * @stability unstable
65 * @category converting
66 * @since 4.0.0
67 */
68export const incomingLocalFromOutgoing = <R extends Rpc.Any>(self: Outgoing<R>): IncomingLocal<R> => {
69 if (self._tag === "OutgoingEnvelope") {
70 return new IncomingEnvelope({ envelope: self.envelope })
71 }
72 return new IncomingRequestLocal({
73 annotations: Context.get(self.rpc.annotations, ClusterSchema.Dynamic)(
74 self.rpc.annotations,
75 self.envelope as any
76 ),
77 envelope: self.envelope,
78 respond: self.respond,
79 lastSentReply: Option.none()
80 })
81}
82
83/**
84 * Represents an incoming persisted request whose payload has not yet been decoded with the RPC
85 * schema.
86 *
87 * **Details**
88 *
89 * It carries the last reply that was sent and a callback for persisting encoded
90 * replies.
91 *
92 * @stability unstable
93 * @category models
94 * @since 4.0.0
95 */
96export class IncomingRequest<R extends Rpc.Any> extends Data.TaggedClass("IncomingRequest")<{
97 readonly envelope: Envelope.PartialRequest
98 readonly lastSentReply: Option.Option<Reply.Encoded>
99 readonly respond: (reply: Reply.ReplyWithContext<R>) => Effect.Effect<void, MalformedMessage | PersistenceError>
100 /**
101 * Codec that filled the payload and reply holes of this message.
102 *
103 * Messages read from `MessageStorage` use the JSON codec, while the runner
104 * server sets it to the codec of its transport.
105 */
106 readonly codecFor: RpcSerialization.CodecFor
107 /**
108 * Scope that cancels interruptible, non-persisted requests when closed.
109 *
110 * @stability unstable
111 * @since 4.0.0
112 */
113 readonly callerScope?: Scope.Scope | undefined
114}> {}
115
116/**
117 * Represents an incoming request for local delivery with a decoded payload.
118 *
119 * **Details**
120 *
121 * It includes dynamic annotations, the last sent reply, and a callback for
122 * replying with decoded replies.
123 *
124 * @stability unstable
125 * @category models
126 * @since 4.0.0
127 */
128export class IncomingRequestLocal<R extends Rpc.Any> extends Data.TaggedClass("IncomingRequestLocal")<{
129 readonly envelope: Envelope.Request<R>
130 readonly lastSentReply: Option.Option<Reply.Reply<R>>
131 readonly respond: (reply: Reply.Reply<R>) => Effect.Effect<void, MalformedMessage | PersistenceError>
132 readonly annotations: Context.Context<never>
133 /**
134 * Scope that cancels interruptible, non-persisted requests when closed.
135 *
136 * @stability unstable
137 * @since 4.0.0
138 */
139 readonly callerScope?: Scope.Scope | undefined
140}> {}
141
142/**
143 * Represents an incoming control envelope carrying an `AckChunk` or `Interrupt`.
144 *
145 * @stability unstable
146 * @category models
147 * @since 4.0.0
148 */
149export class IncomingEnvelope extends Data.TaggedClass("IncomingEnvelope")<{
150 readonly _tag: "IncomingEnvelope"
151 readonly envelope: Envelope.AckChunk | Envelope.Interrupt
152}> {}
153
154/**
155 * Message produced for storage or transport.
156 *
157 * **Details**
158 *
159 * An outgoing message is either an entity request or a control envelope.
160 *
161 * @stability unstable
162 * @category models
163 * @since 4.0.0
164 */
165export type Outgoing<R extends Rpc.Any> = OutgoingRequest<R> | OutgoingEnvelope
166
167/**
168 * Represents an outgoing entity request with decoded payload and RPC metadata.
169 *
170 * **Details**
171 *
172 * It carries the service context used for serialization, the last received reply,
173 * the reply callback, dynamic annotations, and an optional encoded request cache.
174 *
175 * @stability unstable
176 * @category models
177 * @since 4.0.0
178 */
179export class OutgoingRequest<R extends Rpc.Any> extends Data.TaggedClass("OutgoingRequest")<{
180 readonly envelope: Envelope.Request<R>
181 readonly context: Context.Context<Rpc.Services<R>>
182 readonly lastReceivedReply: Option.Option<Reply.Reply<R>>
183 readonly rpc: R
184 readonly respond: (reply: Reply.Reply<R>) => Effect.Effect<void>
185 readonly annotations: Context.Context<never>
186}> {
187 /**
188 * Cached encoded envelope payload and the codec that produced it. The cache
189 * is reused only when the requested codec is the same function.
190 *
191 * @stability unstable
192 * @since 4.0.0
193 */
194 public encodedCache?: {
195 readonly codecFor: RpcSerialization.CodecFor
196 readonly envelope: Envelope.PartialRequest
197 }
198}
199
200/**
201 * Represents an outgoing control envelope paired with RPC metadata.
202 *
203 * **When to use**
204 *
205 * Use to construct an interrupt envelope for an
206 * in-flight request.
207 *
208 * @stability unstable
209 * @category models
210 * @since 4.0.0
211 */
212export class OutgoingEnvelope extends Data.TaggedClass("OutgoingEnvelope")<{
213 readonly envelope: Envelope.AckChunk | Envelope.Interrupt
214 readonly rpc: Rpc.AnyWithProps
215}> {
216 /**
217 * Creates an outgoing interrupt envelope for the supplied request.
218 *
219 * @stability unstable
220 * @since 4.0.0
221 */
222 static interrupt(options: {
223 readonly address: EntityAddress
224 readonly id: Snowflake
225 readonly requestId: Snowflake
226 }): OutgoingEnvelope {
227 return new OutgoingEnvelope({
228 envelope: new Envelope.Interrupt(options),
229 rpc: neverRpc
230 })
231 }
232}
233
234const codecForJson = Schema.toCodecJson as RpcSerialization.CodecFor
235
236const neverRpc = Rpc.make("Never", {
237 success: Schema.Never as any,
238 error: Schema.Never,
239 payload: {}
240})
241
242/**
243 * Serializes an outgoing message into a partial envelope.
244 *
245 * **Details**
246 *
247 * Control envelopes pass through unchanged. Requests are encoded with their RPC
248 * payload schema, reusing the cached encoded request when available.
249 *
250 * @stability unstable
251 * @category serialization
252 * @since 4.0.0
253 */
254export const serialize = <Rpc extends Rpc.Any>(
255 message: Outgoing<Rpc>,
256 codecFor: RpcSerialization.CodecFor
257): Effect.Effect<Envelope.Partial, MalformedMessage> => {
258 if (message._tag !== "OutgoingRequest") {
259 return Effect.succeed(message.envelope)
260 }
261 return Effect.suspend(() => {
262 const cached = message.encodedCache
263 if (cached?.codecFor === codecFor) {
264 return Effect.succeed(cached.envelope)
265 }
266 return Effect.tap(serializeRequest(message, codecFor), (envelope) =>
267 Effect.sync(() => {
268 message.encodedCache = { codecFor, envelope }
269 }))
270 })
271}
272
273/**
274 * Serializes an outgoing message into its JSON envelope representation.
275 *
276 * **Details**
277 *
278 * Schema encoding failures are converted to `MalformedMessage`.
279 *
280 * @stability unstable
281 * @category serialization
282 * @since 4.0.0
283 */
284export const serializeEnvelope = <Rpc extends Rpc.Any>(
285 message: Outgoing<Rpc>
286): Effect.Effect<Envelope.Encoded, MalformedMessage, never> =>
287 Effect.flatMap(
288 serialize(message, codecForJson),
289 (envelope) => MalformedMessage.refail(Schema.encodeEffect(Envelope.PartialJson)(envelope))
290 )
291
292/**
293 * Encodes the payload of an `OutgoingRequest` with the request's RPC payload
294 * schema and service context.
295 *
296 * **Details**
297 *
298 * The result is a `PartialRequest` suitable for storage or transport.
299 *
300 * @stability unstable
301 * @category serialization
302 * @since 4.0.0
303 */
304export const serializeRequest = <Rpc extends Rpc.Any>(
305 self: OutgoingRequest<Rpc>,
306 codecFor: RpcSerialization.CodecFor
307): Effect.Effect<Envelope.PartialRequest, MalformedMessage> => {
308 const rpc = self.rpc as any as Rpc.AnyWithProps
309 return Schema.encodeEffect(codecFor(rpc.payloadSchema))(self.envelope.payload).pipe(
310 Effect.provideContext(self.context),
311 MalformedMessage.refail,
312 Effect.map((payload) => ({
313 ...self.envelope,
314 payload
315 }))
316 ) as any as Effect.Effect<Envelope.PartialRequest, MalformedMessage>
317}
318
319/**
320 * Decodes a partial envelope back into a locally deliverable incoming message.
321 *
322 * **Details**
323 *
324 * Control envelopes pass through directly. Request envelopes require the original
325 * `OutgoingRequest` so the payload can be decoded with the correct RPC schema and
326 * context.
327 *
328 * @stability unstable
329 * @category serialization
330 * @since 4.0.0
331 */
332export const deserializeLocal = <Rpc extends Rpc.Any>(
333 self: Outgoing<Rpc>,
334 encoded: Envelope.Partial,
335 codecFor: RpcSerialization.CodecFor
336): Effect.Effect<
337 IncomingLocal<Rpc>,
338 MalformedMessage
339> => {
340 if (encoded._tag !== "Request") {
341 return Effect.succeed(new IncomingEnvelope({ envelope: encoded }))
342 } else if (self._tag !== "OutgoingRequest") {
343 return Effect.fail(
344 new MalformedMessage({ cause: new Error("Can only deserialize a Request with an OutgoingRequest message") })
345 )
346 }
347 const rpc = self.rpc as any as Rpc.AnyWithProps
348 return Schema.decodeEffect(codecFor(rpc.payloadSchema))(encoded.payload).pipe(
349 Effect.provideContext(self.context),
350 MalformedMessage.refail,
351 Effect.map((payload) => {
352 const envelope = Envelope.makeRequest({
353 ...encoded,
354 payload
355 } as any) as Envelope.Request<Rpc>
356 return new IncomingRequestLocal({
357 envelope,
358 lastSentReply: Option.none(),
359 respond: self.respond,
360 annotations: Context.get(rpc.annotations, ClusterSchema.Dynamic)(
361 rpc.annotations,
362 envelope as any
363 )
364 })
365 })
366 ) as Effect.Effect<IncomingRequestLocal<Rpc>, MalformedMessage>
367}