1/**
2 * Defines reply values produced by clustered RPC execution.
3 *
4 * Every reply belongs to a request and is either a final `WithExit`, which
5 * carries the final RPC `Exit`, or a streaming `Chunk`, which carries a
6 * non-empty batch of success values. This module includes runtime and encoded
7 * reply shapes, guards, per-RPC schema builders, `ReplyWithContext` for
8 * carrying encoding services, and serialization helpers for storage or
9 * transport.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import type { NonEmptyReadonlyArray } from "../Array.ts"
15import * as Context from "../Context.ts"
16import * as Data from "../Data.ts"
17import * as Effect from "../Effect.ts"
18import * as Exit from "../Exit.ts"
19import * as Option from "../Option.ts"
20import { hasProperty } from "../Predicate.ts"
21import * as Rpc from "../rpc/Rpc.ts"
22import type * as RpcMessage from "../rpc/RpcMessage.ts"
23import type * as RpcSchema from "../rpc/RpcSchema.ts"
24import type * as RpcSerialization from "../rpc/RpcSerialization.ts"
25import * as Schema from "../Schema.ts"
26import * as SchemaIssue from "../SchemaIssue.ts"
27import * as SchemaParser from "../SchemaParser.ts"
28import * as SchemaTransformation from "../SchemaTransformation.ts"
29import { MalformedMessage } from "./ClusterError.ts"
30import * as Envelope from "./Envelope.ts"
31import type { OutgoingRequest } from "./Message.ts"
32import { Snowflake, SnowflakeFromBigInt } from "./Snowflake.ts"
33
34const TypeId = "~effect/cluster/Reply"
35
36/**
37 * Returns `true` when the supplied value is a runtime cluster reply, based on the
38 * reply type identifier.
39 *
40 * @stability unstable
41 * @category guards
42 * @since 4.0.0
43 */
44export const isReply = (u: unknown): u is Reply<Rpc.Any> => hasProperty(u, TypeId)
45
46/**
47 * Runtime reply sent for an RPC request, either as a final exit or a chunk of a
48 * streaming success value.
49 *
50 * @stability unstable
51 * @category models
52 * @since 4.0.0
53 */
54export type Reply<R extends Rpc.Any> = WithExit<R> | Chunk<R>
55
56/**
57 * JSON-serializable form of a cluster reply.
58 *
59 * @stability unstable
60 * @category models
61 * @since 4.0.0
62 */
63export type Encoded = WithExitEncoded | ChunkEncoded
64
65/**
66 * Schema for reply values that are already in encoded form.
67 *
68 * **Details**
69 *
70 * Per-RPC payload validation is performed by `Reply(rpc)`.
71 *
72 * @stability unstable
73 * @category schemas
74 * @since 4.0.0
75 */
76export const Encoded: Schema.Codec<Encoded> = Envelope.OpaqueHole as any
77
78/**
79 * Represents a cluster reply paired with the RPC definition and service context required to
80 * serialize it for transport.
81 *
82 * **When to use**
83 *
84 * Use to carry a runtime reply together with the RPC schema and services needed
85 * to encode it for storage or transport.
86 *
87 * @stability unstable
88 * @category models
89 * @since 4.0.0
90 */
91export class ReplyWithContext<R extends Rpc.Any> extends Data.TaggedClass("ReplyWithContext")<{
92 readonly reply: Reply<R>
93 readonly context: Context.Context<Rpc.Services<R>>
94 readonly rpc: R
95}> {
96 /**
97 * Creates a terminal reply context that dies with the supplied defect.
98 *
99 * @stability unstable
100 * @since 4.0.0
101 */
102 static fromDefect(options: {
103 readonly id: Snowflake
104 readonly requestId: Snowflake
105 readonly defect: unknown
106 }): ReplyWithContext<any> {
107 return new ReplyWithContext({
108 reply: new WithExit({
109 requestId: options.requestId,
110 id: options.id,
111 exit: Exit.die(Schema.encodeSync(Schema.Defect())(options.defect))
112 }),
113 context: Context.empty() as any,
114 rpc: neverRpc
115 })
116 }
117 /**
118 * Creates a terminal reply context that interrupts the supplied request.
119 *
120 * @stability unstable
121 * @since 4.0.0
122 */
123 static interrupt(options: {
124 readonly id: Snowflake
125 readonly requestId: Snowflake
126 }): ReplyWithContext<any> {
127 return new ReplyWithContext({
128 reply: new WithExit({
129 requestId: options.requestId,
130 id: options.id,
131 exit: Exit.interrupt()
132 }),
133 context: Context.empty() as any,
134 rpc: neverRpc
135 })
136 }
137}
138
139const neverRpc = Rpc.make("Never", {
140 success: Schema.Never as any,
141 error: Schema.Never,
142 payload: {}
143})
144
145/**
146 * Wire-format representation of a terminal reply containing the request id, reply
147 * id, and encoded RPC exit value.
148 *
149 * @stability unstable
150 * @category models
151 * @since 4.0.0
152 */
153export interface WithExitEncoded<A = unknown, E = unknown> {
154 readonly _tag: "WithExit"
155 readonly requestId: string
156 readonly id: string
157 readonly exit: RpcMessage.ExitEncoded<A, E>
158}
159
160/**
161 * Wire-format representation of a streaming reply chunk, including the request id,
162 * reply id, sequence number, and non-empty encoded values.
163 *
164 * @stability unstable
165 * @category models
166 * @since 4.0.0
167 */
168export interface ChunkEncoded {
169 readonly _tag: "Chunk"
170 readonly requestId: string
171 readonly id: string
172 readonly sequence: number
173 readonly values: NonEmptyReadonlyArray<unknown>
174}
175
176// Keyed by codec first: the storage path and the transport path can compile
177// different codecs for the same RPC.
178const schemaCaches = new WeakMap<RpcSerialization.CodecFor, WeakMap<Rpc.Any, Schema.Top>>()
179
180/**
181 * Represents a streaming RPC reply chunk for a request, carrying a non-empty
182 * batch of success values together with the reply id and sequence number.
183 *
184 * @stability unstable
185 * @category models
186 * @since 4.0.0
187 */
188export class Chunk<R extends Rpc.Any> extends Data.TaggedClass("Chunk")<{
189 readonly requestId: Snowflake
190 readonly id: Snowflake
191 readonly sequence: number
192 readonly values: NonEmptyReadonlyArray<Rpc.SuccessChunk<R>>
193}> {
194 /**
195 * Marks this value as a runtime cluster reply.
196 *
197 * @stability unstable
198 * @since 4.0.0
199 */
200 readonly [TypeId] = TypeId
201
202 /**
203 * Creates an empty chunk reply for the supplied request id.
204 *
205 * @stability unstable
206 * @since 4.0.0
207 */
208 static emptyFrom(requestId: Snowflake) {
209 return new Chunk({
210 requestId,
211 id: Snowflake(BigInt(0)),
212 sequence: 0,
213 values: [undefined]
214 })
215 }
216
217 /**
218 * Schema that accepts any runtime chunk reply without validating payload values.
219 *
220 * @stability unstable
221 * @since 4.0.0
222 */
223 static readonly Any = Schema.declare((u): u is Chunk<never> => isReply(u) && u._tag === "Chunk")
224
225 /**
226 * Transformation between encoded chunk records and `Chunk` instances.
227 *
228 * @stability unstable
229 * @since 4.0.0
230 */
231 static readonly transform: SchemaTransformation.Transformation<any, any> = SchemaTransformation.transform({
232 decode: (a: any) => new Chunk(a),
233 encode: (a) => a as any
234 })
235
236 /**
237 * Builds a chunk schema from the streaming success schema of an RPC.
238 *
239 * @stability unstable
240 * @since 4.0.0
241 */
242 static schema<R extends Rpc.Any>(
243 rpc: R
244 ): Schema.declareConstructor<Chunk<R>, Chunk<R>, readonly [Rpc.SuccessExitSchema<R>]> {
245 const successSchema = ((rpc as any as Rpc.AnyWithProps).successSchema as RpcSchema.Stream<any, any>).success
246 if (!successSchema) {
247 return Schema.Never as any
248 }
249 return this.schemaFrom(successSchema) as any
250 }
251
252 /**
253 * Builds a chunk schema that validates each success value with the supplied schema.
254 *
255 * @stability unstable
256 * @since 4.0.0
257 */
258 static schemaFrom<Success extends Schema.Constraint>(
259 success: Success
260 ): Schema.declareConstructor<Chunk<Rpc.Any>, Chunk<Rpc.Any>, readonly [Success]> {
261 // TODO: extract to a helper function
262 return Schema.declareConstructor<Chunk<Rpc.Any>>()(
263 [success],
264 ([success]) => (input, ast, options) => {
265 if (!isReply(input) || input._tag !== "Chunk") {
266 return Effect.fail(new SchemaIssue.InvalidType(ast, input, options))
267 }
268 return Effect.mapBothEager(SchemaParser.decodeEffect(Schema.NonEmptyArray(success))(input.values, options), {
269 onFailure: (issue) => SchemaIssue.makeCompositeAtKey(ast, "values", issue, input, options),
270 onSuccess: (values) => new Chunk({ ...input, values } as any)
271 })
272 },
273 {
274 expected: "Reply.Chunk",
275 toCodecJson: ([success]) =>
276 Schema.link<Chunk<Rpc.Any>>()(
277 Schema.Struct({
278 _tag: Schema.Literal("Chunk"),
279 requestId: SnowflakeFromBigInt,
280 id: SnowflakeFromBigInt,
281 sequence: Schema.Int,
282 values: Schema.NonEmptyArray(success)
283 }),
284 SchemaTransformation.transform({
285 decode: (encoded) => new Chunk(encoded),
286 encode: (result) => ({ ...result })
287 })
288 )
289 }
290 )
291 }
292
293 /**
294 * Returns a copy of this chunk associated with the supplied request id.
295 *
296 * @stability unstable
297 * @since 4.0.0
298 */
299 withRequestId(requestId: Snowflake): Chunk<R> {
300 return new Chunk({
301 ...this,
302 requestId
303 })
304 }
305}
306
307/**
308 * Represents a terminal RPC reply for a request, carrying the final `Exit` for the remote
309 * call.
310 *
311 * **When to use**
312 *
313 * Use to represent the final success, typed failure, defect, or interruption
314 * for a clustered RPC request.
315 *
316 * @stability unstable
317 * @category models
318 * @since 4.0.0
319 */
320export class WithExit<R extends Rpc.Any> extends Data.TaggedClass("WithExit")<{
321 readonly requestId: Snowflake
322 readonly id: Snowflake
323 readonly exit: Rpc.Exit<R>
324}> {
325 /**
326 * Marks this value as a runtime cluster reply.
327 *
328 * @stability unstable
329 * @since 4.0.0
330 */
331 readonly [TypeId] = TypeId
332
333 /**
334 * Returns `true` when the value is a terminal `WithExit` reply.
335 *
336 * @stability unstable
337 * @since 4.0.0
338 */
339 static is(u: unknown): u is WithExit<any> {
340 return isReply(u) && u._tag === "WithExit"
341 }
342
343 /**
344 * Builds a terminal reply schema from the exit schema of an RPC.
345 *
346 * @stability unstable
347 * @since 4.0.0
348 */
349 static schema<R extends Rpc.Any>(
350 rpc: R
351 ): Schema.declareConstructor<
352 WithExit<R>,
353 WithExit<R>,
354 readonly [Schema.Exit<Rpc.SuccessExitSchema<R>, Rpc.ErrorExitSchema<R>, Rpc.DefectSchema>]
355 > {
356 return this.schemaFrom(Rpc.exitSchema(rpc))
357 }
358
359 /**
360 * Builds a terminal reply schema that validates the encoded exit value.
361 *
362 * @stability unstable
363 * @since 4.0.0
364 */
365 static schemaFrom<
366 Success extends Schema.Constraint,
367 Error extends Schema.Constraint,
368 Defect extends Schema.Constraint
369 >(
370 exitSchema: Schema.Exit<Success, Error, Defect>
371 ): Schema.declareConstructor<
372 WithExit<Rpc.Any>,
373 WithExit<Rpc.Any>,
374 readonly [Schema.Exit<Success, Error, Defect>]
375 > {
376 // TODO: extract to a helper function
377 return Schema.declareConstructor<WithExit<Rpc.Any>>()(
378 [exitSchema],
379 ([exit]) => (input, ast, options) => {
380 if (!isReply(input) || input._tag !== "WithExit") {
381 return Effect.fail(new SchemaIssue.InvalidType(ast, input, options))
382 }
383 return Effect.mapBothEager(SchemaParser.decodeEffect(exit)(input.exit, options), {
384 onFailure: (issue) => SchemaIssue.makeCompositeAtKey(ast, "exit", issue, input, options),
385 onSuccess: (exit) => new WithExit({ ...input, exit: exit as any })
386 })
387 },
388 {
389 expected: "Reply.WithExit",
390 toCodecJson: ([exit]) =>
391 Schema.link<WithExit<Rpc.Any>>()(
392 Schema.Struct({
393 _tag: Schema.Literal("WithExit"),
394 requestId: SnowflakeFromBigInt,
395 id: SnowflakeFromBigInt,
396 exit
397 }),
398 SchemaTransformation.transform({
399 decode: (encoded) => new WithExit(encoded as any),
400 encode: (result) => ({ ...result })
401 })
402 )
403 }
404 )
405 }
406
407 /**
408 * Returns a copy of this terminal reply associated with the supplied request id.
409 *
410 * @stability unstable
411 * @since 4.0.0
412 */
413 withRequestId(requestId: Snowflake): WithExit<R> {
414 return new WithExit({
415 ...this,
416 requestId
417 })
418 }
419}
420
421/**
422 * Builds the transport codec for replies to the specified RPC, covering terminal
423 * `WithExit` replies and streaming `Chunk` replies.
424 *
425 * @stability unstable
426 * @category schemas
427 * @since 4.0.0
428 */
429export const Reply = <R extends Rpc.Any>(
430 rpc: R,
431 codecFor: RpcSerialization.CodecFor
432): Schema.Codec<
433 WithExit<R> | Chunk<R>,
434 Encoded,
435 Rpc.ServicesClient<R>,
436 Rpc.ServicesServer<R>
437> => {
438 let schemaCache = schemaCaches.get(codecFor)
439 if (schemaCache === undefined) {
440 schemaCache = new WeakMap()
441 schemaCaches.set(codecFor, schemaCache)
442 }
443 const cached = schemaCache.get(rpc)
444 if (cached !== undefined) {
445 return cached as any
446 }
447 const schema = codecFor(Schema.Union([WithExit.schema(rpc), Chunk.schema(rpc)]))
448 schemaCache.set(rpc, schema)
449 return schema as any
450}
451
452/**
453 * Serializes a `ReplyWithContext` into its encoded wire representation, using the
454 * reply's RPC schema and context and refailing encoding errors as
455 * `MalformedMessage`.
456 *
457 * @stability unstable
458 * @category serialization
459 * @since 4.0.0
460 */
461export const serialize = <R extends Rpc.Any>(
462 self: ReplyWithContext<R>,
463 codecFor: RpcSerialization.CodecFor
464): Effect.Effect<Encoded, MalformedMessage> => {
465 const schema = Reply(self.rpc, codecFor)
466 return MalformedMessage.refail(
467 Effect.provideContext(
468 Schema.encodeEffect(schema)(self.reply),
469 self.context
470 )
471 )
472}
473
474/**
475 * Serializes a `ReplyWithContext`, falling back to a serializable defect reply
476 * when the original reply cannot be encoded.
477 *
478 * @stability unstable
479 * @category serialization
480 * @since 4.0.0
481 */
482export const serializeOrDefect = <R extends Rpc.Any>(
483 self: ReplyWithContext<R>,
484 codecFor: RpcSerialization.CodecFor
485): Effect.Effect<Encoded> =>
486 Effect.catchTag(
487 serialize(self, codecFor),
488 "MalformedMessage",
489 (error) =>
490 Effect.orDie(serialize(
491 ReplyWithContext.fromDefect({
492 id: self.reply.id,
493 requestId: self.reply.requestId,
494 defect: error
495 }),
496 codecFor
497 ))
498 )
499
500/**
501 * Serializes an outgoing request's last received reply when one exists, returning
502 * `None` when no reply has been received and refailing encoding errors as
503 * `MalformedMessage`.
504 *
505 * @stability unstable
506 * @category serialization
507 * @since 4.0.0
508 */
509export const serializeLastReceived = <R extends Rpc.Any>(
510 self: OutgoingRequest<R>,
511 codecFor: RpcSerialization.CodecFor
512): Effect.Effect<Option.Option<Encoded>, MalformedMessage> => {
513 const lastReceivedReply = self.lastReceivedReply
514 if (lastReceivedReply._tag === "None") {
515 return Effect.succeedNone
516 }
517 const schema = Reply(self.rpc, codecFor)
518 return MalformedMessage.refail(
519 Effect.provideContext(Schema.encodeEffect(schema)(lastReceivedReply.value), self.context)
520 ).pipe(
521 Effect.map(Option.some)
522 )
523}