1/**
2 * Stores encoded results for `Persistable` requests.
3 *
4 * The `Persistence` service creates scoped stores keyed by each request's
5 * `PrimaryKey`. Stores read and write schema-encoded `Exit` values with
6 * optional TTLs, letting request workflows reuse expensive or idempotent results
7 * across fibers, process restarts, or workers that share a backing store.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import * as Arr from "../Array.ts"
13import * as Clock from "../Clock.ts"
14import * as Context from "../Context.ts"
15import * as Duration from "../Duration.ts"
16import * as Effect from "../Effect.ts"
17import * as Exit from "../Exit.ts"
18import { identity } from "../Function.ts"
19import { sqlCleanupBatchSize } from "../internal/persistence.ts"
20import * as Layer from "../Layer.ts"
21import * as PrimaryKey from "../PrimaryKey.ts"
22import * as Schedule from "../Schedule.ts"
23import * as Schema from "../Schema.ts"
24import type * as Scope from "../Scope.ts"
25import * as SqlClient from "../sql/SqlClient.ts"
26import type { SqlError } from "../sql/SqlError.ts"
27import * as KeyValueStore from "./KeyValueStore.ts"
28import * as Persistable from "./Persistable.ts"
29import * as Redis from "./Redis.ts"
30
31const ErrorTypeId = "~effect/persistence/Persistence/PersistenceError" as const
32
33/**
34 * Error raised by persistence and backing-store operations.
35 *
36 * @stability unstable
37 * @category errors
38 * @since 4.0.0
39 */
40export class PersistenceError extends Schema.Error<PersistenceError>(ErrorTypeId)({
41 _tag: Schema.tag("PersistenceError"),
42 message: Schema.String,
43 cause: Schema.optional(Schema.Defect())
44}) {
45 /**
46 * Marks this value as a persistence error for runtime guards.
47 *
48 * @stability unstable
49 * @since 4.0.0
50 */
51 readonly [ErrorTypeId]: typeof ErrorTypeId = ErrorTypeId
52}
53
54/**
55 * Service for creating scoped stores of persisted `Persistable` request
56 * results.
57 *
58 * @stability unstable
59 * @category services
60 * @since 4.0.0
61 */
62export class Persistence extends Context.Service<Persistence, {
63 readonly make: (options: {
64 readonly storeId: string
65 readonly timeToLive?: (exit: Exit.Exit<unknown, unknown>, key: Persistable.Any) => Duration.Input
66 }) => Effect.Effect<PersistenceStore, never, Scope.Scope>
67}>()("effect/persistence/Persistence") {}
68
69/**
70 * Typed store for persisted `Exit` values keyed by `Persistable` requests.
71 *
72 * @stability unstable
73 * @category models
74 * @since 4.0.0
75 */
76export interface PersistenceStore {
77 readonly get: <A extends Schema.Constraint, E extends Schema.Constraint>(
78 key: Persistable.Persistable<A, E>
79 ) => Effect.Effect<
80 Exit.Exit<A["Type"], E["Type"]> | undefined,
81 PersistenceError | Schema.SchemaError,
82 A["DecodingServices"] | E["DecodingServices"]
83 >
84 readonly getMany: <A extends Schema.Constraint, E extends Schema.Constraint>(
85 keys: Iterable<Persistable.Persistable<A, E>>
86 ) => Effect.Effect<
87 Array<Exit.Exit<A["Type"], E["Type"]> | undefined>,
88 PersistenceError | Schema.SchemaError,
89 A["DecodingServices"] | E["DecodingServices"]
90 >
91 readonly set: <A extends Schema.Constraint, E extends Schema.Constraint>(
92 key: Persistable.Persistable<A, E>,
93 value: Exit.Exit<A["Type"], E["Type"]>
94 ) => Effect.Effect<void, PersistenceError | Schema.SchemaError, A["EncodingServices"] | E["EncodingServices"]>
95 readonly setMany: <A extends Schema.Constraint, E extends Schema.Constraint>(
96 entries: Iterable<readonly [Persistable.Persistable<A, E>, Exit.Exit<A["Type"], E["Type"]>]>
97 ) => Effect.Effect<void, PersistenceError | Schema.SchemaError, A["EncodingServices"] | E["EncodingServices"]>
98 readonly remove: <A extends Schema.Constraint, E extends Schema.Constraint>(
99 key: Persistable.Persistable<A, E>
100 ) => Effect.Effect<void, PersistenceError>
101 readonly clear: Effect.Effect<void, PersistenceError>
102}
103
104/**
105 * Service for creating raw backing stores for persistence store ids.
106 *
107 * @stability unstable
108 * @category services
109 * @since 4.0.0
110 */
111export class BackingPersistence extends Context.Service<BackingPersistence, {
112 readonly make: (storeId: string) => Effect.Effect<BackingPersistenceStore, never, Scope.Scope>
113}>()("effect/persistence/BackingPersistence") {}
114
115/**
116 * Raw persistence backing store for JSON-compatible objects with optional
117 * TTLs.
118 *
119 * @stability unstable
120 * @category services
121 * @since 4.0.0
122 */
123export interface BackingPersistenceStore {
124 readonly get: (key: string) => Effect.Effect<object | undefined, PersistenceError>
125 readonly getMany: (
126 keys: Arr.NonEmptyArray<string>
127 ) => Effect.Effect<Arr.NonEmptyArray<object | undefined>, PersistenceError>
128 readonly set: (
129 key: string,
130 value: object,
131 ttl: Duration.Duration | undefined
132 ) => Effect.Effect<void, PersistenceError>
133 readonly setMany: (
134 entries: Arr.NonEmptyArray<readonly [key: string, value: object, ttl: Duration.Duration | undefined]>
135 ) => Effect.Effect<void, PersistenceError>
136 readonly remove: (key: string) => Effect.Effect<void, PersistenceError>
137 readonly clear: Effect.Effect<void, PersistenceError>
138}
139
140/**
141 * Provides `Persistence` from `BackingPersistence`.
142 *
143 * **Details**
144 *
145 * The layer serializes and deserializes `Persistable` exits, applies
146 * per-entry TTLs, and skips writes whose TTL is zero or negative.
147 *
148 * @stability unstable
149 * @category layers
150 * @since 4.0.0
151 */
152export const layer = Layer.effect(Persistence)(Effect.gen(function*() {
153 const backing = yield* BackingPersistence
154 const scope = yield* Effect.scope
155 return Persistence.of({
156 make: Effect.fnUntraced(function*(options) {
157 const storage = yield* backing.make(options.storeId)
158 const timeToLive = options.timeToLive ?? (() => Duration.infinity)
159
160 return identity<PersistenceStore>({
161 get: (key) =>
162 Effect.flatMap(
163 storage.get(PrimaryKey.value(key)),
164 (result) => result ? Persistable.deserializeExit(key, result) : Effect.undefined
165 ),
166 getMany: Effect.fnUntraced(function*(keys) {
167 const primaryKeys = Arr.empty<string>()
168 const persistables = Arr.empty<Persistable.Any>()
169 for (const key of keys) {
170 primaryKeys.push(PrimaryKey.value(key))
171 persistables.push(key)
172 }
173 if (!Arr.isArrayNonEmpty(primaryKeys)) return []
174
175 const results = yield* storage.getMany(primaryKeys)
176 if (results.length !== primaryKeys.length) {
177 return yield* new PersistenceError({
178 message: `Expected ${primaryKeys.length} results but got ${results.length} from backing store`
179 })
180 }
181 const out = new Array<Exit.Exit<unknown, unknown> | undefined>(primaryKeys.length)
182 let toRemove: Array<string> | undefined
183 for (let i = 0; i < results.length; i++) {
184 const key = persistables[i]
185 const result = results[i]
186 if (result === undefined) {
187 out[i] = undefined
188 continue
189 }
190 const eff = Persistable.deserializeExit(key, result)
191 const exit = Exit.isExit(eff)
192 ? eff as Exit.Exit<Exit.Exit<any, any>, Schema.SchemaError>
193 : yield* Effect.exit(eff)
194 if (Exit.isFailure(exit)) {
195 toRemove ??= []
196 toRemove.push(PrimaryKey.value(key))
197 out[i] = undefined
198 continue
199 }
200 out[i] = exit.value
201 }
202 if (toRemove) {
203 for (let i = 0; i < toRemove.length; i++) {
204 yield* Effect.forkIn(storage.remove(toRemove[i]), scope)
205 }
206 }
207 return out
208 }),
209 set(key, value) {
210 const ttl = Duration.fromInputUnsafe(timeToLive(value, key))
211 if (Duration.isZero(ttl) || Duration.isNegative(ttl)) return Effect.void
212 return Persistable.serializeExit(key, value).pipe(
213 Effect.flatMap((encoded) =>
214 storage.set(PrimaryKey.value(key), encoded as object, Duration.isFinite(ttl) ? ttl : undefined)
215 )
216 )
217 },
218 setMany: Effect.fnUntraced(function*(entries) {
219 const encodedEntries = Arr.empty<readonly [string, object, Duration.Duration | undefined]>()
220 for (const [key, value] of entries) {
221 const ttl = Duration.fromInputUnsafe(timeToLive(value, key))
222 if (Duration.isZero(ttl) || Duration.isNegative(ttl)) continue
223 const encoded = Persistable.serializeExit(key, value)
224 const exit = Exit.isExit(encoded)
225 ? encoded as Exit.Exit<unknown, Schema.SchemaError>
226 : yield* Effect.exit(encoded)
227 if (Exit.isFailure(exit)) {
228 return yield* exit
229 }
230 encodedEntries.push([PrimaryKey.value(key), exit.value as object, Duration.isFinite(ttl) ? ttl : undefined])
231 }
232 if (!Arr.isArrayNonEmpty(encodedEntries)) return
233 return yield* storage.setMany(encodedEntries)
234 }),
235 remove: (key) => storage.remove(PrimaryKey.value(key)),
236 clear: storage.clear
237 })
238 })
239 })
240}))
241
242/**
243 * Provides an in-memory `BackingPersistence` grouped by store id.
244 *
245 * **Details**
246 *
247 * Entries are process-local and expire according to their stored TTL.
248 *
249 * @stability unstable
250 * @category layers
251 * @since 4.0.0
252 */
253export const layerBackingMemory: Layer.Layer<BackingPersistence> = Layer.sync(BackingPersistence)(
254 () => {
255 const stores = new Map<string, Map<string, readonly [object, expires: number | null]>>()
256 const getStore = (storeId: string) => {
257 let store = stores.get(storeId)
258 if (store === undefined) {
259 store = new Map<string, readonly [object, expires: number | null]>()
260 stores.set(storeId, store)
261 }
262 return store
263 }
264 return BackingPersistence.of({
265 make: (storeId) =>
266 Effect.clockWith((clock) => {
267 const map = getStore(storeId)
268 const unsafeGet = (key: string): object | undefined => {
269 const value = map.get(key)
270 if (value === undefined) {
271 return undefined
272 } else if (value[1] !== null && value[1] <= clock.currentTimeMillisUnsafe()) {
273 map.delete(key)
274 return undefined
275 }
276 return value[0]
277 }
278 return Effect.succeed<BackingPersistenceStore>({
279 get: (key) => Effect.sync(() => unsafeGet(key)),
280 getMany: (keys) => Effect.sync(() => Arr.map(keys, unsafeGet)),
281 set: (key, value, ttl) => Effect.sync(() => map.set(key, [value, unsafeTtlToExpires(clock, ttl)])),
282 setMany: (entries) =>
283 Effect.sync(() => {
284 for (const [key, value, ttl] of entries) {
285 map.set(key, [value, unsafeTtlToExpires(clock, ttl)])
286 }
287 }),
288 remove: (key) => Effect.sync(() => map.delete(key)),
289 clear: Effect.sync(() => map.clear())
290 })
291 })
292 })
293 }
294)
295
296/**
297 * Provides SQL-backed persistence using one table per store id.
298 *
299 * **Details**
300 *
301 * Each table is created if needed and stores JSON-encoded values with optional
302 * expiration timestamps.
303 *
304 * @stability unstable
305 * @category layers
306 * @since 4.0.0
307 */
308export const layerBackingSqlMultiTable: Layer.Layer<
309 BackingPersistence,
310 never,
311 SqlClient.SqlClient
312> = Layer.effect(BackingPersistence)(Effect.gen(function*() {
313 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
314 return BackingPersistence.of({
315 make: Effect.fnUntraced(function*(storeId) {
316 const clock = yield* Clock.Clock
317 const tableName = `effect_persistence_${storeId}`
318 const table = sql(tableName)
319 yield* sql.onDialectOrElse({
320 mysql: () =>
321 sql`
322 CREATE TABLE IF NOT EXISTS ${table} (
323 id VARCHAR(191) PRIMARY KEY,
324 value TEXT NOT NULL,
325 expires BIGINT
326 )
327 `,
328 pg: () =>
329 sql`
330 CREATE TABLE IF NOT EXISTS ${table} (
331 id TEXT PRIMARY KEY,
332 value TEXT NOT NULL,
333 expires BIGINT
334 )
335 `,
336 mssql: () =>
337 sql`
338 IF NOT EXISTS (SELECT * FROM sysobjects WHERE name=${tableName} AND xtype='U')
339 CREATE TABLE ${table} (
340 id NVARCHAR(450) PRIMARY KEY,
341 value NVARCHAR(MAX) NOT NULL,
342 expires BIGINT
343 )
344 `,
345 // sqlite
346 orElse: () =>
347 sql`
348 CREATE TABLE IF NOT EXISTS ${table} (
349 id TEXT PRIMARY KEY,
350 value TEXT NOT NULL,
351 expires INTEGER
352 )
353 `
354 }).pipe(Effect.orDie)
355
356 // Cleanup expired entries on startup
357 yield* Effect.ignore(
358 sql`DELETE FROM ${table} WHERE expires IS NOT NULL AND expires <= ${clock.currentTimeMillisUnsafe()}`
359 )
360
361 type UpsertFn = (
362 entries: Array<{ id: string; value: string; expires: number | null }>
363 ) => Effect.Effect<unknown, SqlError>
364
365 const upsert = sql.onDialectOrElse({
366 pg: (): UpsertFn => (entries) =>
367 sql`
368 INSERT INTO ${table} ${sql.insert(entries)}
369 ON CONFLICT (id) DO UPDATE SET value=EXCLUDED.value, expires=EXCLUDED.expires
370 `.unprepared,
371 mysql: (): UpsertFn => (entries) =>
372 sql`
373 INSERT INTO ${table} ${sql.insert(entries)}
374 ON DUPLICATE KEY UPDATE value=VALUES(value), expires=VALUES(expires)
375 `.unprepared,
376 mssql: (): UpsertFn => (entries) =>
377 Effect.forEach(
378 entries,
379 (entry) =>
380 sql`
381 MERGE ${table} AS target
382 USING (SELECT ${entry.id} AS id, ${entry.value} AS value, ${entry.expires} AS expires) AS source
383 ON target.id = source.id
384 WHEN MATCHED THEN UPDATE SET value = source.value, expires = source.expires
385 WHEN NOT MATCHED THEN INSERT (id, value, expires)
386 VALUES (source.id, source.value, source.expires);
387 `,
388 { discard: true }
389 ),
390 // sqlite
391 orElse: (): UpsertFn => (entries) =>
392 sql`
393 INSERT INTO ${table} ${sql.insert(entries)}
394 ON CONFLICT(id) DO UPDATE SET value=excluded.value, expires=excluded.expires
395 `.unprepared
396 })
397
398 return identity<BackingPersistenceStore>({
399 get: (key) =>
400 sql<
401 { value: string }
402 >`SELECT value FROM ${table} WHERE id = ${key} AND (expires IS NULL OR expires > ${clock.currentTimeMillisUnsafe()})`
403 .pipe(
404 Effect.mapError((cause) =>
405 new PersistenceError({
406 message: `Failed to get key ${key} from backing store`,
407 cause
408 })
409 ),
410 Effect.flatMap((rows) => {
411 if (rows.length === 0) {
412 return Effect.undefined
413 }
414 try {
415 return Effect.succeed(JSON.parse(rows[0].value))
416 } catch (cause) {
417 return Effect.fail(
418 new PersistenceError({
419 message: `Failed to parse value for key ${key} from backing store`,
420 cause
421 })
422 )
423 }
424 })
425 ),
426 getMany: (keys) =>
427 sql<{ id: string; value: string }>`SELECT id, value FROM ${table} WHERE ${
428 sql.in("id", keys)
429 } AND (expires IS NULL OR expires > ${clock.currentTimeMillisUnsafe()})`.unprepared.pipe(
430 Effect.mapError((cause) =>
431 new PersistenceError({
432 message: `Failed to getMany from backing store`,
433 cause
434 })
435 ),
436 Effect.flatMap((rows) => {
437 const values = new Map<string, object>()
438 for (let i = 0; i < rows.length; i++) {
439 const row = rows[i]
440 try {
441 values.set(row.id, JSON.parse(row.value))
442 } catch {
443 // ignore
444 }
445 }
446 return Effect.succeed(keys.map((key) => values.get(key)) as Arr.NonEmptyArray<object | undefined>)
447 })
448 ),
449 set: (key, value, ttl) =>
450 Effect.suspend(() => {
451 try {
452 return upsert([{ id: key, value: JSON.stringify(value), expires: unsafeTtlToExpires(clock, ttl) }]).pipe(
453 Effect.mapError((cause) =>
454 new PersistenceError({
455 message: `Failed to set key ${key} in backing store`,
456 cause
457 })
458 ),
459 Effect.asVoid
460 )
461 } catch (cause) {
462 return Effect.fail(
463 new PersistenceError({
464 message: `Failed to serialize value for key ${key} to backing store`,
465 cause
466 })
467 )
468 }
469 }),
470 setMany: (entries) =>
471 Effect.suspend(() => {
472 try {
473 const encoded = entries.map(([key, value, ttl]) => ({
474 id: key,
475 value: JSON.stringify(value),
476 expires: unsafeTtlToExpires(clock, ttl)
477 }))
478 return upsert(encoded).pipe(
479 Effect.mapError((cause) =>
480 new PersistenceError({
481 message: `Failed to setMany in backing store`,
482 cause
483 })
484 ),
485 Effect.asVoid
486 )
487 } catch (cause) {
488 return Effect.fail(
489 new PersistenceError({
490 message: `Failed to serialize values into backing store`,
491 cause
492 })
493 )
494 }
495 }),
496 remove: (key) =>
497 sql`DELETE FROM ${table} WHERE id = ${key}`.pipe(
498 Effect.mapError((cause) =>
499 new PersistenceError({
500 message: `Failed to remove key ${key} from backing store`,
501 cause
502 })
503 ),
504 Effect.asVoid
505 ),
506 clear: sql`DELETE FROM ${table}`.pipe(
507 Effect.mapError((cause) =>
508 new PersistenceError({
509 message: `Failed to clear backing store`,
510 cause
511 })
512 ),
513 Effect.asVoid
514 )
515 })
516 })
517 })
518}))
519
520/**
521 * Provides SQL-backed persistence using a shared `effect_persistence` table.
522 *
523 * **Details**
524 *
525 * Rows are partitioned by `store_id` and store JSON-encoded values with
526 * optional expiration timestamps.
527 *
528 * @stability unstable
529 * @category layers
530 * @since 4.0.0
531 */
532export const layerBackingSql: Layer.Layer<
533 BackingPersistence,
534 never,
535 SqlClient.SqlClient
536> = Layer.effect(BackingPersistence)(Effect.gen(function*() {
537 const sql = (yield* SqlClient.SqlClient).withoutTransforms()
538 const table = sql("effect_persistence")
539 yield* sql.onDialectOrElse({
540 mysql: () =>
541 sql`
542 CREATE TABLE IF NOT EXISTS ${table} (
543 store_id VARCHAR(191) NOT NULL,
544 id VARCHAR(191) NOT NULL,
545 value TEXT NOT NULL,
546 expires BIGINT,
547 PRIMARY KEY (store_id, id)
548 )
549 `,
550 pg: () =>
551 sql`
552 CREATE TABLE IF NOT EXISTS ${table} (
553 store_id TEXT NOT NULL,
554 id TEXT NOT NULL,
555 value TEXT NOT NULL,
556 expires BIGINT,
557 PRIMARY KEY (store_id, id)
558 )
559 `,
560 mssql: () =>
561 sql`
562 IF NOT EXISTS (SELECT * FROM sysobjects WHERE name=${"effect_persistence"} AND xtype='U')
563 CREATE TABLE ${table} (
564 store_id NVARCHAR(191) NOT NULL,
565 id NVARCHAR(191) NOT NULL,
566 value NVARCHAR(MAX) NOT NULL,
567 expires BIGINT,
568 PRIMARY KEY (store_id, id)
569 )
570 `,
571 // sqlite
572 orElse: () =>
573 sql`
574 CREATE TABLE IF NOT EXISTS ${table} (
575 store_id TEXT NOT NULL,
576 id TEXT NOT NULL,
577 value TEXT NOT NULL,
578 expires INTEGER,
579 PRIMARY KEY (store_id, id)
580 )
581 `
582 }).pipe(Effect.orDie)
583
584 yield* sql.onDialectOrElse({
585 pg: () =>
586 sql`CREATE INDEX IF NOT EXISTS effect_persistence_expires_idx ON ${table} (expires) WHERE expires IS NOT NULL`
587 .pipe(Effect.orDie, Effect.asVoid),
588 mysql: () =>
589 Effect.gen(function*() {
590 const indexExists = sql<{ readonly count: number }>`
591 SELECT COUNT(*) AS count FROM information_schema.statistics
592 WHERE table_schema = DATABASE()
593 AND table_name = 'effect_persistence'
594 AND index_name = 'effect_persistence_expires_idx'
595 `.pipe(
596 Effect.map((rows) => Number(rows[0].count) > 0)
597 )
598
599 yield* sql`CREATE INDEX effect_persistence_expires_idx ON ${table} (expires)`.pipe(
600 Effect.catch((error) => Effect.flatMap(indexExists, (exists) => exists ? Effect.void : Effect.fail(error)))
601 )
602 }).pipe(Effect.orDie),
603 mssql: () =>
604 Effect.gen(function*() {
605 const indexExists = sql<{ readonly count: number }>`
606 SELECT COUNT(*) AS count FROM sys.indexes
607 WHERE name = N'effect_persistence_expires_idx'
608 AND object_id = OBJECT_ID(N'effect_persistence')
609 `.pipe(
610 Effect.map((rows) => Number(rows[0].count) > 0)
611 )
612
613 yield* sql`CREATE INDEX effect_persistence_expires_idx ON ${table} (expires) WHERE expires IS NOT NULL`.pipe(
614 Effect.catch((error) => Effect.flatMap(indexExists, (exists) => exists ? Effect.void : Effect.fail(error)))
615 )
616 }).pipe(Effect.orDie),
617 // sqlite
618 orElse: () =>
619 sql`CREATE INDEX IF NOT EXISTS effect_persistence_expires_idx ON ${table} (expires) WHERE expires IS NOT NULL`
620 .pipe(Effect.orDie, Effect.asVoid)
621 })
622
623 const cleanupBatchDelay = Duration.millis(10)
624 const cleanupInterval = Duration.minutes(5)
625
626 const deleteExpiredBatch = sql.onDialectOrElse({
627 pg: () => (expiresAtOrBefore: number) =>
628 sql<{ readonly count: number }>`
629 WITH deleted_entries AS (
630 DELETE FROM ${table}
631 WHERE ctid IN (
632 SELECT ctid FROM ${table}
633 WHERE expires IS NOT NULL AND expires <= ${expiresAtOrBefore}
634 LIMIT ${sql.literal(String(sqlCleanupBatchSize))}
635 )
636 RETURNING 1
637 )
638 SELECT COUNT(*)::INT AS count FROM deleted_entries
639 `.pipe(Effect.map((rows) => rows[0].count)),
640 mysql: () =>
641 Effect.fnUntraced(
642 function*(expiresAtOrBefore: number) {
643 const connection = yield* sql.reserve
644 const [statement, parameters] = sql`
645 DELETE FROM ${table}
646 WHERE expires IS NOT NULL AND expires <= ${expiresAtOrBefore}
647 LIMIT ${sql.literal(String(sqlCleanupBatchSize))}
648 `.compile()
649 yield* connection.execute(statement, parameters, undefined)
650 const rows = yield* connection.executeValues("SELECT ROW_COUNT()", [])
651 return Number(rows[0][0])
652 },
653 Effect.scoped
654 ),
655 mssql: () => (expiresAtOrBefore: number) =>
656 sql<{ readonly store_id: string }>`
657 WITH expired_entries AS (
658 SELECT TOP ${sql.literal(String(sqlCleanupBatchSize))} store_id, id FROM ${table}
659 WITH (UPDLOCK, READPAST, READCOMMITTEDLOCK)
660 WHERE expires IS NOT NULL AND expires <= ${expiresAtOrBefore}
661 )
662 DELETE persistence
663 OUTPUT DELETED.store_id
664 FROM ${table} AS persistence
665 INNER JOIN expired_entries
666 ON persistence.store_id = expired_entries.store_id
667 AND persistence.id = expired_entries.id
668 `.pipe(Effect.map((deletedEntries) => deletedEntries.length)),
669 // Some sqlite clients do not support interactive transactions, so use one bounded statement.
670 orElse: () => (expiresAtOrBefore: number) =>
671 sql<{ readonly deleted: number }>`
672 DELETE FROM ${table}
673 WHERE rowid IN (
674 SELECT rowid FROM ${table}
675 WHERE expires IS NOT NULL AND expires <= ${expiresAtOrBefore}
676 LIMIT ${sql.literal(String(sqlCleanupBatchSize))}
677 )
678 RETURNING 1 AS deleted
679 `.pipe(Effect.map((deletedEntries) => deletedEntries.length))
680 })
681
682 const deleteExpired = Effect.gen(function*() {
683 const expiresAtOrBefore = yield* Clock.currentTimeMillis
684 return yield* deleteExpiredBatch(expiresAtOrBefore).pipe(
685 Effect.repeat({
686 while: (deletedCount) => deletedCount === sqlCleanupBatchSize,
687 schedule: Schedule.spaced(cleanupBatchDelay)
688 })
689 )
690 })
691
692 yield* deleteExpired.pipe(
693 Effect.catch((cause) => Effect.logWarning("Failed to clean up expired persistence entries", cause)),
694 Effect.repeat(Schedule.spaced(cleanupInterval)),
695 Effect.forkScoped
696 )
697
698 type UpsertFn = (
699 entries: Array<{ store_id: string; id: string; value: string; expires: number | null }>
700 ) => Effect.Effect<unknown, SqlError>
701
702 const upsert = sql.onDialectOrElse({
703 pg: (): UpsertFn => (entries) =>
704 sql`
705 INSERT INTO ${table} ${sql.insert(entries)}
706 ON CONFLICT (store_id, id) DO UPDATE SET value=EXCLUDED.value, expires=EXCLUDED.expires
707 `.unprepared,
708 mysql: (): UpsertFn => (entries) =>
709 sql`
710 INSERT INTO ${table} ${sql.insert(entries)}
711 ON DUPLICATE KEY UPDATE value=VALUES(value), expires=VALUES(expires)
712 `.unprepared,
713 mssql: (): UpsertFn => (entries) =>
714 Effect.forEach(
715 entries,
716 (entry) =>
717 sql`
718 MERGE ${table} AS target
719 USING (SELECT ${entry.store_id} AS store_id, ${entry.id} AS id, ${entry.value} AS value, ${entry.expires} AS expires) AS source
720 ON target.store_id = source.store_id AND target.id = source.id
721 WHEN MATCHED THEN UPDATE SET value = source.value, expires = source.expires
722 WHEN NOT MATCHED THEN INSERT (store_id, id, value, expires)
723 VALUES (source.store_id, source.id, source.value, source.expires);
724 `,
725 { discard: true }
726 ),
727 // sqlite
728 orElse: (): UpsertFn => (entries) =>
729 sql`
730 INSERT INTO ${table} ${sql.insert(entries)}
731 ON CONFLICT(store_id, id) DO UPDATE SET value=excluded.value, expires=excluded.expires
732 `.unprepared
733 })
734
735 return BackingPersistence.of({
736 make: Effect.fnUntraced(function*(storeId) {
737 const clock = yield* Clock.Clock
738
739 return identity<BackingPersistenceStore>({
740 get: (key) =>
741 sql<
742 { value: string }
743 >`SELECT value FROM ${table} WHERE store_id = ${storeId} AND id = ${key} AND (expires IS NULL OR expires > ${clock.currentTimeMillisUnsafe()})`
744 .pipe(
745 Effect.mapError((cause) =>
746 new PersistenceError({
747 message: `Failed to get key ${key} from backing store`,
748 cause
749 })
750 ),
751 Effect.flatMap((rows) => {
752 if (rows.length === 0) {
753 return Effect.undefined
754 }
755 try {
756 return Effect.succeed(JSON.parse(rows[0].value))
757 } catch (cause) {
758 return Effect.fail(
759 new PersistenceError({
760 message: `Failed to parse value for key ${key} from backing store`,
761 cause
762 })
763 )
764 }
765 })
766 ),
767 getMany: (keys) =>
768 sql<{ id: string; value: string }>`SELECT id, value FROM ${table} WHERE store_id = ${storeId} AND ${
769 sql.in("id", keys)
770 } AND (expires IS NULL OR expires > ${clock.currentTimeMillisUnsafe()})`.unprepared.pipe(
771 Effect.mapError((cause) =>
772 new PersistenceError({
773 message: `Failed to getMany from backing store`,
774 cause
775 })
776 ),
777 Effect.flatMap((rows) => {
778 const values = new Map<string, object>()
779 for (let i = 0; i < rows.length; i++) {
780 const row = rows[i]
781 try {
782 values.set(row.id, JSON.parse(row.value))
783 } catch {
784 // ignore
785 }
786 }
787 return Effect.succeed(keys.map((key) => values.get(key)) as Arr.NonEmptyArray<object | undefined>)
788 })
789 ),
790 set: (key, value, ttl) =>
791 Effect.suspend(() => {
792 try {
793 return upsert([{
794 store_id: storeId,
795 id: key,
796 value: JSON.stringify(value),
797 expires: unsafeTtlToExpires(clock, ttl)
798 }])
799 .pipe(
800 Effect.mapError((cause) =>
801 new PersistenceError({
802 message: `Failed to set key ${key} in backing store`,
803 cause
804 })
805 ),
806 Effect.asVoid
807 )
808 } catch (cause) {
809 return Effect.fail(
810 new PersistenceError({
811 message: `Failed to serialize value for key ${key} to backing store`,
812 cause
813 })
814 )
815 }
816 }),
817 setMany: (entries) =>
818 Effect.suspend(() => {
819 try {
820 const encoded = entries.map(([key, value, ttl]) => ({
821 store_id: storeId,
822 id: key,
823 value: JSON.stringify(value),
824 expires: unsafeTtlToExpires(clock, ttl)
825 }))
826 return upsert(encoded).pipe(
827 Effect.mapError((cause) =>
828 new PersistenceError({
829 message: `Failed to setMany in backing store`,
830 cause
831 })
832 ),
833 Effect.asVoid
834 )
835 } catch (cause) {
836 return Effect.fail(
837 new PersistenceError({
838 message: `Failed to serialize values into backing store`,
839 cause
840 })
841 )
842 }
843 }),
844 remove: (key) =>
845 sql`DELETE FROM ${table} WHERE store_id = ${storeId} AND id = ${key}`.pipe(
846 Effect.mapError((cause) =>
847 new PersistenceError({
848 message: `Failed to remove key ${key} from backing store`,
849 cause
850 })
851 ),
852 Effect.asVoid
853 ),
854 clear: sql`DELETE FROM ${table} WHERE store_id = ${storeId}`.pipe(
855 Effect.mapError((cause) =>
856 new PersistenceError({
857 message: `Failed to clear backing store`,
858 cause
859 })
860 ),
861 Effect.asVoid
862 )
863 })
864 })
865 })
866}))
867
868/**
869 * Provides Redis-backed persistence.
870 *
871 * **Details**
872 *
873 * Each store id is used as a key prefix, values are JSON-encoded, and finite
874 * TTLs are stored with Redis expiration.
875 *
876 * @stability unstable
877 * @category layers
878 * @since 4.0.0
879 */
880export const layerBackingRedis: Layer.Layer<
881 BackingPersistence,
882 never,
883 Redis.Redis
884> = Layer.effect(BackingPersistence)(Effect.gen(function*() {
885 const redis = yield* Redis.Redis
886 const setMany = redis.eval(setManyRedis)
887
888 return BackingPersistence.of({
889 make: (prefix) =>
890 Effect.sync(() => {
891 const prefixed = (key: string) => `${prefix}:${key}`
892 const parse = (str: string | null) => {
893 if (str === null) {
894 return Effect.undefined
895 }
896 try {
897 return Effect.succeed(JSON.parse(str))
898 } catch (cause) {
899 return Effect.fail(
900 new PersistenceError({
901 message: `Failed to parse value from Redis`,
902 cause
903 })
904 )
905 }
906 }
907 return identity<BackingPersistenceStore>({
908 get: (key) =>
909 Effect.flatMap(
910 Effect.mapError(
911 redis.send<string>("GET", prefixed(key)),
912 ({ cause }) =>
913 new PersistenceError({
914 message: `Failed to get key ${key} from Redis`,
915 cause
916 })
917 ),
918 parse
919 ),
920 getMany: (keys) =>
921 Effect.flatMap(
922 Effect.mapError(
923 redis.send<Arr.NonEmptyArray<string>>("mget", ...keys.map(prefixed)),
924 ({ cause }) =>
925 new PersistenceError({
926 message: `Failed to getMany from Redis`,
927 cause
928 })
929 ),
930 (values) => {
931 const out = new Array<object | undefined>(keys.length) as Arr.NonEmptyArray<object | undefined>
932 for (let i = 0; i < keys.length; i++) {
933 const value = values[i]
934 try {
935 out[i] = value === null ? undefined : JSON.parse(value)
936 } catch {
937 // TODO: remove bad entries?
938 out[i] = undefined
939 }
940 }
941 return Effect.succeed(out)
942 }
943 ),
944 set: (key, value, ttl) =>
945 Effect.mapError(
946 ttl === undefined
947 ? redis.send("SET", prefixed(key), JSON.stringify(value))
948 : redis.send(
949 "SET",
950 prefixed(key),
951 JSON.stringify(value),
952 "PX",
953 String(Math.ceil(Duration.toMillis(ttl)))
954 ),
955 ({ cause }) =>
956 new PersistenceError({
957 message: `Failed to set key ${key} in Redis`,
958 cause
959 })
960 ),
961 setMany: (entries) =>
962 Effect.suspend(() => {
963 const sets = new Map<string, string>()
964 const expires = new Map<string, number>()
965 for (const [key, value, ttl] of entries) {
966 const pkey = prefixed(key)
967 sets.set(pkey, JSON.stringify(value))
968 if (ttl) {
969 expires.set(pkey, Math.ceil(Duration.toMillis(ttl)))
970 }
971 }
972 return Effect.mapError(
973 setMany({ sets, expires }),
974 ({ cause }) =>
975 new PersistenceError({
976 message: `Failed to setMany in Redis`,
977 cause
978 })
979 )
980 }),
981 remove: (key) =>
982 Effect.mapError(
983 redis.send("DEL", prefixed(key)),
984 ({ cause }) => new PersistenceError({ message: `Failed to remove key ${key} from Redis`, cause })
985 ),
986 clear: redis.send<Array<string>>("KEYS", `${prefix}:*`).pipe(
987 Effect.flatMap((keys) => keys.length === 0 ? Effect.void : redis.send("DEL", ...keys)),
988 Effect.mapError(({ cause }) =>
989 new PersistenceError({
990 message: `Failed to clear keys from Redis`,
991 cause
992 })
993 )
994 )
995 })
996 })
997 })
998}))
999
1000const setManyRedis = Redis.script(
1001 (options: {
1002 readonly sets: Map<string, string>
1003 readonly expires: Map<string, number>
1004 }) => [
1005 ...options.sets.keys(),
1006 ...options.expires.keys(),
1007 options.sets.size,
1008 options.expires.size,
1009 ...options.sets.values(),
1010 ...options.expires.values()
1011 ],
1012 {
1013 numberOfKeys: (options) => options.sets.size + options.expires.size,
1014 lua: `
1015local num_sets = tonumber(ARGV[1])
1016local num_expires = tonumber(ARGV[2])
1017local index = 3
1018
1019for i = 1, num_sets do
1020 local key = KEYS[i]
1021 local value = ARGV[index]
1022 redis.call("SET", key, value)
1023 index = index + 1
1024end
1025
1026for i = 1, num_expires do
1027 local key = KEYS[num_sets + i]
1028 local expire = tonumber(ARGV[index])
1029 redis.call("PEXPIRE", key, expire)
1030 index = index + 1
1031end
1032`
1033 }
1034)
1035
1036/**
1037 * Provides `BackingPersistence` using a `KeyValueStore`.
1038 *
1039 * **Details**
1040 *
1041 * Each store id becomes a key prefix, and values are stored as JSON with
1042 * optional expiration timestamps.
1043 *
1044 * @stability unstable
1045 * @category layers
1046 * @since 4.0.0
1047 */
1048export const layerBackingKvs: Layer.Layer<
1049 BackingPersistence,
1050 never,
1051 KeyValueStore.KeyValueStore
1052> = Layer.effect(BackingPersistence)(Effect.gen(function*() {
1053 const backing = yield* KeyValueStore.KeyValueStore
1054 const clock = yield* Clock.Clock
1055 return BackingPersistence.of({
1056 make: (storeId) =>
1057 Effect.sync(() => {
1058 const store = KeyValueStore.prefix(backing, storeId)
1059 const get = (key: string) =>
1060 Effect.flatMap(
1061 Effect.mapError(
1062 store.get(key),
1063 (error) =>
1064 new PersistenceError({
1065 message: `Failed to get key ${key} from backing store`,
1066 cause: error
1067 })
1068 ),
1069 (str) => {
1070 if (str === undefined) {
1071 return Effect.undefined
1072 }
1073 try {
1074 const parsed = JSON.parse(str)
1075 if (!Array.isArray(parsed)) return Effect.undefined
1076 const [value, expires] = parsed as [object, number | null]
1077 if (expires !== null && expires <= clock.currentTimeMillisUnsafe()) {
1078 return Effect.as(Effect.ignore(store.remove(key)), undefined)
1079 }
1080 return Effect.succeed(value)
1081 } catch (cause) {
1082 return Effect.fail(
1083 new PersistenceError({
1084 message: `Failed to parse value for key ${key} from backing store`,
1085 cause
1086 })
1087 )
1088 }
1089 }
1090 )
1091 return identity<BackingPersistenceStore>({
1092 get,
1093 getMany: (keys) => Effect.forEach(keys, get, { concurrency: "unbounded" }),
1094 set: (key, value, ttl) =>
1095 Effect.suspend(() => {
1096 try {
1097 return Effect.mapError(
1098 store.set(key, JSON.stringify([value, unsafeTtlToExpires(clock, ttl)])),
1099 (cause) =>
1100 new PersistenceError({
1101 message: `Failed to set key ${key} in backing store`,
1102 cause
1103 })
1104 )
1105 } catch (cause) {
1106 return Effect.fail(
1107 new PersistenceError({
1108 message: `Failed to serialize value for key ${key} to backing store`,
1109 cause
1110 })
1111 )
1112 }
1113 }),
1114 setMany: (entries) =>
1115 Effect.forEach(entries, ([key, value, ttl]) => {
1116 const expires = unsafeTtlToExpires(clock, ttl)
1117 const encoded = JSON.stringify([value, expires])
1118 return store.set(key, encoded)
1119 }, { concurrency: "unbounded", discard: true }).pipe(
1120 Effect.mapError((cause) =>
1121 new PersistenceError({
1122 message: `Failed to setMany in backing store`,
1123 cause
1124 })
1125 )
1126 ),
1127 remove: (key) =>
1128 Effect.mapError(
1129 store.remove(key),
1130 (cause) => new PersistenceError({ message: `Failed to remove key ${key} from backing store`, cause })
1131 ),
1132 clear: Effect.mapError(store.clear, (cause) =>
1133 new PersistenceError({ message: `Failed to clear backing store`, cause }))
1134 })
1135 })
1136 })
1137}))
1138
1139/**
1140 * Provides `Persistence` backed by the current `KeyValueStore`.
1141 *
1142 * @stability unstable
1143 * @category layers
1144 * @since 4.0.0
1145 */
1146export const layerKvs: Layer.Layer<Persistence, never, KeyValueStore.KeyValueStore> = layer.pipe(
1147 Layer.provide(layerBackingKvs)
1148)
1149
1150/**
1151 * Provides `Persistence` backed by process-local in-memory storage.
1152 *
1153 * @stability unstable
1154 * @category layers
1155 * @since 4.0.0
1156 */
1157export const layerMemory: Layer.Layer<Persistence> = layer.pipe(
1158 Layer.provide(layerBackingMemory)
1159)
1160
1161/**
1162 * Provides `Persistence` backed by the current `Redis` service.
1163 *
1164 * @stability unstable
1165 * @category layers
1166 * @since 4.0.0
1167 */
1168export const layerRedis: Layer.Layer<Persistence, never, Redis.Redis> = layer.pipe(
1169 Layer.provide(layerBackingRedis)
1170)
1171
1172/**
1173 * Provides `Persistence` backed by SQL with one table per store id.
1174 *
1175 * @stability unstable
1176 * @category layers
1177 * @since 4.0.0
1178 */
1179export const layerSqlMultiTable: Layer.Layer<Persistence, never, SqlClient.SqlClient> = layer.pipe(
1180 Layer.provide(layerBackingSqlMultiTable)
1181)
1182
1183/**
1184 * Provides `Persistence` backed by SQL using a shared persistence table.
1185 *
1186 * @stability unstable
1187 * @category layers
1188 * @since 4.0.0
1189 */
1190export const layerSql: Layer.Layer<Persistence, never, SqlClient.SqlClient> = layer.pipe(
1191 Layer.provide(layerBackingSql)
1192)
1193
1194/**
1195 * Converts a TTL to an absolute expiration timestamp in milliseconds.
1196 *
1197 * **Details**
1198 *
1199 * Returns `null` for no TTL and uses `clock.currentTimeMillisUnsafe`, so it is
1200 * intended for backing-store internals.
1201 *
1202 * @stability unstable
1203 * @category converting
1204 * @since 4.0.0
1205 */
1206export const unsafeTtlToExpires = (clock: Clock.Clock, ttl: Duration.Duration | undefined): number | null =>
1207 ttl ? clock.currentTimeMillisUnsafe() + Duration.toMillis(ttl) : null