1/**
2 * Persists cluster mailbox messages and replies in SQL.
3 *
4 * The SQL-backed `MessageStorage` stores encoded cluster envelopes and reply
5 * chunks so runners can recover mailbox state after restarts. It supports
6 * redelivering unprocessed messages, deduplicating requests by primary key, and
7 * replaying reply chunks until they are acknowledged. This module includes the
8 * storage constructor, layers, migrations, optional table prefixes, and the row
9 * mapping needed by encoded message storage.
10 *
11 * Request deduplication keys that exceed the 255-character `message_id`
12 * column are hashed with the `Crypto` service before they are written, so
13 * composed keys of any length are supported; shorter keys are stored as
14 * plaintext, byte-compatible with rows written by previous versions.
15 *
16 * @stability unstable
17 * @since 4.0.0
18 */
19// eslint-disable effect/no-bigint-literals
20import * as Arr from "../Array.ts"
21import * as Crypto from "../Crypto.ts"
22import * as Effect from "../Effect.ts"
23import * as Hex from "../encoding/Hex.ts"
24import * as Layer from "../Layer.ts"
25import * as Option from "../Option.ts"
26import type * as PlatformError from "../PlatformError.ts"
27import * as Schedule from "../Schedule.ts"
28import * as Migrator from "../sql/Migrator.ts"
29import * as SqlClient from "../sql/SqlClient.ts"
30import type { Row } from "../sql/SqlConnection.ts"
31import { isSqlError, type SqlError } from "../sql/SqlError.ts"
32import type { Fragment } from "../sql/Statement.ts"
33import { PersistenceError } from "./ClusterError.ts"
34import type * as EntityAddress from "./EntityAddress.ts"
35import type * as Envelope from "./Envelope.ts"
36import * as MessageStorage from "./MessageStorage.ts"
37import { SaveResultEncoded } from "./MessageStorage.ts"
38import type * as Reply from "./Reply.ts"
39import * as ShardId from "./ShardId.ts"
40import type { ShardingConfig } from "./ShardingConfig.ts"
41import * as Snowflake from "./Snowflake.ts"
42
43const withTracerDisabled = Effect.withTracerEnabled(false)
44
45/**
46 * Creates a SQL-backed encoded message storage driver, running its migrations
47 * and using the optional table prefix.
48 *
49 * **When to use**
50 *
51 * Use when you need the SQL-backed encoded driver directly, such as when
52 * composing a custom message storage adapter.
53 *
54 * **Details**
55 *
56 * The optional `prefix` controls the table names for messages, replies, and
57 * migrations; when omitted, `cluster` is used.
58 *
59 * **Gotchas**
60 *
61 * Changing `prefix` after deployment points the runtime at a different set of
62 * tables, including the migration history table.
63 *
64 * @see {@link make} for the decoded `MessageStorage` constructor
65 *
66 * @stability unstable
67 * @category constructors
68 * @since 4.0.0
69 */
70export const makeEncoded: (options?: {
71 readonly prefix?: string | undefined
72}) => Effect.Effect<
73 MessageStorage.Encoded,
74 never,
75 SqlClient.SqlClient | Crypto.Crypto
76> = Effect.fnUntraced(function*(options) {
77 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
78 const crypto = yield* Crypto.Crypto
79 const prefix = options?.prefix ?? "cluster"
80 const table = (name: string) => `${prefix}_${name}`
81
82 yield* Effect.orDie(
83 Migrator.make({})({
84 loader: migrations(options),
85 table: table("migrations")
86 })
87 )
88
89 const messageKindAckChunk = sql.literal(String(messageKind.AckChunk))
90 const messageKindInterrupt = sql.literal(String(messageKind.Interrupt))
91 const replyKindWithExit = sql.literal(String(replyKind.WithExit))
92
93 const messagesTable = table("messages")
94 const messagesTableSql = sql(messagesTable)
95
96 const repliesTable = table("replies")
97 const repliesTableSql = sql(repliesTable)
98
99 // The composed primary key (`entityType/entityId/tag/id`) can legally exceed
100 // the 255-character `message_id` column: entity_type(150) + entity_id(255) +
101 // tag(50) alone total 458 characters before the RPC primary key is appended.
102 // Keys that fit are stored as-is, keeping `message_id` byte-compatible with
103 // rows written by previous versions; longer keys are stored as a SHA-256
104 // digest (64 hex characters, collision probability negligible at a 2^128
105 // birthday bound). Digests never contain "/" while composed keys always do,
106 // so the two encodings cannot collide.
107 const encoder = new TextEncoder()
108 const messageIdForPrimaryKey = (primaryKey: string): Effect.Effect<string, PlatformError.PlatformError> =>
109 primaryKey.length <= 255
110 ? Effect.succeed(primaryKey)
111 : Effect.map(crypto.digest("SHA-256", encoder.encode(primaryKey)), Hex.encode)
112
113 const messageIdEnforcesWidth = sql.onDialectOrElse({
114 mssql: () => true,
115 mysql: () => true,
116 pg: () => true,
117 orElse: () => false
118 })
119 // sqlite's TEXT message_id column stored over-length plaintext keys before
120 // digests were introduced; those legacy rows need a plaintext fallback read
121 const mayHaveLegacyRow = (primaryKey: string): boolean => !messageIdEnforcesWidth && primaryKey.length > 255
122
123 const envelopeToRow = (
124 envelope: Envelope.Encoded,
125 message_id: string | null,
126 deliver_at: number | null
127 ): MessageRow => {
128 switch (envelope._tag) {
129 case "Request":
130 return {
131 id: envelope.requestId,
132 message_id,
133 shard_id: ShardId.toString(envelope.address.shardId),
134 entity_type: envelope.address.entityType,
135 entity_id: envelope.address.entityId,
136 kind: messageKind.Request,
137 tag: envelope.tag,
138 payload: JSON.stringify(envelope.payload),
139 headers: JSON.stringify(envelope.headers),
140 trace_id: envelope.traceId ?? null,
141 span_id: envelope.spanId ?? null,
142 sampled: envelope.sampled === undefined
143 ? null
144 : supportsBooleans
145 ? envelope.sampled
146 : envelope.sampled
147 ? 1
148 : 0,
149 request_id: envelope.requestId,
150 reply_id: null,
151 deliver_at
152 }
153 case "AckChunk":
154 return {
155 id: envelope.id,
156 message_id,
157 shard_id: ShardId.toString(envelope.address.shardId),
158 entity_type: envelope.address.entityType,
159 entity_id: envelope.address.entityId,
160 kind: messageKind.AckChunk,
161 tag: null,
162 payload: null,
163 headers: null,
164 trace_id: null,
165 span_id: null,
166 sampled: null,
167 request_id: envelope.requestId,
168 reply_id: envelope.replyId,
169 deliver_at
170 }
171 case "Interrupt":
172 return {
173 id: envelope.id,
174 message_id,
175 shard_id: ShardId.toString(envelope.address.shardId),
176 entity_type: envelope.address.entityType,
177 entity_id: envelope.address.entityId,
178 kind: messageKind.Interrupt,
179 payload: null,
180 tag: null,
181 headers: null,
182 trace_id: null,
183 span_id: null,
184 sampled: null,
185 request_id: envelope.requestId,
186 reply_id: null,
187 deliver_at
188 }
189 }
190 }
191
192 const replyToRow = (reply: Reply.Encoded): ReplyRow => ({
193 id: reply.id,
194 kind: replyKind[reply._tag],
195 request_id: reply.requestId,
196 payload: reply._tag === "WithExit" ? JSON.stringify(reply.exit) : JSON.stringify(reply.values),
197 sequence: reply._tag === "Chunk" ? reply.sequence : null
198 })
199
200 const supportsBooleans = sql.onDialectOrElse({
201 mssql: () => false,
202 sqlite: () => false,
203 orElse: () => true
204 })
205
206 const messageFromRow = (row: MessageRow & ReplyJoinRow): {
207 readonly envelope: Envelope.Encoded
208 readonly lastSentReply: Option.Option<Reply.Encoded>
209 } => {
210 switch (Number(row.kind) as 0 | 1 | 2) {
211 case 0:
212 return {
213 envelope: {
214 _tag: "Request",
215 requestId: String(row.id),
216 address: {
217 shardId: ShardId.fromStringEncoded(row.shard_id),
218 entityType: row.entity_type,
219 entityId: row.entity_id
220 },
221 tag: row.tag!,
222 payload: JSON.parse(row.payload!),
223 headers: JSON.parse(row.headers!),
224 ...(row.trace_id ?
225 ({
226 traceId: row.trace_id,
227 spanId: row.span_id!,
228 sampled: !!row.sampled
229 }) :
230 undefined)
231 },
232 lastSentReply: row.reply_reply_id ?
233 Option.some({
234 _tag: "Chunk",
235 id: String(row.reply_reply_id),
236 requestId: String(row.request_id),
237 sequence: Number(row.reply_sequence!),
238 values: JSON.parse(row.reply_payload!)
239 }) :
240 Option.none()
241 }
242 case 1:
243 return {
244 envelope: {
245 _tag: "AckChunk",
246 id: String(row.id),
247 requestId: String(row.request_id!),
248 replyId: String(row.reply_id!),
249 address: {
250 shardId: ShardId.fromStringEncoded(row.shard_id),
251 entityType: row.entity_type,
252 entityId: row.entity_id
253 }
254 },
255 lastSentReply: Option.none()
256 }
257 case 2:
258 return {
259 envelope: {
260 _tag: "Interrupt",
261 id: String(row.id),
262 requestId: String(row.request_id!),
263 address: {
264 shardId: ShardId.fromStringEncoded(row.shard_id),
265 entityType: row.entity_type,
266 entityId: row.entity_id
267 }
268 },
269 lastSentReply: Option.none()
270 }
271 }
272 }
273
274 const sqlFalse = sql.literal(supportsBooleans ? "FALSE" : "0")
275 const sqlTrue = sql.literal(supportsBooleans ? "TRUE" : "1")
276
277 const selectByMessageId = (message_id: string): Effect.Effect<ReadonlyArray<Row>, SqlError> =>
278 sql`
279 SELECT m.id, r.id as reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence
280 FROM ${messagesTableSql} m
281 LEFT JOIN ${repliesTableSql} r ON r.id = m.last_reply_id
282 WHERE m.message_id = ${message_id}
283 `
284
285 const insertEnvelope: (
286 row: MessageRow,
287 message_id: string
288 ) => Effect.Effect<ReadonlyArray<Row>, SqlError> = sql.onDialectOrElse({
289 pg: () => (row, message_id) =>
290 sql`
291 INSERT INTO ${messagesTableSql} ${sql.insert(row)}
292 ON CONFLICT (message_id) DO NOTHING
293 RETURNING id
294 `.pipe(Effect.flatMap((rows) => {
295 // inserted a new row
296 if (rows.length > 0) return Effect.succeed([])
297 return selectByMessageId(message_id)
298 })),
299 mysql: () => (row, message_id) =>
300 Effect.flatMap(
301 sql`INSERT IGNORE INTO ${messagesTableSql} ${sql.insert(row)}`.raw,
302 (row: any) => {
303 if (row.affectedRows > 0) {
304 return Effect.succeed([])
305 }
306 return selectByMessageId(message_id)
307 }
308 ),
309 mssql: () => (row, message_id) =>
310 sql`
311 MERGE ${messagesTableSql} WITH (HOLDLOCK) AS target
312 USING (SELECT ${message_id} as message_id) AS source
313 ON target.message_id = source.message_id
314 WHEN NOT MATCHED THEN
315 INSERT ${sql.insert(row)}
316 OUTPUT
317 inserted.id,
318 CASE
319 WHEN inserted.id IS NULL THEN (
320 SELECT r.id, r.kind, r.payload
321 FROM ${repliesTableSql} r
322 WHERE r.id = target.last_reply_id
323 )
324 END as reply_id,
325 CASE
326 WHEN inserted.id IS NULL THEN (
327 SELECT r.kind
328 FROM ${repliesTableSql} r
329 WHERE r.id = target.last_reply_id
330 )
331 END as reply_kind,
332 CASE
333 WHEN inserted.id IS NULL THEN (
334 SELECT r.payload
335 FROM ${repliesTableSql} r
336 WHERE r.id = target.last_reply_id
337 )
338 END as reply_payload,
339 CASE
340 WHEN inserted.id IS NULL THEN (
341 SELECT r.sequence
342 FROM ${repliesTableSql} r
343 WHERE r.id = target.last_reply_id
344 )
345 END as reply_sequence;
346 `,
347 orElse: () => (row, message_id) =>
348 selectByMessageId(message_id).pipe(
349 Effect.tap(sql`INSERT OR IGNORE INTO ${messagesTableSql} ${sql.insert(row)}`),
350 sql.withTransaction,
351 Effect.retry({ times: 3 })
352 )
353 })
354
355 const tenMinutesAgo = sql.onDialectOrElse({
356 mssql: () => sql.literal(`DATEADD(MINUTE, -10, GETDATE())`),
357 mysql: () => sql.literal(`NOW() - INTERVAL 10 MINUTE`),
358 pg: () => sql.literal(`NOW() - INTERVAL '10 minutes'`),
359 orElse: () => sql.literal(`DATETIME('now', '-10 minute')`)
360 })
361 const sqlNowString = sql.onDialectOrElse({
362 pg: () => "NOW()",
363 mysql: () => "NOW()",
364 mssql: () => "GETDATE()",
365 orElse: () => "CURRENT_TIMESTAMP"
366 })
367 const sqlNow = sql.literal(sqlNowString)
368
369 const wrapString = sql.onDialectOrElse({
370 mssql: () => (s: string) => `N'${s}'`,
371 orElse: () => (s: string) => `'${s}'`
372 })
373 const forUpdate = sql.onDialectOrElse({
374 sqlite: () => sql.literal(""),
375 orElse: () => sql.literal("FOR UPDATE")
376 })
377
378 const emptyFragment = sql.literal("")
379 // mssql only limits with OFFSET/FETCH, which requires the ORDER BY that
380 // both queries already have
381 const limitFragment = sql.onDialectOrElse({
382 mssql: () => (limit: number) => sql.literal(`OFFSET 0 ROWS FETCH NEXT ${Math.floor(limit)} ROWS ONLY`),
383 orElse: () => (limit: number) => sql.literal(`LIMIT ${Math.floor(limit)}`)
384 })
385 const groupAddresses = (addresses: ReadonlyArray<EntityAddress.EntityAddress>) => {
386 const byShard = new Map<string, Map<string, Array<EntityAddress.EntityAddress>>>()
387 for (const address of addresses) {
388 const shardId = address.shardId.toString()
389 let byEntityType = byShard.get(shardId)
390 if (byEntityType === undefined) {
391 byShard.set(shardId, byEntityType = new Map())
392 }
393 const group = byEntityType.get(address.entityType)
394 if (group === undefined) {
395 byEntityType.set(address.entityType, [address])
396 } else {
397 group.push(address)
398 }
399 }
400 return Array.from(
401 byShard,
402 ([shardId, byEntityType]) =>
403 Array.from(byEntityType, ([entityType, addresses]) => ({ shardId, entityType, addresses }))
404 ).flat()
405 }
406 type UnprocessedOptions = {
407 readonly limit?: number | undefined
408 readonly addresses?: ReadonlyArray<EntityAddress.EntityAddress> | undefined
409 }
410 const unprocessedFilters = (options?: UnprocessedOptions | undefined) => ({
411 addressFilter: options?.addresses !== undefined
412 ? sql`AND (${
413 sql.or(
414 groupAddresses(options.addresses).map((group) =>
415 sql.and([
416 sql`m.shard_id = ${group.shardId}`,
417 sql`m.entity_type = ${group.entityType}`,
418 sql.in("m.entity_id", group.addresses.map((address) => address.entityId))
419 ])
420 )
421 )
422 })`
423 : emptyFragment,
424 limit: options?.limit !== undefined ? limitFragment(options.limit) : emptyFragment
425 })
426 type UnprocessedFilters = ReturnType<typeof unprocessedFilters>
427
428 const getUnprocessedMessagesForDialect = sql.onDialectOrElse({
429 pg: () => (shardIds: ReadonlyArray<string>, now: number, filters: UnprocessedFilters) =>
430 sql<MessageJoinRow>`
431 WITH messages AS (
432 UPDATE ${messagesTableSql} m
433 SET last_read = ${sqlNow}
434 FROM (
435 SELECT m.*
436 FROM ${messagesTableSql} m
437 WHERE m.shard_id IN (${sql.literal(shardIds.map(wrapString).join(","))})
438 ${filters.addressFilter}
439 AND NOT EXISTS (
440 SELECT 1 FROM ${repliesTableSql}
441 WHERE request_id = m.request_id
442 AND (kind = ${replyKindWithExit} OR acked = ${sqlFalse})
443 )
444 AND m.processed = ${sqlFalse}
445 AND (m.last_read IS NULL OR m.last_read < ${tenMinutesAgo})
446 AND (m.deliver_at IS NULL OR m.deliver_at <= ${sql.literal(String(now))})
447 ORDER BY m.rowid ASC
448 ${filters.limit}
449 FOR UPDATE
450 ) AS ids
451 LEFT JOIN ${repliesTableSql} r ON r.id = ids.last_reply_id
452 WHERE m.id = ids.id
453 RETURNING ids.*, r.id as reply_reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence
454 )
455 SELECT * FROM messages ORDER BY rowid ASC
456 `,
457 orElse: () => (shardIds: ReadonlyArray<string>, now: number, filters: UnprocessedFilters) =>
458 sql<MessageJoinRow>`
459 SELECT m.*, r.id as reply_reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence
460 FROM ${messagesTableSql} m
461 LEFT JOIN ${repliesTableSql} r ON r.id = m.last_reply_id
462 WHERE m.shard_id IN (${sql.literal(shardIds.map(wrapString).join(","))})
463 ${filters.addressFilter}
464 AND NOT EXISTS (
465 SELECT 1 FROM ${repliesTableSql}
466 WHERE request_id = m.request_id
467 AND (kind = ${replyKindWithExit} OR acked = ${sqlFalse})
468 )
469 AND processed = ${sqlFalse}
470 AND (m.last_read IS NULL OR m.last_read < ${tenMinutesAgo})
471 AND (m.deliver_at IS NULL OR m.deliver_at <= ${sql.literal(String(now))})
472 ORDER BY m.rowid ASC
473 ${filters.limit}
474 ${forUpdate}
475 `.unprepared.pipe(
476 Effect.tap((rows) => {
477 if (rows.length === 0) {
478 return Effect.void
479 }
480 return sql`
481 UPDATE ${messagesTableSql}
482 SET last_read = ${sqlNow}
483 WHERE id IN (${sql.literal(rows.map((row) => row.id).join(","))})
484 `.unprepared
485 }),
486 sql.withTransaction
487 )
488 })
489 const getUnprocessedMessages = (
490 shardIds: ReadonlyArray<string>,
491 now: number,
492 options?: UnprocessedOptions | undefined
493 ) =>
494 options?.addresses?.length === 0
495 ? Effect.succeed([])
496 : getUnprocessedMessagesForDialect(shardIds, now, unprocessedFilters(options))
497
498 const resetUnprocessed = sql.onDialectOrElse({
499 // Match the claim query's lock order to avoid deadlocks during resets.
500 pg: () => (filter: Fragment) =>
501 sql`
502 WITH messages AS (
503 SELECT id FROM ${messagesTableSql}
504 WHERE processed = ${sqlFalse} AND ${filter}
505 ORDER BY rowid ASC
506 FOR UPDATE
507 )
508 UPDATE ${messagesTableSql}
509 SET last_read = NULL
510 WHERE id IN (SELECT id FROM messages)
511 `,
512 orElse: () => (filter: Fragment) =>
513 sql`
514 UPDATE ${messagesTableSql}
515 SET last_read = NULL
516 WHERE processed = ${sqlFalse} AND ${filter}
517 `
518 })
519
520 const encoded: MessageStorage.Encoded = {
521 saveEnvelope: ({ deliverAt, envelope, primaryKey }) =>
522 Effect.suspend(() => {
523 let insert: Effect.Effect<ReadonlyArray<Row>, SqlError | PlatformError.PlatformError>
524 if (primaryKey !== null) {
525 insert = Effect.flatMap(messageIdForPrimaryKey(primaryKey), (messageId) => {
526 const row = envelopeToRow(envelope, messageId, deliverAt)
527 if (!mayHaveLegacyRow(primaryKey)) {
528 return insertEnvelope(row, messageId)
529 }
530 return Effect.flatMap(
531 selectByMessageId(primaryKey),
532 (rows) => rows.length > 0 ? Effect.succeed(rows) : insertEnvelope(row, messageId)
533 )
534 })
535 } else {
536 const row = envelopeToRow(envelope, null, deliverAt)
537 insert = Effect.as(sql`INSERT INTO ${messagesTableSql} ${sql.insert(row)}`.unprepared, [])
538 if (envelope._tag === "AckChunk") {
539 insert = sql`UPDATE ${repliesTableSql} SET acked = ${sqlTrue} WHERE id = ${envelope.replyId}`.pipe(
540 Effect.andThen(
541 sql`UPDATE ${messagesTableSql} SET processed = ${sqlTrue} WHERE processed = ${sqlFalse} AND request_id = ${envelope.requestId} AND kind = ${messageKindAckChunk}`
542 ),
543 Effect.andThen(insert),
544 sql.withTransaction
545 )
546 }
547 }
548 return insert.pipe(
549 Effect.map((rows) => {
550 if (rows.length === 0) {
551 return SaveResultEncoded.Success()
552 }
553 const row = rows[0]
554 const replyKindNum = typeof row.reply_kind === "bigint" ? Number(row.reply_kind) : row.reply_kind
555 const lastReceivedReply: Option.Option<Reply.Encoded> = row.reply_id
556 ? Option.some(
557 replyKindNum === replyKind.WithExit
558 ? {
559 id: String(row.reply_id),
560 requestId: String(row.id),
561 _tag: "WithExit",
562 exit: JSON.parse(row.reply_payload as string)
563 }
564 : {
565 id: String(row.reply_id),
566 requestId: String(row.id),
567 _tag: "Chunk",
568 sequence: Number(row.reply_sequence),
569 values: JSON.parse(row.reply_payload as string)
570 }
571 )
572 : Option.none()
573 return SaveResultEncoded.Duplicate({
574 originalId: Snowflake.Snowflake(row.id as any),
575 lastReceivedReply
576 })
577 })
578 )
579 }).pipe(
580 Effect.provideService(SqlClient.SafeIntegers, true),
581 PersistenceError.refail,
582 withTracerDisabled
583 ),
584
585 saveReply: (reply) =>
586 Effect.suspend(() => {
587 const row = replyToRow(reply)
588 const update = reply._tag === "Chunk" ?
589 sql`UPDATE ${messagesTableSql} SET last_reply_id = ${reply.id} WHERE id = ${reply.requestId}` :
590 sql`UPDATE ${messagesTableSql} SET processed = ${sqlTrue}, last_reply_id = ${reply.id} WHERE request_id = ${reply.requestId}`
591 return update.unprepared.pipe(
592 Effect.andThen(sql`INSERT INTO ${repliesTableSql} ${sql.insert(row)}`),
593 sql.withTransaction
594 )
595 }).pipe(
596 Effect.asVoid,
597 PersistenceError.refail,
598 withTracerDisabled
599 ),
600
601 clearReplies: Effect.fnUntraced(
602 function*(requestId, options) {
603 const expectedReplyId = options?.expectedReplyId
604 if (expectedReplyId === undefined) {
605 yield* sql`DELETE FROM ${repliesTableSql} WHERE request_id = ${String(requestId)} AND kind = 0`
606 yield* sql`DELETE FROM ${messagesTableSql} WHERE request_id = ${
607 String(requestId)
608 } AND kind = ${messageKindInterrupt}`
609 yield* sql`UPDATE ${messagesTableSql} SET processed = ${sqlFalse}, last_reply_id = NULL, last_read = NULL WHERE request_id = ${
610 String(requestId)
611 }`
612 return
613 }
614 const id = String(requestId)
615 const expected = String(expectedReplyId)
616 // Check at the storage boundary, not at the earlier workflow read.
617 yield* sql`DELETE FROM ${repliesTableSql} WHERE request_id = ${id} AND kind = 0
618 AND EXISTS (SELECT 1 FROM ${messagesTableSql} WHERE id = ${id} AND last_reply_id = ${expected})`
619 yield* sql`UPDATE ${messagesTableSql} SET processed = ${sqlFalse}, last_reply_id = NULL, last_read = NULL
620 WHERE request_id = ${id} AND last_reply_id = ${expected}`
621 },
622 sql.withTransaction,
623 PersistenceError.refail,
624 withTracerDisabled
625 ),
626
627 requestIdForPrimaryKey: (primaryKey) =>
628 messageIdForPrimaryKey(primaryKey).pipe(
629 Effect.flatMap((messageId) =>
630 sql<{ id: string | bigint }>`SELECT id FROM ${messagesTableSql} WHERE message_id = ${messageId}`
631 ),
632 Effect.flatMap((rows) =>
633 rows.length === 0 && mayHaveLegacyRow(primaryKey)
634 ? sql<{ id: string | bigint }>`SELECT id FROM ${messagesTableSql} WHERE message_id = ${primaryKey}`
635 : Effect.succeed(rows)
636 ),
637 Effect.map((rows) => Option.map(Option.fromNullishOr(rows[0]?.id), Snowflake.Snowflake)),
638 Effect.provideService(SqlClient.SafeIntegers, true),
639 PersistenceError.refail,
640 withTracerDisabled
641 ),
642
643 repliesFor: (requestIds) =>
644 // replies where:
645 // - the request is in the list
646 // - the kind is WithExit
647 // - or the kind is Chunk and has not been acked yet
648 sql<ReplyRow>`
649 SELECT id, kind, request_id, payload, sequence
650 FROM ${repliesTableSql}
651 WHERE request_id IN (${sql.literal(requestIds.join(","))})
652 AND (
653 kind = ${replyKindWithExit}
654 OR (
655 kind IS NULL
656 AND acked = ${sqlFalse}
657 )
658 )
659 ORDER BY rowid ASC
660 `.unprepared.pipe(
661 Effect.provideService(SqlClient.SafeIntegers, true),
662 Effect.map(Arr.map(replyFromRow)),
663 PersistenceError.refail,
664 withTracerDisabled
665 ),
666
667 repliesForUnfiltered: (requestIds) =>
668 sql<ReplyRow>`
669 SELECT id, kind, request_id, payload, sequence
670 FROM ${repliesTableSql}
671 WHERE request_id IN (${sql.literal(requestIds.join(","))})
672 ORDER BY rowid ASC
673 `.unprepared.pipe(
674 Effect.provideService(SqlClient.SafeIntegers, true),
675 Effect.map(Arr.map(replyFromRow)),
676 PersistenceError.refail,
677 withTracerDisabled
678 ),
679
680 unprocessedMessages: Effect.fnUntraced(
681 function*(shardIds, now, options) {
682 const rows = yield* getUnprocessedMessages(shardIds, now, options)
683 if (rows.length === 0) {
684 return []
685 }
686 const messages: Array<{
687 readonly envelope: Envelope.Encoded
688 readonly lastSentReply: Option.Option<Reply.Encoded>
689 }> = new Array(rows.length)
690 const ids = new Array<string>(rows.length)
691 for (let i = 0; i < rows.length; i++) {
692 messages[i] = messageFromRow(rows[i])
693 ids[i] = String(rows[i].id)
694 }
695 return messages
696 },
697 Effect.provideService(SqlClient.SafeIntegers, true),
698 PersistenceError.refail,
699 withTracerDisabled
700 ),
701
702 unprocessedMessagesById(ids, now) {
703 const idArr = Array.from(ids, (id) => String(id))
704 return sql<MessageRow & ReplyJoinRow>`
705 SELECT m.*, r.id as reply_reply_id, r.kind as reply_kind, r.payload as reply_payload, r.sequence as reply_sequence
706 FROM ${messagesTableSql} m
707 LEFT JOIN ${repliesTableSql} r ON r.id = m.last_reply_id
708 WHERE m.id IN (${sql.literal(idArr.join(","))})
709 AND NOT EXISTS (
710 SELECT 1 FROM ${repliesTableSql}
711 WHERE request_id = m.request_id
712 AND (kind = ${replyKindWithExit} OR acked = ${sqlFalse})
713 )
714 AND m.processed = ${sqlFalse}
715 AND (m.deliver_at IS NULL OR m.deliver_at <= ${sql.literal(String(now))})
716 ORDER BY m.rowid ASC
717 `.unprepared.pipe(
718 Effect.map(Arr.map(messageFromRow)),
719 Effect.provideService(SqlClient.SafeIntegers, true),
720 PersistenceError.refail,
721 withTracerDisabled
722 )
723 },
724
725 resetRequests: (requestIds) =>
726 requestIds.length === 0
727 ? Effect.void
728 : sql`
729 UPDATE ${messagesTableSql}
730 SET last_read = NULL
731 WHERE processed = ${sqlFalse}
732 AND ${sql.in("id", requestIds.map(String))}
733 `.pipe(
734 Effect.asVoid,
735 PersistenceError.refail,
736 withTracerDisabled
737 ),
738
739 resetAddresses: (addresses) =>
740 addresses.length === 0
741 ? Effect.void
742 : resetUnprocessed(sql`(${
743 sql.or(
744 groupAddresses(addresses).map(
745 (group) =>
746 sql.and([
747 sql`shard_id = ${group.shardId}`,
748 sql`entity_type = ${group.entityType}`,
749 sql.in("entity_id", group.addresses.map((address) => address.entityId))
750 ])
751 )
752 )
753 })`).pipe(
754 Effect.asVoid,
755 PersistenceError.refail,
756 withTracerDisabled
757 ),
758
759 clearAddress: (address) =>
760 sql`
761 DELETE FROM ${repliesTableSql}
762 WHERE request_id IN (
763 SELECT id FROM ${messagesTableSql}
764 WHERE entity_type = ${address.entityType}
765 AND entity_id = ${address.entityId}
766 )
767 `.pipe(
768 Effect.andThen(
769 sql`
770 DELETE FROM ${messagesTableSql}
771 WHERE entity_type = ${address.entityType}
772 AND entity_id = ${address.entityId}
773 `
774 ),
775 sql.withTransaction,
776 Effect.asVoid,
777 PersistenceError.refail,
778 withTracerDisabled
779 ),
780
781 resetShards: (shardIds) =>
782 resetUnprocessed(sql`shard_id IN (${sql.literal(shardIds.map(wrapString).join(","))})`).pipe(
783 Effect.asVoid,
784 PersistenceError.refail,
785 withTracerDisabled
786 ),
787
788 withTransaction: (effect) =>
789 sql.withTransaction(effect).pipe(
790 Effect.catchIf(isSqlError, Effect.die)
791 )
792 }
793 return encoded
794}, withTracerDisabled)
795
796/**
797 * Creates a SQL-backed `MessageStorage` implementation, running its migrations
798 * and using the optional table prefix.
799 *
800 * @stability unstable
801 * @category constructors
802 * @since 4.0.0
803 */
804export const make: (options?: {
805 readonly prefix?: string | undefined
806}) => Effect.Effect<
807 MessageStorage.MessageStorage["Service"],
808 never,
809 SqlClient.SqlClient | Snowflake.Generator | Crypto.Crypto
810> = (options) => Effect.flatMap(makeEncoded(options), MessageStorage.makeEncoded)
811
812/**
813 * Layer that provides SQL-backed `MessageStorage` using the default table prefix
814 * and the default snowflake generator.
815 *
816 * **When to use**
817 *
818 * Use when a cluster should persist mailbox messages and replies in SQL using
819 * the default `cluster` table prefix and the standard snowflake generator.
820 *
821 * **Details**
822 *
823 * The layer runs the SQL migrations through `make`, provides `MessageStorage`,
824 * and supplies `Snowflake.layerGenerator` internally. Callers still provide
825 * `SqlClient`, `ShardingConfig`, and `Crypto.Crypto`, which is used to hash
826 * message deduplication keys that would overflow the fixed-width
827 * `message_id` column.
828 *
829 * **Gotchas**
830 *
831 * This layer always uses the `cluster` table prefix. Use `layerWith` before
832 * deployment if you need a different stable prefix, because changing prefixes
833 * later points the runtime at a different set of tables.
834 *
835 * @see {@link layerWith} for the same SQL storage layer with a custom table prefix
836 * @see {@link make} for the lower-level service constructor that uses an existing `Snowflake.Generator`
837 *
838 * @stability unstable
839 * @category layers
840 * @since 4.0.0
841 */
842export const layer: Layer.Layer<
843 MessageStorage.MessageStorage,
844 never,
845 SqlClient.SqlClient | ShardingConfig | Crypto.Crypto
846> = Layer.effect(MessageStorage.MessageStorage, make()).pipe(
847 Layer.provide(Snowflake.layerGenerator)
848)
849
850/**
851 * Layer that provides SQL-backed `MessageStorage` using a custom table prefix.
852 *
853 * @stability unstable
854 * @category layers
855 * @since 4.0.0
856 */
857export const layerWith = (options: {
858 readonly prefix?: string | undefined
859}): Layer.Layer<MessageStorage.MessageStorage, never, SqlClient.SqlClient | ShardingConfig | Crypto.Crypto> =>
860 Layer.effect(MessageStorage.MessageStorage, make(options)).pipe(
861 Layer.provide(Snowflake.layerGenerator)
862 )
863
864// -------------------------------------------------------------------------------------------------
865// internal
866// -------------------------------------------------------------------------------------------------
867
868const migrations = (options?: {
869 readonly prefix?: string | undefined
870}) => {
871 const prefix = options?.prefix ?? "cluster"
872 const table = (name: string) => `${prefix}_${name}`
873 const messagesTable = table("messages")
874 const repliesTable = table("replies")
875
876 return Migrator.fromRecord({
877 "0001_create_tables": Effect.gen(function*() {
878 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
879 const messagesTableSql = sql(messagesTable)
880 const repliesTableSql = sql(repliesTable)
881
882 yield* sql.onDialectOrElse({
883 mssql: () =>
884 sql`
885 IF OBJECT_ID(N'${messagesTableSql}', N'U') IS NULL
886 CREATE TABLE ${messagesTableSql} (
887 id BIGINT PRIMARY KEY,
888 rowid BIGINT IDENTITY(1,1),
889 message_id VARCHAR(255),
890 shard_id VARCHAR(50) NOT NULL,
891 entity_type VARCHAR(150) NOT NULL,
892 entity_id VARCHAR(255) NOT NULL,
893 kind INT NOT NULL,
894 tag VARCHAR(50),
895 payload TEXT,
896 headers TEXT,
897 trace_id VARCHAR(32),
898 span_id VARCHAR(16),
899 sampled BIT,
900 processed BIT NOT NULL DEFAULT 0,
901 request_id BIGINT NOT NULL,
902 reply_id BIGINT,
903 last_reply_id BIGINT,
904 last_read DATETIME,
905 deliver_at BIGINT,
906 UNIQUE (message_id)
907 )
908 `,
909 mysql: () =>
910 sql`
911 CREATE TABLE IF NOT EXISTS ${messagesTableSql} (
912 id BIGINT NOT NULL,
913 rowid BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY,
914 message_id VARCHAR(255),
915 shard_id VARCHAR(50) NOT NULL,
916 entity_type VARCHAR(150) NOT NULL,
917 entity_id VARCHAR(255) NOT NULL,
918 kind INT NOT NULL,
919 tag VARCHAR(50),
920 payload TEXT,
921 headers TEXT,
922 trace_id VARCHAR(32),
923 span_id VARCHAR(16),
924 sampled BOOLEAN,
925 processed BOOLEAN NOT NULL DEFAULT FALSE,
926 request_id BIGINT NOT NULL,
927 reply_id BIGINT,
928 last_reply_id BIGINT,
929 last_read DATETIME,
930 deliver_at BIGINT,
931 UNIQUE (id),
932 UNIQUE (message_id)
933 )
934 `,
935 pg: () =>
936 sql`
937 CREATE TABLE IF NOT EXISTS ${messagesTableSql} (
938 id BIGINT PRIMARY KEY,
939 rowid BIGSERIAL,
940 message_id VARCHAR(255),
941 shard_id VARCHAR(50) NOT NULL,
942 entity_type VARCHAR(150) NOT NULL,
943 entity_id VARCHAR(255) NOT NULL,
944 kind INT NOT NULL,
945 tag VARCHAR(50),
946 payload TEXT,
947 headers TEXT,
948 trace_id VARCHAR(32),
949 span_id VARCHAR(16),
950 sampled BOOLEAN,
951 processed BOOLEAN NOT NULL DEFAULT FALSE,
952 request_id BIGINT NOT NULL,
953 reply_id BIGINT,
954 last_reply_id BIGINT,
955 last_read TIMESTAMP,
956 deliver_at BIGINT,
957 UNIQUE (message_id)
958 )
959 `.pipe(Effect.ignore),
960 orElse: () =>
961 // sqlite
962 sql`
963 CREATE TABLE IF NOT EXISTS ${messagesTableSql} (
964 id INTEGER PRIMARY KEY,
965 message_id TEXT,
966 shard_id TEXT NOT NULL,
967 entity_type TEXT NOT NULL,
968 entity_id TEXT NOT NULL,
969 kind INTEGER NOT NULL,
970 tag TEXT,
971 payload TEXT,
972 headers TEXT,
973 trace_id TEXT,
974 span_id TEXT,
975 sampled BOOLEAN,
976 processed BOOLEAN NOT NULL DEFAULT FALSE,
977 request_id INTEGER NOT NULL,
978 reply_id INTEGER,
979 last_reply_id INTEGER,
980 last_read TEXT,
981 deliver_at INTEGER,
982 UNIQUE (message_id)
983 )
984 `
985 })
986
987 // Add message indexes optimized for the specific query patterns
988 const shardLookupIndex = `${messagesTable}_shard_idx`
989 const requestIdLookupIndex = `${messagesTable}_request_id_idx`
990 yield* sql.onDialectOrElse({
991 mssql: () =>
992 sql`
993 IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = ${shardLookupIndex})
994 CREATE INDEX ${sql(shardLookupIndex)}
995 ON ${messagesTableSql} (shard_id, processed, last_read, deliver_at);
996
997 IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = ${requestIdLookupIndex})
998 CREATE INDEX ${sql(requestIdLookupIndex)}
999 ON ${messagesTableSql} (request_id);
1000 `,
1001 mysql: () =>
1002 sql`
1003 CREATE INDEX ${sql(shardLookupIndex)}
1004 ON ${messagesTableSql} (shard_id, processed, last_read, deliver_at);
1005
1006 CREATE INDEX ${sql(requestIdLookupIndex)}
1007 ON ${messagesTableSql} (request_id);
1008 `.unprepared.pipe(Effect.ignore),
1009 pg: () =>
1010 Effect.gen(function*() {
1011 yield* sql`
1012 CREATE INDEX IF NOT EXISTS ${sql(shardLookupIndex)}
1013 ON ${messagesTableSql} (shard_id, processed, last_read, deliver_at)
1014 `
1015 yield* sql`
1016 CREATE INDEX IF NOT EXISTS ${sql(requestIdLookupIndex)}
1017 ON ${messagesTableSql} (request_id)
1018 `
1019 }).pipe(
1020 sql.withTransaction,
1021 Effect.tapDefect((error) =>
1022 Effect.annotateLogs(Effect.logDebug("Failed to create indexes", error), {
1023 package: "@effect/cluster",
1024 module: "SqlMessageStorage"
1025 })
1026 ),
1027 Effect.retry({
1028 schedule: Schedule.spaced(1000)
1029 })
1030 ),
1031 orElse: () =>
1032 // sqlite
1033 Effect.all([
1034 sql`
1035 CREATE INDEX IF NOT EXISTS ${sql(shardLookupIndex)}
1036 ON ${messagesTableSql} (shard_id, processed, last_read, deliver_at)
1037 `,
1038 sql`
1039 CREATE INDEX IF NOT EXISTS ${sql(requestIdLookupIndex)}
1040 ON ${messagesTableSql} (request_id)
1041 `
1042 ]).pipe(sql.withTransaction)
1043 })
1044
1045 yield* sql.onDialectOrElse({
1046 mssql: () =>
1047 sql`
1048 IF OBJECT_ID(N'${repliesTableSql}', N'U') IS NULL
1049 CREATE TABLE ${repliesTableSql} (
1050 id BIGINT PRIMARY KEY,
1051 rowid BIGINT IDENTITY(1,1),
1052 kind INT,
1053 request_id BIGINT NOT NULL,
1054 payload TEXT NOT NULL,
1055 sequence INT,
1056 acked BIT NOT NULL DEFAULT 0,
1057 CONSTRAINT ${sql(repliesTable + "_one_exit")} UNIQUE (request_id, kind),
1058 CONSTRAINT ${sql(repliesTable + "_sequence")} UNIQUE (request_id, sequence)
1059 )
1060 `,
1061 mysql: () =>
1062 sql`
1063 CREATE TABLE IF NOT EXISTS ${repliesTableSql} (
1064 id BIGINT NOT NULL,
1065 rowid BIGINT AUTO_INCREMENT PRIMARY KEY,
1066 kind INT,
1067 request_id BIGINT NOT NULL,
1068 payload TEXT NOT NULL,
1069 sequence INT,
1070 acked BOOLEAN NOT NULL DEFAULT FALSE,
1071 UNIQUE (id),
1072 UNIQUE (request_id, kind),
1073 UNIQUE (request_id, sequence)
1074 )
1075 `,
1076 pg: () =>
1077 sql`
1078 CREATE TABLE IF NOT EXISTS ${repliesTableSql} (
1079 id BIGINT PRIMARY KEY,
1080 rowid BIGSERIAL,
1081 kind INT,
1082 request_id BIGINT NOT NULL,
1083 payload TEXT NOT NULL,
1084 sequence INT,
1085 acked BOOLEAN NOT NULL DEFAULT FALSE,
1086 UNIQUE (request_id, kind),
1087 UNIQUE (request_id, sequence)
1088 )
1089 `,
1090 orElse: () =>
1091 // sqlite
1092 sql`
1093 CREATE TABLE IF NOT EXISTS ${repliesTableSql} (
1094 id INTEGER PRIMARY KEY,
1095 kind INTEGER,
1096 request_id INTEGER NOT NULL,
1097 payload TEXT NOT NULL,
1098 sequence INTEGER,
1099 acked BOOLEAN NOT NULL DEFAULT FALSE,
1100 UNIQUE (request_id, kind),
1101 UNIQUE (request_id, sequence)
1102 )
1103 `
1104 })
1105
1106 // Add reply indexes optimized for request_id lookups
1107 const replyLookupIndex = `${repliesTable}_request_lookup_idx`
1108 yield* sql.onDialectOrElse({
1109 mssql: () =>
1110 sql`
1111 IF NOT EXISTS (SELECT * FROM sys.indexes WHERE name = ${replyLookupIndex})
1112 CREATE INDEX ${sql(replyLookupIndex)}
1113 ON ${repliesTableSql} (request_id, kind, acked);
1114 `,
1115 mysql: () =>
1116 sql`
1117 CREATE INDEX ${sql(replyLookupIndex)}
1118 ON ${repliesTableSql} (request_id, kind, acked);
1119 `.unprepared.pipe(Effect.ignore),
1120 pg: () =>
1121 sql`
1122 CREATE INDEX IF NOT EXISTS ${sql(replyLookupIndex)}
1123 ON ${repliesTableSql} (request_id, kind, acked);
1124 `.pipe(
1125 Effect.tapDefect((error) =>
1126 Effect.annotateLogs(Effect.logDebug("Failed to create indexes", error), {
1127 package: "@effect/cluster",
1128 module: "SqlMessageStorage"
1129 })
1130 ),
1131 Effect.retry({
1132 schedule: Schedule.spaced(1000)
1133 })
1134 ),
1135 orElse: () =>
1136 // sqlite
1137 sql`
1138 CREATE INDEX IF NOT EXISTS ${sql(replyLookupIndex)}
1139 ON ${repliesTableSql} (request_id, kind, acked);
1140 `
1141 })
1142 }),
1143 "0002_entity_type_size": Effect.gen(function*() {
1144 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
1145 const messagesTableSql = sql(messagesTable)
1146
1147 // resize entity_type to 150 characters
1148 yield* sql.onDialectOrElse({
1149 mssql: () =>
1150 sql`
1151 ALTER TABLE ${messagesTableSql} ALTER COLUMN entity_type VARCHAR(150) NOT NULL;
1152 `,
1153 mysql: () =>
1154 sql`
1155 ALTER TABLE ${messagesTableSql} MODIFY entity_type VARCHAR(150) NOT NULL;
1156 `.unprepared.pipe(Effect.ignore),
1157 pg: () =>
1158 sql`
1159 ALTER TABLE ${messagesTableSql} ALTER COLUMN entity_type TYPE VARCHAR(150);
1160 `,
1161 orElse: () =>
1162 // sqlite
1163 Effect.void
1164 })
1165 }),
1166 "0003_pg_messages_rowid_index": Effect.gen(function*() {
1167 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
1168 const messagesTableSql = sql(messagesTable)
1169 const rowIdIndex = `${messagesTable}_rowid_idx`
1170
1171 yield* sql.onDialectOrElse({
1172 pg: () =>
1173 sql`
1174 CREATE INDEX IF NOT EXISTS ${sql(rowIdIndex)}
1175 ON ${messagesTableSql} (rowid)
1176 `,
1177 orElse: () => Effect.void
1178 })
1179 })
1180 })
1181}
1182
1183const messageKind = {
1184 "Request": 0,
1185 "AckChunk": 1,
1186 "Interrupt": 2
1187} as const satisfies Record<Envelope.Envelope.Any["_tag"], number>
1188
1189const replyKind = {
1190 "WithExit": 0,
1191 "Chunk": null
1192} as const satisfies Record<Reply.Reply<any>["_tag"], number | null>
1193
1194const replyFromRow = (row: ReplyRow): Reply.Encoded =>
1195 row.kind !== null && Number(row.kind) === replyKind.WithExit ?
1196 {
1197 _tag: "WithExit",
1198 id: String(row.id),
1199 requestId: String(row.request_id),
1200 exit: JSON.parse(row.payload)
1201 } :
1202 {
1203 _tag: "Chunk",
1204 id: String(row.id),
1205 requestId: String(row.request_id),
1206 values: JSON.parse(row.payload),
1207 sequence: Number(row.sequence!)
1208 }
1209
1210type MessageRow = {
1211 readonly id: string | bigint
1212 readonly message_id: string | null
1213 readonly shard_id: string
1214 readonly entity_type: string
1215 readonly entity_id: string
1216 readonly kind: 0 | 1 | 2 | 0n | 1n | 2n
1217 readonly tag: string | null
1218 readonly payload: string | null
1219 readonly headers: string | null
1220 readonly trace_id: string | null
1221 readonly span_id: string | null
1222 readonly sampled: boolean | number | bigint | null
1223 readonly request_id: string | bigint | null
1224 readonly reply_id: string | bigint | null
1225 readonly deliver_at: number | bigint | null
1226}
1227
1228type ReplyRow = {
1229 readonly id: string | bigint
1230 readonly kind: 0 | null | 0n
1231 readonly request_id: string | bigint
1232 readonly payload: string
1233 readonly sequence: number | bigint | null
1234}
1235
1236type ReplyJoinRow = {
1237 readonly reply_reply_id: string | bigint | null
1238 readonly reply_payload: string | null
1239 readonly reply_sequence: number | bigint | null
1240}
1241
1242type MessageJoinRow = MessageRow & ReplyJoinRow & {
1243 readonly sequence: number | bigint
1244}