1/**
2 * Stores Effect Cluster messages and replies behind a pluggable backend.
3 *
4 * `MessageStorage` is the boundary between cluster runner logic and the storage
5 * system that keeps mailbox state recoverable. It saves requests, control
6 * envelopes, and replies; finds unprocessed messages for assigned shards;
7 * tracks duplicate requests; and manages reply handlers waiting for responses.
8 * This module also includes the encoded storage-driver contract and no-op or
9 * in-memory implementations for local use and tests.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Arr from "../Array.ts"
15import { Clock } from "../Clock.ts"
16import * as Context from "../Context.ts"
17import * as Data from "../Data.ts"
18import * as Effect from "../Effect.ts"
19import * as Exit from "../Exit.ts"
20import { constFalse, identity } from "../Function.ts"
21import * as Latch from "../Latch.ts"
22import * as Layer from "../Layer.ts"
23import * as Option from "../Option.ts"
24import type { Predicate } from "../Predicate.ts"
25import type * as Rpc from "../rpc/Rpc.ts"
26import type * as RpcSerialization from "../rpc/RpcSerialization.ts"
27import * as Schema from "../Schema.ts"
28import { EntityNotAssignedToRunner, MalformedMessage, type PersistenceError } from "./ClusterError.ts"
29import * as DeliverAt from "./DeliverAt.ts"
30import type { EntityAddress } from "./EntityAddress.ts"
31import * as Envelope from "./Envelope.ts"
32import * as ClusterAbandon from "./internal/clusterAbandon.ts"
33import * as Message from "./Message.ts"
34import * as Reply from "./Reply.ts"
35import * as ShardId from "./ShardId.ts"
36import type { ShardingConfig } from "./ShardingConfig.ts"
37import * as Snowflake from "./Snowflake.ts"
38
39const codecForJson = Schema.toCodecJson as RpcSerialization.CodecFor
40
41/**
42 * Service for cluster mailbox persistence and reply delivery.
43 *
44 * **Details**
45 *
46 * It stores outgoing requests, control envelopes, and replies; reads unprocessed
47 * messages; manages reply handlers; and provides transaction wrapping for storage
48 * operations.
49 *
50 * @stability unstable
51 * @category services
52 * @since 4.0.0
53 */
54export class MessageStorage extends Context.Service<MessageStorage, {
55 /**
56 * Save the provided message and its associated metadata.
57 */
58 readonly saveRequest: <R extends Rpc.Any>(
59 envelope: Message.OutgoingRequest<R>
60 ) => Effect.Effect<SaveResult<R>, PersistenceError | MalformedMessage>
61
62 /**
63 * Save the provided message and its associated metadata.
64 */
65 readonly saveEnvelope: (
66 envelope: Message.OutgoingEnvelope
67 ) => Effect.Effect<void, PersistenceError | MalformedMessage>
68
69 /**
70 * Save the provided `Reply` and its associated metadata.
71 */
72 readonly saveReply: <R extends Rpc.Any>(
73 reply: Reply.ReplyWithContext<R>
74 ) => Effect.Effect<void, PersistenceError | MalformedMessage>
75
76 /**
77 * Clear replies for the given request id. When options.expectedReplyId is
78 * provided, clear only if it is still the latest reply at the storage boundary.
79 * Without it, clear unconditionally.
80 * Custom storage implementations must honor this condition to protect
81 * workflows from stale concurrent resumes.
82 */
83 readonly clearReplies: (
84 requestId: Snowflake.Snowflake,
85 options?: { readonly expectedReplyId?: Snowflake.Snowflake | undefined }
86 ) => Effect.Effect<void, PersistenceError>
87
88 /**
89 * Retrieves the replies for the specified requests.
90 *
91 * **Details**
92 *
93 * This returns:
94 *
95 * - Un-acknowledged chunk replies
96 * - `WithExit` replies
97 */
98 readonly repliesFor: <R extends Rpc.Any>(
99 requests: Iterable<Message.OutgoingRequest<R>>
100 ) => Effect.Effect<Array<Reply.Reply<R>>, PersistenceError | MalformedMessage>
101
102 /**
103 * Retrieves the encoded replies for the specified request ids.
104 */
105 readonly repliesForUnfiltered: (
106 requestIds: Iterable<Snowflake.Snowflake>
107 ) => Effect.Effect<Array<Reply.Encoded>, PersistenceError | MalformedMessage>
108
109 /**
110 * Retrieves the request id for the specified primary key.
111 */
112 readonly requestIdForPrimaryKey: (
113 options: {
114 readonly address: EntityAddress
115 readonly tag: string
116 readonly id: string
117 }
118 ) => Effect.Effect<Option.Option<Snowflake.Snowflake>, PersistenceError>
119
120 /**
121 * For locally sent messages, register a handler to process the replies.
122 */
123 readonly registerReplyHandler: <R extends Rpc.Any>(
124 message: Message.OutgoingRequest<R> | Message.IncomingRequest<R>
125 ) => Effect.Effect<void, EntityNotAssignedToRunner>
126
127 /**
128 * Unregister the reply handler for the specified message.
129 */
130 readonly unregisterReplyHandler: (requestId: Snowflake.Snowflake) => Effect.Effect<void>
131
132 /**
133 * Unregister the reply handlers for the specified ShardId.
134 *
135 * By default the waiters are resumed with `EntityNotAssignedToRunner`, so
136 * they can re-route the wait. With `interrupt: true` the waiters are
137 * interrupted with a cluster-abandon annotation instead, for when the runner
138 * is shutting down and the reply can never be observed here.
139 */
140 readonly unregisterShardReplyHandlers: (
141 shardId: ShardId.ShardId,
142 options?: { readonly interrupt?: boolean | undefined }
143 ) => Effect.Effect<void>
144
145 /**
146 * Retrieves the unprocessed messages for the specified shards.
147 *
148 * **Details**
149 *
150 * A message is unprocessed when:
151 *
152 * - Requests that have no `WithExit` replies or no unacknowledged chunk replies
153 * - The latest `AckChunk` envelope
154 * - All `Interrupt` envelopes for unprocessed requests
155 *
156 * The `limit` option bounds the number of messages returned in a single
157 * read, and the `addresses` option restricts the read to the provided
158 * entity addresses. Only the returned messages are claimed; other messages
159 * stay eligible for later reads.
160 */
161 readonly unprocessedMessages: (
162 shardIds: Iterable<ShardId.ShardId>,
163 options?: {
164 readonly limit?: number | undefined
165 readonly addresses?: ReadonlyArray<EntityAddress> | undefined
166 } | undefined
167 ) => Effect.Effect<Array<Message.Incoming<any>>, PersistenceError>
168
169 /**
170 * Retrieves the unprocessed messages by id.
171 */
172 readonly unprocessedMessagesById: <R extends Rpc.Any>(
173 messageIds: Iterable<Snowflake.Snowflake>
174 ) => Effect.Effect<Array<Message.Incoming<R>>, PersistenceError>
175
176 /**
177 * Reset the mailbox state for the provided shards.
178 */
179 readonly resetShards: (
180 shardIds: Iterable<ShardId.ShardId>
181 ) => Effect.Effect<void, PersistenceError>
182
183 /**
184 * Reset the mailbox state for the provided address.
185 */
186 readonly resetAddress: (
187 address: EntityAddress
188 ) => Effect.Effect<void, PersistenceError>
189
190 /** Release request claims without changing replies or processed state. */
191 readonly resetRequests: (
192 requestIds: ReadonlyArray<Snowflake.Snowflake>
193 ) => Effect.Effect<void, PersistenceError>
194
195 /**
196 * Reset the mailbox state for the provided addresses.
197 */
198 readonly resetAddresses: (
199 addresses: ReadonlyArray<EntityAddress>
200 ) => Effect.Effect<void, PersistenceError>
201
202 /**
203 * Clear all messages and replies for the provided address.
204 */
205 readonly clearAddress: (
206 address: EntityAddress
207 ) => Effect.Effect<void, PersistenceError>
208
209 /**
210 * Used to wrap requests with transactions.
211 */
212 readonly withTransaction: <A, E, R>(
213 effect: Effect.Effect<A, E, R>
214 ) => Effect.Effect<A, E, R>
215}>()("effect/cluster/MessageStorage") {}
216
217/**
218 * Result of saving a request or envelope into message storage.
219 *
220 * **Details**
221 *
222 * A duplicate result carries the original request ID and the last reply already
223 * received for the duplicated request.
224 *
225 * @stability unstable
226 * @category models
227 * @since 4.0.0
228 */
229export type SaveResult<R extends Rpc.Any> = SaveResult.Success | SaveResult.Duplicate<R>
230
231/**
232 * Constructors and matchers for decoded save results.
233 *
234 * @stability unstable
235 * @category constructors
236 * @since 4.0.0
237 */
238export const SaveResult = Data.taggedEnum<SaveResult.Constructor>()
239
240/**
241 * Constructors and matchers for encoded save results returned by storage
242 * drivers.
243 *
244 * @stability unstable
245 * @category constructors
246 * @since 4.0.0
247 */
248export const SaveResultEncoded = Data.taggedEnum<SaveResult.Encoded>()
249
250/**
251 * Variants and helper types for `SaveResult`.
252 *
253 * @stability unstable
254 * @since 4.0.0
255 */
256export declare namespace SaveResult {
257 /**
258 * Encoded storage-driver form of `SaveResult`.
259 *
260 * **Details**
261 *
262 * Duplicate results contain an encoded last received reply instead of a decoded
263 * reply.
264 *
265 * @stability unstable
266 * @category models
267 * @since 4.0.0
268 */
269 export type Encoded = SaveResult.Success | SaveResult.DuplicateEncoded
270
271 /**
272 * Variant indicating that the message was saved as a new storage entry.
273 *
274 * @stability unstable
275 * @category models
276 * @since 4.0.0
277 */
278 export interface Success {
279 readonly _tag: "Success"
280 }
281
282 /**
283 * Variant indicating that the request duplicates an existing stored request.
284 *
285 * **Details**
286 *
287 * It carries the original request ID and the latest decoded reply, when one is
288 * available.
289 *
290 * @stability unstable
291 * @category models
292 * @since 4.0.0
293 */
294 export interface Duplicate<R extends Rpc.Any> {
295 readonly _tag: "Duplicate"
296 readonly originalId: Snowflake.Snowflake
297 readonly lastReceivedReply: Option.Option<Reply.Reply<R>>
298 }
299
300 /**
301 * Encoded duplicate-save variant returned by lower-level storage drivers.
302 *
303 * **Details**
304 *
305 * It carries the original request ID and the latest encoded reply, when one is
306 * available.
307 *
308 * @stability unstable
309 * @category models
310 * @since 4.0.0
311 */
312 export interface DuplicateEncoded {
313 readonly _tag: "Duplicate"
314 readonly originalId: Snowflake.Snowflake
315 readonly lastReceivedReply: Option.Option<Reply.Encoded>
316 }
317
318 /**
319 * Generic tagged enum constructor type for `SaveResult`.
320 *
321 * @stability unstable
322 * @category utility types
323 * @since 4.0.0
324 */
325 export interface Constructor extends Data.TaggedEnum.WithGenerics<1> {
326 readonly taggedEnum: SaveResult<this["A"] extends Rpc.Any ? this["A"] : never>
327 }
328}
329
330/**
331 * Low-level storage-driver contract for encoded envelopes and replies.
332 *
333 * **Details**
334 *
335 * Implementations persist encoded messages, track primary keys and delayed
336 * delivery, read unprocessed messages, and provide transaction wrapping.
337 *
338 * @stability unstable
339 * @category services
340 * @since 4.0.0
341 */
342export type Encoded = {
343 /**
344 * Save the provided message and its associated metadata.
345 */
346 readonly saveEnvelope: (
347 options: {
348 readonly envelope: Envelope.Encoded
349 readonly primaryKey: string | null
350 readonly deliverAt: number | null
351 }
352 ) => Effect.Effect<SaveResult.Encoded, PersistenceError>
353
354 /**
355 * Save the provided `Reply` and its associated metadata.
356 */
357 readonly saveReply: (reply: Reply.Encoded) => Effect.Effect<void, PersistenceError>
358
359 /**
360 * Remove the replies for the specified request. If options.expectedReplyId
361 * is provided, compare it with the latest reply atomically before clearing.
362 * Without it, clear unconditionally.
363 * Custom storage implementations must honor this condition to protect
364 * workflows from stale concurrent resumes.
365 */
366 readonly clearReplies: (
367 requestId: Snowflake.Snowflake,
368 options?: { readonly expectedReplyId?: Snowflake.Snowflake | undefined }
369 ) => Effect.Effect<void, PersistenceError>
370
371 /**
372 * Retrieves the request id for the specified primary key.
373 */
374 readonly requestIdForPrimaryKey: (
375 primaryKey: string
376 ) => Effect.Effect<Option.Option<Snowflake.Snowflake>, PersistenceError>
377
378 /**
379 * Retrieves the replies for the specified requests.
380 *
381 * **Details**
382 *
383 * This returns:
384 *
385 * - Un-acknowledged chunk replies
386 * - `WithExit` replies
387 */
388 readonly repliesFor: (requestIds: Arr.NonEmptyArray<string>) => Effect.Effect<
389 Array<Reply.Encoded>,
390 PersistenceError
391 >
392
393 /**
394 * Retrieves the replies for the specified request ids.
395 */
396 readonly repliesForUnfiltered: (requestIds: Arr.NonEmptyArray<string>) => Effect.Effect<
397 Array<Reply.Encoded>,
398 PersistenceError
399 >
400
401 /**
402 * Retrieves the unprocessed messages for the given shards.
403 *
404 * **Details**
405 *
406 * A message is unprocessed when:
407 *
408 * - Requests that have no `WithExit` replies or no unacknowledged chunk replies
409 * - The latest `AckChunk` envelope
410 * - All `Interrupt` envelopes for unprocessed requests
411 *
412 * The `limit` option bounds the number of rows returned by a single read,
413 * and the `addresses` option restricts the read to the provided entity
414 * addresses. Only the returned rows are claimed; other rows stay eligible
415 * for later reads.
416 */
417 readonly unprocessedMessages: (
418 shardIds: Arr.NonEmptyArray<string>,
419 now: number,
420 options?: {
421 readonly limit?: number | undefined
422 readonly addresses?: ReadonlyArray<EntityAddress> | undefined
423 } | undefined
424 ) => Effect.Effect<
425 Array<{
426 readonly envelope: Envelope.Encoded
427 readonly lastSentReply: Option.Option<Reply.Encoded>
428 }>,
429 PersistenceError
430 >
431
432 /**
433 * Retrieves the unprocessed messages by id.
434 */
435 readonly unprocessedMessagesById: (
436 messageIds: Arr.NonEmptyArray<Snowflake.Snowflake>,
437 now: number
438 ) => Effect.Effect<
439 Array<{
440 readonly envelope: Envelope.Encoded
441 readonly lastSentReply: Option.Option<Reply.Encoded>
442 }>,
443 PersistenceError
444 >
445
446 /** Release request claims without changing replies or processed state. */
447 readonly resetRequests: (
448 requestIds: ReadonlyArray<Snowflake.Snowflake>
449 ) => Effect.Effect<void, PersistenceError>
450
451 /**
452 * Reset the mailbox state for the provided addresses.
453 */
454 readonly resetAddresses: (
455 addresses: ReadonlyArray<EntityAddress>
456 ) => Effect.Effect<void, PersistenceError>
457
458 /**
459 * Clear all messages and replies for the provided address.
460 */
461 readonly clearAddress: (
462 address: EntityAddress
463 ) => Effect.Effect<void, PersistenceError>
464
465 /**
466 * Reset the mailbox state for the provided shards.
467 */
468 readonly resetShards: (
469 shardIds: Arr.NonEmptyArray<string>
470 ) => Effect.Effect<void, PersistenceError>
471
472 /**
473 * Used to wrap requests with transactions.
474 */
475 readonly withTransaction: <A, E, R>(
476 effect: Effect.Effect<A, E, R>
477 ) => Effect.Effect<A, E, R>
478}
479
480/**
481 * Cursor options for reading encoded unprocessed messages across shard sets.
482 *
483 * **Details**
484 *
485 * The fields distinguish existing shards from newly assigned shards and carry the
486 * driver-specific pagination cursor.
487 *
488 * @stability unstable
489 * @category models
490 * @since 4.0.0
491 */
492export type EncodedUnprocessedOptions<A> = {
493 readonly existingShards: Array<number>
494 readonly newShards: Array<number>
495 readonly cursor: Option.Option<A>
496}
497
498/**
499 * Cursor options for reading encoded replies across request sets.
500 *
501 * **Details**
502 *
503 * The fields distinguish existing requests from new requests and carry the
504 * driver-specific pagination cursor.
505 *
506 * @stability unstable
507 * @category models
508 * @since 4.0.0
509 */
510export type EncodedRepliesOptions<A> = {
511 readonly existingRequests: Array<string>
512 readonly newRequests: Array<string>
513 readonly cursor: Option.Option<A>
514}
515
516/**
517 * Wraps a concrete message storage implementation with reply-handler management.
518 *
519 * **Details**
520 *
521 * The returned service can register waiting reply handlers, notify them when
522 * replies are saved, and fail them when a request or shard is unregistered.
523 *
524 * @stability unstable
525 * @category constructors
526 * @since 4.0.0
527 */
528export const make = (
529 storage:
530 & Omit<
531 MessageStorage["Service"],
532 "saveReply" | "registerReplyHandler" | "unregisterReplyHandler" | "unregisterShardReplyHandlers"
533 >
534 & {
535 /**
536 * Save the provided `Reply`, returning the reply as it was persisted.
537 *
538 * Implementations that persist a different representation, such as a
539 * defect reply when the original cannot be encoded, return that fallback
540 * so reply handlers deliver exactly what storage recorded.
541 */
542 readonly saveReply: <R extends Rpc.Any>(
543 reply: Reply.ReplyWithContext<R>
544 ) => Effect.Effect<Reply.ReplyWithContext<R>, PersistenceError | MalformedMessage>
545 }
546): Effect.Effect<MessageStorage["Service"]> =>
547 Effect.sync(() => {
548 type ReplyHandler = {
549 readonly message: Message.OutgoingRequest<any> | Message.IncomingRequest<any>
550 readonly shardSet: Set<ReplyHandler>
551 readonly respond: (reply: Reply.ReplyWithContext<any>) => Effect.Effect<void, PersistenceError | MalformedMessage>
552 readonly resume: (effect: Effect.Effect<void, EntityNotAssignedToRunner>) => void
553 }
554 const replyHandlers = new Map<Snowflake.Snowflake, Array<ReplyHandler>>()
555 const replyHandlersShard = new Map<string, Set<ReplyHandler>>()
556 return MessageStorage.of({
557 ...storage,
558 registerReplyHandler: (message) => {
559 const requestId = message.envelope.requestId
560 return Effect.callback<void, EntityNotAssignedToRunner>((resume) => {
561 const shardId = message.envelope.address.shardId.toString()
562 let handlers = replyHandlers.get(requestId)
563 if (handlers === undefined) {
564 handlers = []
565 replyHandlers.set(requestId, handlers)
566 }
567 let shardSet = replyHandlersShard.get(shardId)
568 if (!shardSet) {
569 shardSet = new Set()
570 replyHandlersShard.set(shardId, shardSet)
571 }
572 const entry: ReplyHandler = {
573 message,
574 shardSet,
575 respond: message._tag === "IncomingRequest" ? message.respond : (reply) => message.respond(reply.reply),
576 resume
577 }
578 handlers.push(entry)
579 shardSet.add(entry)
580 return Effect.sync(() => {
581 const index = handlers.indexOf(entry)
582 handlers.splice(index, 1)
583 shardSet.delete(entry)
584 })
585 })
586 },
587 unregisterReplyHandler: (requestId) =>
588 Effect.sync(() => {
589 const handlers = replyHandlers.get(requestId)
590 if (!handlers) return
591 replyHandlers.delete(requestId)
592 for (let i = 0; i < handlers.length; i++) {
593 const handler = handlers[i]
594 handler.shardSet.delete(handler)
595 handler.resume(Effect.fail(
596 new EntityNotAssignedToRunner({
597 address: handler.message.envelope.address
598 })
599 ))
600 }
601 }),
602 unregisterShardReplyHandlers: (shardId, options) =>
603 Effect.sync(() => {
604 const id = shardId.toString()
605 const shardSet = replyHandlersShard.get(id)
606 if (!shardSet) return
607 replyHandlersShard.delete(id)
608 shardSet.forEach((handler) => {
609 replyHandlers.delete(handler.message.envelope.requestId)
610 handler.resume(
611 options?.interrupt ? ClusterAbandon.interrupt : Effect.fail(
612 new EntityNotAssignedToRunner({
613 address: handler.message.envelope.address
614 })
615 )
616 )
617 })
618 }),
619 saveReply(reply) {
620 const requestId = reply.reply.requestId
621 return Effect.flatMap(storage.saveReply(reply), (persisted) => {
622 const handlers = replyHandlers.get(requestId)
623 if (!handlers) {
624 return Effect.void
625 } else if (persisted.reply._tag === "WithExit") {
626 replyHandlers.delete(requestId)
627 for (let i = 0; i < handlers.length; i++) {
628 const handler = handlers[i]
629 handler.shardSet.delete(handler)
630 handler.resume(Effect.void)
631 }
632 }
633 return handlers.length === 1
634 ? handlers[0].respond(persisted)
635 : Effect.forEach(handlers, (handler) => handler.respond(persisted))
636 })
637 }
638 })
639 })
640
641/**
642 * Builds a `MessageStorage` service from an encoded storage driver.
643 *
644 * **Details**
645 *
646 * The adapter handles envelope and reply encoding and decoding, primary-key
647 * generation, delayed delivery checks, duplicate decoding, and malformed-message
648 * defect replies.
649 *
650 * @stability unstable
651 * @category constructors
652 * @since 4.0.0
653 */
654export const makeEncoded: (encoded: Encoded) => Effect.Effect<
655 MessageStorage["Service"],
656 never,
657 Snowflake.Generator
658> = Effect.fnUntraced(function*(encoded: Encoded) {
659 const snowflakeGen = yield* Snowflake.Generator
660 const clock = yield* Clock
661
662 const storage: MessageStorage["Service"] = yield* make({
663 saveRequest: (message) =>
664 Message.serializeEnvelope(message).pipe(
665 Effect.flatMap((envelope) =>
666 encoded.saveEnvelope({
667 envelope,
668 primaryKey: Envelope.primaryKey(message.envelope),
669 deliverAt: DeliverAt.toMillis(message.envelope.payload)
670 })
671 ),
672 Effect.flatMap((result) => {
673 if (result._tag === "Success" || Option.isNone(result.lastReceivedReply)) {
674 return Effect.succeed(result as SaveResult<any>)
675 }
676 const duplicate = result
677 const schema = Reply.Reply(message.rpc, codecForJson)
678 return Schema.decodeEffect(schema)(result.lastReceivedReply.value).pipe(
679 Effect.provideContext(message.context),
680 MalformedMessage.refail,
681 Effect.map((reply) =>
682 SaveResult.Duplicate({
683 originalId: duplicate.originalId,
684 lastReceivedReply: Option.some(reply)
685 })
686 )
687 )
688 })
689 ),
690 saveEnvelope: (message) =>
691 Message.serializeEnvelope(message).pipe(
692 Effect.flatMap((envelope) =>
693 encoded.saveEnvelope({
694 envelope,
695 primaryKey: null,
696 deliverAt: null
697 })
698 ),
699 Effect.asVoid
700 ),
701 saveReply: (reply) =>
702 Reply.serialize(reply, codecForJson).pipe(
703 Effect.map((encodedReply) => ({ encodedReply, persisted: reply })),
704 Effect.catchTag("MalformedMessage", (error) => {
705 const persisted = Reply.ReplyWithContext.fromDefect({
706 id: reply.reply.id,
707 requestId: reply.reply.requestId,
708 defect: error
709 })
710 return Effect.map(
711 Effect.orDie(Reply.serialize(persisted, codecForJson)),
712 (encodedReply) => ({ encodedReply, persisted })
713 )
714 }),
715 Effect.flatMap(({ encodedReply, persisted }) => Effect.as(encoded.saveReply(encodedReply), persisted))
716 ),
717 clearReplies: encoded.clearReplies,
718 repliesFor: Effect.fnUntraced(function*(messages) {
719 const requestIds = Arr.empty<string>()
720 const map = new Map<string, Message.OutgoingRequest<any>>()
721 for (const message of messages) {
722 const id = String(message.envelope.requestId)
723 requestIds.push(id)
724 map.set(id, message)
725 }
726 if (!Arr.isArrayNonEmpty(requestIds)) return []
727 const encodedReplies = yield* encoded.repliesFor(requestIds)
728 return yield* decodeReplies(map, encodedReplies)
729 }),
730 repliesForUnfiltered: (ids) => {
731 const arr = Array.from(ids, String)
732 if (!Arr.isArrayNonEmpty(arr)) return Effect.succeed([])
733 return encoded.repliesForUnfiltered(arr)
734 },
735 requestIdForPrimaryKey(options) {
736 const primaryKey = Envelope.primaryKeyByAddress(options)
737 return encoded.requestIdForPrimaryKey(primaryKey)
738 },
739 unprocessedMessages(shardIds, options) {
740 const storage = this as unknown as MessageStorage["Service"]
741 const shards = Array.from(shardIds, (id) => id.toString())
742 if (!Arr.isArrayNonEmpty(shards)) return Effect.succeed([])
743 if (options?.addresses !== undefined && options.addresses.length === 0) return Effect.succeed([])
744 return Effect.flatMap(
745 Effect.suspend(() => encoded.unprocessedMessages(shards, clock.currentTimeMillisUnsafe(), options)),
746 (messages) => decodeMessages(storage, messages)
747 )
748 },
749 unprocessedMessagesById(messageIds) {
750 const storage = this as unknown as MessageStorage["Service"]
751 const ids = Array.from(messageIds)
752 if (!Arr.isArrayNonEmpty(ids)) return Effect.succeed([])
753 return Effect.flatMap(
754 Effect.suspend(() => encoded.unprocessedMessagesById(ids, clock.currentTimeMillisUnsafe())),
755 (messages) => decodeMessages(storage, messages)
756 )
757 },
758 resetAddress: (address) => encoded.resetAddresses([address]),
759 resetRequests: (requestIds) => requestIds.length === 0 ? Effect.void : encoded.resetRequests(requestIds),
760 resetAddresses: (addresses) => addresses.length === 0 ? Effect.void : encoded.resetAddresses(addresses),
761 clearAddress: encoded.clearAddress,
762 resetShards: (shardIds) => {
763 const shards = Array.from(shardIds, (id) => id.toString())
764 if (!Arr.isArrayNonEmpty(shards)) return Effect.void
765 return encoded.resetShards(shards)
766 },
767 withTransaction: encoded.withTransaction
768 })
769
770 const decodeMessages = (
771 storage: MessageStorage["Service"],
772 envelopes: Array<{
773 readonly envelope: Envelope.Encoded
774 readonly lastSentReply: Option.Option<Reply.Encoded>
775 }>
776 ) => {
777 const messages: Array<Message.Incoming<any>> = []
778 let index = 0
779
780 // if we have a malformed message, we should not return it and update
781 // the storage with a defect
782 const decodeMessage = Effect.catch(
783 Effect.suspend(() => {
784 const envelope = envelopes[index]
785 if (!envelope) return Effect.undefined
786 return decodeEnvelopeWithReply(envelope)
787 }),
788 (error) => {
789 const envelope = envelopes[index]
790 return storage.saveReply(Reply.ReplyWithContext.fromDefect({
791 id: snowflakeGen.nextUnsafe(),
792 requestId: Snowflake.Snowflake(envelope.envelope.requestId),
793 defect: error.toString()
794 })).pipe(
795 Effect.forkDetach,
796 Effect.asVoid
797 )
798 }
799 )
800 return Effect.as(
801 Effect.whileLoop({
802 while: () => index < envelopes.length,
803 body: () => decodeMessage,
804 step: (message) => {
805 const envelope = envelopes[index++]
806 if (!message) return
807 messages.push(
808 message.envelope._tag === "Request"
809 ? new Message.IncomingRequest({
810 envelope: message.envelope,
811 lastSentReply: envelope.lastSentReply,
812 respond: storage.saveReply,
813 codecFor: codecForJson
814 })
815 : new Message.IncomingEnvelope({
816 envelope: message.envelope
817 })
818 )
819 }
820 }),
821 messages
822 )
823 }
824
825 const decodeReplies = (
826 messages: Map<string, Message.OutgoingRequest<any>>,
827 encodedReplies: Array<Reply.Encoded>
828 ) => {
829 const replies: Array<Reply.Reply<any>> = []
830 const ignoredRequests = new Set<string>()
831 let index = 0
832
833 const decodeReply: Effect.Effect<void | Reply.Reply<any>> = Effect.catch(
834 Effect.suspend(() => {
835 const reply = encodedReplies[index]
836 if (ignoredRequests.has(reply.requestId)) return Effect.void
837 const message = messages.get(reply.requestId)
838 if (!message) return Effect.void
839 const schema = Reply.Reply(message.rpc, codecForJson)
840 return Schema.decodeEffect(schema)(reply).pipe(
841 Effect.provideContext(message.context)
842 ) as Effect.Effect<Reply.Reply<any>, Schema.SchemaError>
843 }),
844 (error) => {
845 const reply = encodedReplies[index]
846 ignoredRequests.add(reply.requestId)
847 return Effect.succeed(
848 new Reply.WithExit({
849 id: snowflakeGen.nextUnsafe(),
850 requestId: Snowflake.Snowflake(reply.requestId),
851 exit: Exit.die(error)
852 })
853 )
854 }
855 )
856
857 return Effect.as(
858 Effect.whileLoop({
859 while: () => index < encodedReplies.length,
860 body: () => decodeReply,
861 step: (reply) => {
862 index++
863 if (reply) replies.push(reply)
864 }
865 }),
866 replies
867 )
868 }
869
870 return storage
871})
872
873/**
874 * No-op `MessageStorage` service that does not persist messages or replies.
875 *
876 * @stability unstable
877 * @category constructors
878 * @since 4.0.0
879 */
880export const noop: MessageStorage["Service"] = Effect.runSync(make({
881 saveRequest: () => Effect.succeed(SaveResult.Success()),
882 saveEnvelope: () => Effect.void,
883 saveReply: (reply) => Effect.succeed(reply),
884 clearReplies: () => Effect.void,
885 repliesFor: () => Effect.succeed([]),
886 repliesForUnfiltered: () => Effect.succeed([]),
887 requestIdForPrimaryKey: () => Effect.succeedNone,
888 unprocessedMessages: () => Effect.succeed([]),
889 unprocessedMessagesById: () => Effect.succeed([]),
890 resetAddress: () => Effect.void,
891 resetRequests: () => Effect.void,
892 resetAddresses: () => Effect.void,
893 clearAddress: () => Effect.void,
894 resetShards: () => Effect.void,
895 withTransaction: identity
896}))
897
898/**
899 * In-memory storage entry for a request envelope.
900 *
901 * **Details**
902 *
903 * It stores the encoded envelope, last acknowledged chunk, accumulated replies,
904 * and optional delivery time.
905 *
906 * @stability unstable
907 * @category models
908 * @since 4.0.0
909 */
910export type MemoryEntry = {
911 readonly envelope: Envelope.Encoded
912 lastReceivedChunk: Reply.ChunkEncoded | undefined
913 replies: Array<Reply.Encoded>
914 deliverAt: number | null
915}
916
917/**
918 * Provides a context reference used in tests to simulate a transaction.
919 *
920 * @stability unstable
921 * @category services
922 * @since 4.0.0
923 */
924export const MemoryTransaction = Context.Reference<boolean>("effect/cluster/MessageStorage/MemoryTransaction", {
925 defaultValue: constFalse
926})
927
928// the same claim window the SQL driver uses (`last_read < ten minutes ago`)
929const claimExpirationMillis = 10 * 60 * 1000
930
931/**
932 * Service that provides an in-memory message storage driver with inspectable backing state.
933 *
934 * **Details**
935 *
936 * It provides a `MessageStorage` service, the encoded driver implementation, and
937 * maps used to track requests, primary keys, unprocessed envelopes, reply IDs,
938 * and the journal.
939 *
940 * @stability unstable
941 * @category services
942 * @since 4.0.0
943 */
944export class MemoryDriver extends Context.Service<MemoryDriver>()("effect/cluster/MessageStorage/MemoryDriver", {
945 make: Effect.gen(function*() {
946 const clock = yield* Clock
947 const requests = new Map<string, MemoryEntry>()
948 const requestsByPrimaryKey = new Map<string, MemoryEntry>()
949 const unprocessed = new Set<Envelope.Encoded>()
950 const replyIds = new Set<string>()
951 const lastRead = new Map<Envelope.Encoded, number>()
952
953 const journal: Array<Envelope.Encoded> = []
954
955 const addressKey = (address: Envelope.Encoded["address"] | EntityAddress) =>
956 `${address.shardId.group}/${address.shardId.id}/${address.entityType}/${address.entityId}`
957
958 const resetAddresses = (addresses: ReadonlyArray<EntityAddress>) =>
959 addresses.length === 0
960 ? Effect.void
961 : Effect.sync(() => {
962 const keys = new Set(addresses.map(addressKey))
963 for (const envelope of journal) {
964 if (keys.has(addressKey(envelope.address))) {
965 lastRead.delete(envelope)
966 }
967 }
968 })
969
970 const cursors = new WeakMap<{}, number>()
971
972 const unprocessedWith = (predicate: Predicate<Envelope.Encoded>) => {
973 const messages: Array<{
974 readonly envelope: Envelope.Encoded
975 readonly lastSentReply: Option.Option<Reply.Encoded>
976 }> = []
977 const now = clock.currentTimeMillisUnsafe()
978 for (const envelope of unprocessed) {
979 if (!predicate(envelope)) {
980 continue
981 }
982 if (envelope._tag === "Request") {
983 const entry = requests.get(envelope.requestId)
984 if (entry?.deliverAt && entry.deliverAt > now) {
985 continue
986 }
987 messages.push({
988 envelope,
989 lastSentReply: Option.fromNullishOr(entry?.replies[entry.replies.length - 1])
990 })
991 } else {
992 messages.push({
993 envelope,
994 lastSentReply: Option.none()
995 })
996 }
997 }
998 return messages
999 }
1000
1001 const replyLatch = yield* Latch.make()
1002
1003 function repliesFor(requestIds: Array<string>) {
1004 const replies = Arr.empty<Reply.Encoded>()
1005 for (const requestId of requestIds) {
1006 const request = requests.get(requestId)
1007 if (!request) continue
1008 else if (request.lastReceivedChunk === undefined) {
1009 replies.push(...request.replies)
1010 continue
1011 }
1012 const sequence = request.lastReceivedChunk.sequence
1013 for (const reply of request.replies) {
1014 if (reply._tag === "Chunk" && reply.sequence <= sequence) {
1015 continue
1016 }
1017 replies.push(reply)
1018 }
1019 }
1020 return replies
1021 }
1022
1023 const encoded: Encoded = {
1024 saveEnvelope: ({ deliverAt, envelope: envelope_, primaryKey }) =>
1025 Effect.sync(() => {
1026 const envelope = JSON.parse(JSON.stringify(envelope_)) as Envelope.Encoded
1027 const existing = primaryKey
1028 ? requestsByPrimaryKey.get(primaryKey)
1029 : envelope._tag === "Request" && requests.get(envelope.requestId)
1030 if (existing) {
1031 return SaveResultEncoded.Duplicate({
1032 originalId: Snowflake.Snowflake(existing.envelope.requestId),
1033 lastReceivedReply: Option.fromNullishOr(
1034 existing.replies.length === 1 && existing.replies[0]._tag === "WithExit"
1035 ? existing.replies[0]
1036 : existing.lastReceivedChunk
1037 )
1038 })
1039 }
1040 if (envelope._tag === "Request") {
1041 const entry: MemoryEntry = { envelope, replies: [], lastReceivedChunk: undefined, deliverAt }
1042 requests.set(envelope.requestId, entry)
1043 if (primaryKey) {
1044 requestsByPrimaryKey.set(primaryKey, entry)
1045 }
1046 } else if (envelope._tag === "AckChunk") {
1047 const entry = requests.get(envelope.requestId)
1048 if (entry) {
1049 entry.lastReceivedChunk = entry.replies.find((r): r is Reply.ChunkEncoded =>
1050 r._tag === "Chunk" && r.id === envelope.replyId
1051 ) ?? entry.lastReceivedChunk
1052 }
1053 }
1054 unprocessed.add(envelope)
1055 journal.push(envelope)
1056 return SaveResultEncoded.Success()
1057 }),
1058 saveReply: (reply_) =>
1059 Effect.sync(() => {
1060 const reply = JSON.parse(JSON.stringify(reply_)) as Reply.Encoded
1061 const entry = requests.get(reply.requestId)
1062 if (!entry || replyIds.has(reply.id)) return
1063 if (reply._tag === "WithExit") {
1064 unprocessed.delete(entry.envelope)
1065 lastRead.delete(entry.envelope)
1066 }
1067 entry.replies.push(reply)
1068 replyIds.add(reply.id)
1069 replyLatch.openUnsafe()
1070 }),
1071 clearReplies: (id, options) =>
1072 Effect.sync(() => {
1073 const entry = requests.get(String(id))
1074 if (!entry) return
1075 if (
1076 options?.expectedReplyId !== undefined && entry.replies.at(-1)?.id !== String(options.expectedReplyId)
1077 ) return
1078 entry.replies = []
1079 entry.lastReceivedChunk = undefined
1080 unprocessed.add(entry.envelope)
1081 lastRead.delete(entry.envelope)
1082 }),
1083 requestIdForPrimaryKey: (primaryKey) =>
1084 Effect.sync(() => {
1085 const entry = requestsByPrimaryKey.get(primaryKey)
1086 return Option.map(Option.fromNullishOr(entry?.envelope.requestId), Snowflake.Snowflake)
1087 }),
1088 repliesFor: (requestIds) => Effect.sync(() => repliesFor(requestIds)),
1089 repliesForUnfiltered: (requestIds) =>
1090 Effect.sync(() => requestIds.flatMap((id) => requests.get(String(id))?.replies ?? [])),
1091 unprocessedMessages: (shardIds, now, options) =>
1092 options?.addresses?.length === 0 ? Effect.succeed([]) : Effect.sync(() => {
1093 if (unprocessed.size === 0) return []
1094 const limit = options?.limit ?? Infinity
1095 const addressFilter = options?.addresses && new Set(options.addresses.map(addressKey))
1096 const messages = Arr.empty<{
1097 envelope: Envelope.Encoded
1098 lastSentReply: Option.Option<Reply.Encoded>
1099 }>()
1100 for (let index = 0; index < journal.length; index++) {
1101 if (messages.length >= limit) break
1102 const envelope = journal[index]
1103 const shardId = ShardId.make(envelope.address.shardId.group, envelope.address.shardId.id)
1104 if (!unprocessed.has(envelope as any) || !shardIds.includes(shardId.toString())) {
1105 continue
1106 }
1107 if (addressFilter && !addressFilter.has(addressKey(envelope.address))) {
1108 continue
1109 }
1110 if (envelope._tag === "Request") {
1111 const entry = requests.get(envelope.requestId)!
1112 if (entry.deliverAt && entry.deliverAt > now) {
1113 continue
1114 }
1115 const claimedAt = lastRead.get(envelope)
1116 if (claimedAt !== undefined && claimedAt > now - claimExpirationMillis) {
1117 continue
1118 }
1119 // Memory storage models claim expiry, but not SQL reply filtering or transactions.
1120 messages.push({
1121 envelope,
1122 lastSentReply: Option.fromNullishOr(entry.replies[entry.replies.length - 1])
1123 })
1124 lastRead.set(envelope, now)
1125 } else {
1126 messages.push({
1127 envelope,
1128 lastSentReply: Option.none()
1129 })
1130 unprocessed.delete(envelope)
1131 }
1132 }
1133 return messages
1134 }),
1135 unprocessedMessagesById: (ids) =>
1136 Effect.sync(() => {
1137 const envelopeIds = new Set<string>()
1138 for (const id of ids) {
1139 envelopeIds.add(String(id))
1140 }
1141 return unprocessedWith((envelope) => envelopeIds.has(envelope.requestId))
1142 }),
1143 resetAddresses,
1144 resetRequests: (requestIds) =>
1145 Effect.sync(() => {
1146 for (const id of requestIds) {
1147 const entry = requests.get(String(id))
1148 if (entry) lastRead.delete(entry.envelope)
1149 }
1150 }),
1151 clearAddress: (address) =>
1152 Effect.sync(() => {
1153 for (const [primaryKey, entry] of requestsByPrimaryKey) {
1154 const envelope = entry.envelope
1155 const sameAddress = address.entityType === envelope.address.entityType &&
1156 address.entityId === envelope.address.entityId
1157 if (sameAddress) {
1158 requestsByPrimaryKey.delete(primaryKey)
1159 }
1160 }
1161 for (let i = journal.length - 1; i >= 0; i--) {
1162 const envelope = journal[i]
1163 const sameAddress = address.entityType === envelope.address.entityType &&
1164 address.entityId === envelope.address.entityId
1165 if (!sameAddress) {
1166 continue
1167 }
1168 unprocessed.delete(envelope)
1169 lastRead.delete(envelope)
1170 if (envelope._tag === "Request") {
1171 requests.delete(envelope.requestId)
1172 }
1173 journal.splice(i, 1)
1174 }
1175 }),
1176 resetShards: (shardIds) =>
1177 Effect.sync(() => {
1178 const shards = new Set(shardIds)
1179 for (const envelope of journal) {
1180 const shardId = ShardId.make(envelope.address.shardId.group, envelope.address.shardId.id)
1181 if (shards.has(shardId.toString())) {
1182 lastRead.delete(envelope)
1183 }
1184 }
1185 }),
1186 withTransaction: Effect.provideService(MemoryTransaction, true)
1187 }
1188
1189 const storage = yield* makeEncoded(encoded)
1190
1191 return {
1192 storage,
1193 encoded,
1194 requests,
1195 requestsByPrimaryKey,
1196 unprocessed,
1197 replyIds,
1198 journal,
1199 cursors
1200 } as const
1201 })
1202}) {
1203 /**
1204 * Layer that provides the in-memory message storage driver.
1205 *
1206 * @stability unstable
1207 * @since 4.0.0
1208 */
1209 static readonly layer: Layer.Layer<MemoryDriver> = Layer.effect(this)(this.make).pipe(
1210 Layer.provide(Snowflake.layerGenerator)
1211 )
1212}
1213
1214/**
1215 * Layer that provides the no-op `MessageStorage` service.
1216 *
1217 * @stability unstable
1218 * @category layers
1219 * @since 4.0.0
1220 */
1221export const layerNoop: Layer.Layer<MessageStorage> = Layer.succeed(MessageStorage, noop)
1222
1223/**
1224 * Layer that provides in-memory message storage and its backing `MemoryDriver`.
1225 *
1226 * @stability unstable
1227 * @category layers
1228 * @since 4.0.0
1229 */
1230export const layerMemory: Layer.Layer<
1231 MessageStorage | MemoryDriver,
1232 never,
1233 ShardingConfig
1234> = Layer.effect(MessageStorage, Effect.map(MemoryDriver, (_) => _.storage)).pipe(
1235 Layer.provideMerge(MemoryDriver.layer)
1236)
1237
1238// --- internal ---
1239
1240const EnvelopeWithReply: Schema.Struct<
1241 {
1242 readonly envelope: Schema.ConstraintDecoder<Envelope.PartialRequest | Envelope.AckChunk | Envelope.Interrupt>
1243 readonly lastSentReply: Schema.Option<Schema.Codec<Reply.Encoded>>
1244 }
1245> = Schema.Struct({
1246 envelope: Schema.toCodecJson(Envelope.Partial),
1247 lastSentReply: Schema.Option(Reply.Encoded)
1248})
1249
1250const decodeEnvelopeWithReply = Schema.decodeEffect(EnvelopeWithReply)