1/**
2 * Redis support shared by persistence modules.
3 *
4 * This module defines a `Redis` service that can send Redis commands, subscribe
5 * to pub/sub channels, and run Lua scripts. It does not create a Redis client
6 * itself; callers provide the client-specific operations. The module also
7 * provides helpers for describing Lua scripts, loading them once, and running
8 * them later by their cached Redis id.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Cache from "../Cache.ts"
14import type * as Cause from "../Cause.ts"
15import * as Context from "../Context.ts"
16import * as Duration from "../Duration.ts"
17import * as Effect from "../Effect.ts"
18import * as Equal from "../Equal.ts"
19import * as Exit from "../Exit.ts"
20import { constant, identity } from "../Function.ts"
21import * as Hash from "../Hash.ts"
22import * as Queue from "../Queue.ts"
23import * as Schema from "../Schema.ts"
24import * as Scope from "../Scope.ts"
25
26/**
27 * A message received from a Redis pub/sub channel.
28 *
29 * @stability unstable
30 * @category models
31 * @since 4.0.0
32 */
33export interface RedisMessage {
34 readonly channel: string
35 readonly message: string
36}
37
38/**
39 * Service for sending Redis commands, subscribing to channels, and evaluating
40 * cached Lua scripts.
41 *
42 * @stability unstable
43 * @category services
44 * @since 4.0.0
45 */
46export class Redis extends Context.Service<Redis, {
47 readonly send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<A, RedisError>
48
49 /**
50 * Subscribes to a Redis pub/sub channel for the lifetime of the current
51 * scope. Node and Deno subscribers reconnect and re-subscribe after an
52 * interruption, so messages published during recovery appear as delivery
53 * gaps. Bun subscribers do not reconnect: a dropped connection fails the
54 * dequeue, and the caller must subscribe again.
55 */
56 readonly subscribe: (
57 channel: string
58 ) => Effect.Effect<Queue.Dequeue<RedisMessage, RedisError>, never, Scope.Scope>
59
60 readonly eval: <
61 Config extends {
62 readonly params: ReadonlyArray<unknown>
63 readonly result: unknown
64 }
65 >(script: Script<Config>) => (...params: Config["params"]) => Effect.Effect<Config["result"], RedisError>
66}>()("effect/persistence/Redis") {}
67
68/**
69 * Creates a `Redis` service from raw command and subscription operations.
70 *
71 * **Details**
72 *
73 * Lua scripts are loaded through `SCRIPT LOAD`, cached, and then invoked with
74 * `EVALSHA`.
75 *
76 * @stability unstable
77 * @category constructors
78 * @since 4.0.0
79 */
80export const make = Effect.fnUntraced(function*(
81 options: {
82 readonly send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) => Effect.Effect<A, RedisError>
83 readonly subscribe: (
84 channel: string,
85 onMessage: (message: RedisMessage) => void
86 ) => Effect.Effect<Effect.Effect<void, RedisError>, RedisError, Scope.Scope>
87 }
88) {
89 const scriptCache = yield* Cache.makeWith(
90 (script: Script<any>) => options.send<string>("SCRIPT", "LOAD", script.lua),
91 {
92 capacity: Number.POSITIVE_INFINITY,
93 timeToLive: (exit) => Exit.isSuccess(exit) ? Duration.infinity : Duration.zero
94 }
95 )
96
97 const eval_ = <
98 Config extends {
99 readonly params: ReadonlyArray<unknown>
100 readonly result: unknown
101 }
102 >(
103 script: Script<Config>
104 ) =>
105 (...params: Config["params"]): Effect.Effect<Config["result"], RedisError> => {
106 const evalSha = (sha: string) =>
107 options.send<Config["result"]>(
108 "EVALSHA",
109 sha,
110 script.numberOfKeys(...params).toString(),
111 ...script.params(...params).map((param) => String(param))
112 )
113 return Cache.get(scriptCache, script).pipe(
114 Effect.flatMap(evalSha),
115 Effect.catchIf(
116 (error) => String(error.cause).includes("NOSCRIPT"),
117 () => Cache.refresh(scriptCache, script).pipe(Effect.flatMap(evalSha))
118 )
119 )
120 }
121
122 const subscribe = Effect.fnUntraced(function*(channel: string) {
123 const queue = yield* Queue.unbounded<RedisMessage, RedisError>()
124 yield* Scope.addFinalizer(yield* Effect.scope, Queue.shutdown(queue))
125 const onFailure = (cause: Cause.Cause<RedisError>) => Queue.failCause(queue, cause).pipe(Effect.asVoid)
126 yield* options.subscribe(channel, (message) => {
127 Queue.offerUnsafe(queue, message)
128 }).pipe(
129 Effect.flatMap((listen) => Effect.forkScoped(Effect.catchCause(listen, onFailure))),
130 Effect.catchCause(onFailure)
131 )
132 return queue
133 })
134
135 return identity<Redis["Service"]>({
136 send: options.send,
137 subscribe,
138 eval: eval_
139 })
140})
141
142type ErrorTypeId = "~effect/persistence/Redis/RedisError"
143const ErrorTypeId: ErrorTypeId = "~effect/persistence/Redis/RedisError"
144
145/**
146 * Error raised by Redis command or script execution.
147 *
148 * @stability unstable
149 * @category errors
150 * @since 4.0.0
151 */
152export class RedisError extends Schema.Error<RedisError>(ErrorTypeId)({
153 _tag: Schema.tag("RedisError"),
154 cause: Schema.Defect()
155}) {
156 /**
157 * Marks this value as a Redis persistence error for runtime guards.
158 *
159 * @stability unstable
160 * @since 4.0.0
161 */
162 readonly [ErrorTypeId]: ErrorTypeId = ErrorTypeId
163}
164
165type ScriptTypeId = "~effect/persistence/Redis/Script"
166const ScriptTypeId: ScriptTypeId = "~effect/persistence/Redis/Script"
167
168/**
169 * Typed descriptor for a Redis Lua script.
170 *
171 * **Details**
172 *
173 * It defines the Lua source, parameter-to-argument mapping, Redis key count,
174 * and result type used by `Redis.eval`.
175 *
176 * @stability unstable
177 * @category scripting
178 * @since 4.0.0
179 */
180export interface Script<
181 Config extends {
182 readonly params: ReadonlyArray<unknown>
183 readonly result: unknown
184 }
185> {
186 readonly [ScriptTypeId]: {
187 readonly params: Config["params"]
188 readonly result: Config["result"]
189 }
190 readonly lua: string
191 readonly params: (...params: Config["params"]) => ReadonlyArray<unknown>
192 readonly numberOfKeys: (...params: Config["params"]) => number
193
194 /**
195 * Set the return type of the script.
196 */
197 withReturnType<Result>(): Script<{
198 params: Config["params"]
199 result: Result
200 }>
201}
202
203const ScriptProto = {
204 [ScriptTypeId]: {
205 params: identity,
206 result: identity
207 },
208 withReturnType() {
209 return this
210 },
211 [Equal.symbol](that: unknown): boolean {
212 return this === that
213 },
214 [Hash.symbol](): number {
215 return Hash.random(this)
216 }
217}
218
219/**
220 * Constructs a typed Redis Lua script descriptor.
221 *
222 * **Details**
223 *
224 * The result type defaults to `void` and can be refined with
225 * `withReturnType`.
226 *
227 * @stability unstable
228 * @category scripting
229 * @since 4.0.0
230 */
231export const script = <Params extends ReadonlyArray<any>>(
232 f: (...params: Params) => ReadonlyArray<unknown>,
233 options: {
234 readonly lua: string
235 readonly numberOfKeys: number | ((...params: Params) => number)
236 }
237): Script<{
238 params: Params
239 result: void
240}> =>
241 Object.setPrototypeOf({
242 ...options,
243 params: f,
244 numberOfKeys: typeof options.numberOfKeys === "number" ? constant(options.numberOfKeys) : options.numberOfKeys
245 }, ScriptProto)