1/**
2 * Defines the transport envelopes exchanged by cluster entities.
3 *
4 * Request envelopes wrap decoded RPC payloads with the target entity address,
5 * RPC tag, request id, headers, and optional tracing context. The module also
6 * includes acknowledgement envelopes for streamed reply chunks, interrupt
7 * envelopes for in-flight requests, JSON codecs for partially decoded
8 * envelopes, guards, request constructors, and storage primary-key helpers.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import type * as Headers from "../http/Headers.ts"
14import * as Predicate from "../Predicate.ts"
15import * as PrimaryKey from "../PrimaryKey.ts"
16import type { ReadonlyRecord } from "../Record.ts"
17import type * as Rpc from "../rpc/Rpc.ts"
18import * as Schema from "../Schema.ts"
19import * as SchemaTransformation from "../SchemaTransformation.ts"
20import { EntityAddress } from "./EntityAddress.ts"
21import { type Snowflake, SnowflakeFromBigInt } from "./Snowflake.ts"
22
23/**
24 * Type identifier used to mark runtime cluster envelope values.
25 *
26 * @stability unstable
27 * @category type IDs
28 * @since 4.0.0
29 */
30export const TypeId = "~effect/cluster/Envelope"
31
32/**
33 * Schema for a value that has already been encoded by the transport's hole
34 * codec.
35 *
36 * **Details**
37 *
38 * Cluster payloads are encoded twice: the entity payload is encoded with the
39 * entity RPC schema, and the result is carried opaquely inside the runner
40 * envelope. This schema names that hole so the outer runner encode leaves it
41 * alone. It is the identity under `Schema.toCodecJson`, so JSON, NDJSON, and
42 * text transports stay wire-compatible. A binary codec compiles it as a
43 * bytes leaf.
44 *
45 * @stability unstable
46 * @category schemas
47 * @since 4.0.0
48 */
49export const OpaqueHole: Schema.declare<any> = Schema.declare(
50 (_: unknown): _ is any => true,
51 {
52 expected: "an already-encoded value",
53 toCodecJson: () => undefined,
54 toCodec: () =>
55 Schema.link<any>()(
56 Schema.Uint8Array,
57 SchemaTransformation.passthrough()
58 )
59 }
60)
61
62/**
63 * Union of cluster envelopes exchanged for an RPC request.
64 *
65 * **Details**
66 *
67 * An envelope is either a request, an acknowledgement for a streamed reply chunk,
68 * or an interrupt signal.
69 *
70 * @stability unstable
71 * @category models
72 * @since 4.0.0
73 */
74export type Envelope<R extends Rpc.Any> = Request<R> | AckChunk | Interrupt
75
76/**
77 * JSON-serializable form of a cluster envelope.
78 *
79 * @stability unstable
80 * @category models
81 * @since 4.0.0
82 */
83export type Encoded = PartialRequestEncoded | AckChunkEncoded | InterruptEncoded
84
85/**
86 * Helper types associated with cluster envelopes.
87 *
88 * @stability unstable
89 * @since 4.0.0
90 */
91export declare namespace Envelope {
92 /**
93 * Envelope type for any RPC protocol.
94 *
95 * @stability unstable
96 * @category models
97 * @since 4.0.0
98 */
99 export type Any = Envelope<any>
100}
101
102/**
103 * Runtime envelope for an RPC request addressed to a specific entity.
104 *
105 * **Details**
106 *
107 * It carries the request ID, entity address, RPC tag, decoded payload, request
108 * headers, and optional tracing context.
109 *
110 * @stability unstable
111 * @category models
112 * @since 4.0.0
113 */
114export interface Request<in out Rpc extends Rpc.Any> {
115 readonly [TypeId]: typeof TypeId
116 readonly _tag: "Request"
117 readonly requestId: Snowflake
118 readonly address: EntityAddress
119 readonly tag: Rpc.Tag<Rpc>
120 readonly payload: Rpc.Payload<Rpc>
121 readonly headers: Headers.Headers
122 readonly traceId?: string
123 readonly spanId?: string
124 readonly sampled?: boolean
125}
126
127/**
128 * Schema for a request envelope before its RPC payload has been decoded.
129 *
130 * **Details**
131 *
132 * The envelope metadata is decoded, while the payload remains `unknown` until it
133 * is decoded with the target RPC payload schema.
134 *
135 * @stability unstable
136 * @category models
137 * @since 4.0.0
138 */
139export class PartialRequest extends Schema.Opaque<PartialRequest>()(Schema.Struct({
140 _tag: Schema.tag("Request"),
141 requestId: SnowflakeFromBigInt,
142 address: EntityAddress,
143 tag: Schema.String,
144 payload: OpaqueHole,
145 headers: Schema.Headers,
146 traceId: Schema.optional(Schema.String),
147 spanId: Schema.optional(Schema.String),
148 sampled: Schema.optional(Schema.Boolean)
149})) {}
150
151/**
152 * Serialized JSON shape of a request envelope.
153 *
154 * **Details**
155 *
156 * Identifiers are encoded as strings and the RPC payload remains unknown until
157 * decoded with the RPC schema.
158 *
159 * @stability unstable
160 * @category models
161 * @since 4.0.0
162 */
163export interface PartialRequestEncoded {
164 readonly _tag: "Request"
165 readonly requestId: string
166 readonly address: {
167 readonly shardId: {
168 readonly group: string
169 readonly id: number
170 }
171 readonly entityType: string
172 readonly entityId: string
173 }
174 readonly tag: string
175 readonly payload: unknown
176 readonly headers: ReadonlyRecord<string, string>
177 readonly traceId?: string
178 readonly spanId?: string
179 readonly sampled?: boolean
180}
181
182/**
183 * Represents an envelope acknowledging receipt of a streamed reply chunk for a
184 * request.
185 *
186 * **Details**
187 *
188 * The `replyId` identifies the chunk reply that has been received.
189 *
190 * @stability unstable
191 * @category models
192 * @since 4.0.0
193 */
194export class AckChunk extends Schema.Class<AckChunk>("effect/cluster/Envelope/AckChunk")({
195 _tag: Schema.tag("AckChunk"),
196 id: SnowflakeFromBigInt,
197 address: EntityAddress,
198 requestId: SnowflakeFromBigInt,
199 replyId: SnowflakeFromBigInt
200}) {
201 /**
202 * Marks this value as a cluster envelope for runtime guards.
203 *
204 * @stability unstable
205 * @since 4.0.0
206 */
207 readonly [TypeId] = TypeId
208
209 /**
210 * Returns a copy of this acknowledgement associated with the supplied request id.
211 *
212 * @stability unstable
213 * @since 4.0.0
214 */
215 withRequestId(requestId: Snowflake): AckChunk {
216 return new AckChunk({
217 ...this,
218 requestId
219 })
220 }
221}
222
223/**
224 * Serialized JSON shape of an `AckChunk` envelope.
225 *
226 * @stability unstable
227 * @category models
228 * @since 4.0.0
229 */
230export interface AckChunkEncoded {
231 readonly _tag: "AckChunk"
232 readonly id: string
233 readonly address: {
234 readonly shardId: {
235 readonly group: string
236 readonly id: number
237 }
238 readonly entityType: string
239 readonly entityId: string
240 }
241 readonly requestId: string
242 readonly replyId: string
243}
244
245/**
246 * Represents an envelope used to interrupt an in-flight entity request.
247 *
248 * @stability unstable
249 * @category models
250 * @since 4.0.0
251 */
252export class Interrupt extends Schema.Class<Interrupt>("effect/cluster/Envelope/Interrupt")({
253 _tag: Schema.tag("Interrupt"),
254 id: SnowflakeFromBigInt,
255 address: EntityAddress,
256 requestId: SnowflakeFromBigInt
257}) {
258 /**
259 * Marks this value as a cluster envelope for runtime guards.
260 *
261 * @stability unstable
262 * @since 4.0.0
263 */
264 readonly [TypeId] = TypeId
265
266 /**
267 * Returns a copy of this interrupt associated with the supplied request id.
268 *
269 * @stability unstable
270 * @since 4.0.0
271 */
272 withRequestId(requestId: Snowflake): Interrupt {
273 return new Interrupt({
274 ...this,
275 requestId
276 })
277 }
278}
279
280/**
281 * Serialized JSON shape of an `Interrupt` envelope.
282 *
283 * @stability unstable
284 * @category models
285 * @since 4.0.0
286 */
287export interface InterruptEncoded {
288 readonly _tag: "Interrupt"
289 readonly id: string
290 readonly address: {
291 readonly shardId: {
292 readonly group: string
293 readonly id: number
294 }
295 readonly entityType: string
296 readonly entityId: string
297 }
298 readonly requestId: string
299}
300
301/**
302 * Schema for partially decoded cluster envelopes.
303 *
304 * **Details**
305 *
306 * It accepts `PartialRequest`, `AckChunk`, and `Interrupt` envelope values.
307 *
308 * @stability unstable
309 * @category schemas
310 * @since 4.0.0
311 */
312export const Partial: Schema.Union<
313 readonly [
314 typeof PartialRequest,
315 typeof AckChunk,
316 typeof Interrupt
317 ]
318> = Schema.Union([PartialRequest, AckChunk, Interrupt])
319
320/**
321 * Decoded value type produced by the `Partial` envelope schema.
322 *
323 * @stability unstable
324 * @category schemas
325 * @since 4.0.0
326 */
327export type Partial = typeof Partial.Type
328
329/**
330 * JSON codec for partial cluster envelopes.
331 *
332 * @stability unstable
333 * @category schemas
334 * @since 4.0.0
335 */
336export const PartialJson: Schema.Codec<
337 AckChunk | Interrupt | PartialRequest,
338 Encoded
339> = Schema.toCodecJson(Partial) as any
340
341/**
342 * Schema for mutable arrays of JSON-encoded partial cluster envelopes.
343 *
344 * @stability unstable
345 * @category schemas
346 * @since 4.0.0
347 */
348export const PartialArray: Schema.mutable<
349 Schema.$Array<Schema.Codec<AckChunk | Interrupt | PartialRequest, Encoded>>
350> = Schema.mutable(Schema.Array(PartialJson))
351
352/**
353 * Helper types associated with request envelopes.
354 *
355 * @stability unstable
356 * @since 4.0.0
357 */
358export declare namespace Request {
359 /**
360 * Request envelope type for any RPC protocol.
361 *
362 * @stability unstable
363 * @category models
364 * @since 4.0.0
365 */
366 export type Any = Request<any>
367}
368
369/**
370 * Returns `true` when the supplied value is a runtime cluster envelope.
371 *
372 * **Details**
373 *
374 * The check is based on the envelope type identifier.
375 *
376 * @stability unstable
377 * @category guards
378 * @since 4.0.0
379 */
380export const isEnvelope = (u: unknown): u is Envelope<any> => Predicate.hasProperty(u, TypeId)
381
382/**
383 * Constructs a runtime request envelope and attaches the envelope type identifier.
384 *
385 * **Details**
386 *
387 * Tracing fields are included only when a `traceId` is provided.
388 *
389 * @stability unstable
390 * @category constructors
391 * @since 4.0.0
392 */
393export const makeRequest = <Rpc extends Rpc.Any>(
394 options: {
395 readonly requestId: Snowflake
396 readonly address: EntityAddress
397 readonly tag: Rpc.Tag<Rpc>
398 readonly payload: Rpc.Payload<Rpc>
399 readonly headers: Headers.Headers
400 readonly traceId?: string | undefined
401 readonly spanId?: string | undefined
402 readonly sampled?: boolean | undefined
403 }
404): Request<Rpc> => ({
405 [TypeId]: TypeId,
406 _tag: "Request",
407 requestId: options.requestId,
408 tag: options.tag,
409 address: options.address,
410 payload: options.payload,
411 headers: options.headers,
412 ...(options.traceId !== undefined ?
413 {
414 traceId: options.traceId!,
415 spanId: options.spanId!,
416 sampled: options.sampled!
417 } :
418 {})
419})
420
421/**
422 * Schema for runtime cluster envelopes recognized by their type identifier.
423 *
424 * @stability unstable
425 * @category serialization
426 * @since 4.0.0
427 */
428export const Envelope = Schema.declare(isEnvelope, {
429 identifier: "Envelope"
430})
431
432/**
433 * Schema for runtime request envelopes.
434 *
435 * @stability unstable
436 * @category serialization
437 * @since 4.0.0
438 */
439export const Request = Schema.declare(
440 (u): u is Request.Any => isEnvelope(u) && u._tag === "Request",
441 { identifier: "Request" }
442)
443
444/**
445 * Transforms plain request data with `makeRequest` and encodes
446 * request envelopes back to their raw representation.
447 *
448 * @stability unstable
449 * @category serialization
450 * @since 4.0.0
451 */
452export const RequestTransform: SchemaTransformation.Transformation<
453 Request.Any,
454 any
455> = SchemaTransformation.transform({
456 decode: (u: any) => makeRequest(u),
457 encode: (u) => u as any
458})
459
460/**
461 * Returns the storage primary key for a request envelope whose payload has a
462 * primary key, or `null` when the envelope is not a keyed request.
463 *
464 * @stability unstable
465 * @category getters
466 * @since 4.0.0
467 */
468export const primaryKey = <R extends Rpc.Any>(envelope: Envelope<R>): string | null => {
469 if (envelope._tag !== "Request" || !PrimaryKey.isPrimaryKey(envelope.payload)) {
470 return null
471 }
472 return primaryKeyByAddress({
473 address: envelope.address,
474 tag: envelope.tag,
475 id: PrimaryKey.value(envelope.payload)
476 })
477}
478
479/**
480 * Builds a storage primary-key string from an entity address, RPC tag, and
481 * payload primary-key ID.
482 *
483 * @stability unstable
484 * @category constructors
485 * @since 4.0.0
486 */
487export const primaryKeyByAddress = (options: {
488 readonly address: EntityAddress
489 readonly tag: string
490 readonly id: string
491}): string =>
492 // storage drivers with fixed-width key columns (e.g. SqlMessageStorage)
493 // hash this composed key at their own boundary
494 `${options.address.entityType}/${options.address.entityId}/${options.tag}/${options.id}`