1/**
2 * Node.js Redis integration backed by `redis` (node-redis).
3 *
4 * This module creates a scoped `node-redis` client and exposes it in two forms:
5 * the generic `Redis` service and the {@link NodeRedis} service for direct
6 * access to the underlying client. `layer` accepts node-redis client options
7 * directly, while `layerConfig` reads them from Effect config. `node-redis`
8 * connects explicitly, so layer construction can fail with a `RedisError`.
9 * Both layers close the client when the layer scope ends.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Config from "effect/Config"
15import * as Context from "effect/Context"
16import * as Deferred from "effect/Deferred"
17import * as Effect from "effect/Effect"
18import * as Fn from "effect/Function"
19import * as Layer from "effect/Layer"
20import * as Redis from "effect/persistence/Redis"
21import { createClient, SocketTimeoutError } from "redis"
22
23type NodeRedisClient = ReturnType<typeof createClient>
24type NodeRedisClientOptions = NonNullable<Parameters<typeof createClient>[0]>
25
26/**
27 * Service tag for the Node Redis integration, exposing the underlying
28 * `node-redis` client and a `use` helper that maps client failures to
29 * `RedisError`.
30 *
31 * @stability unstable
32 * @category services
33 * @since 4.0.0
34 */
35export class NodeRedis extends Context.Service<NodeRedis, {
36 readonly client: NodeRedisClient
37 readonly use: <A>(f: (client: NodeRedisClient) => Promise<A>) => Effect.Effect<A, Redis.RedisError>
38}>()("@effect/platform-node/NodeRedis") {}
39
40const make = Effect.fnUntraced(function*(
41 options?: NodeRedisClientOptions
42) {
43 let ready = false
44 const socket = options?.socket
45 const client = yield* Effect.acquireRelease(
46 Effect.sync((): NodeRedisClient =>
47 createClient({
48 ...options,
49 socket: socket?.reconnectStrategy === undefined
50 ? {
51 ...socket,
52 reconnectStrategy: (retries, cause) => {
53 if (!ready) return cause
54 if (cause instanceof SocketTimeoutError) return false
55 const jitter = Math.floor(Math.random() * 200)
56 const delay = Math.min(2 ** retries * 50, 2000)
57 return delay + jitter
58 }
59 }
60 : socket
61 })
62 ),
63 (client) => Effect.ignoreCause(Effect.promise(() => client.close()))
64 )
65 client.once("ready", () => {
66 ready = true
67 })
68
69 // node-redis rethrows `error` events that have no listener, which would crash
70 // the process on a transient socket failure. Command failures are still
71 // reported as `RedisError`.
72 const runSync = Effect.runSyncWith(yield* Effect.context<never>())
73 client.on("error", (cause) => {
74 runSync(Effect.logWarning("NodeRedis client error", cause))
75 })
76
77 yield* Effect.tryPromise({
78 try: () => client.connect(),
79 catch: (cause) => new Redis.RedisError({ cause })
80 })
81
82 const use = <A>(f: (client: NodeRedisClient) => Promise<A>) =>
83 Effect.tryPromise({
84 try: () => f(client),
85 catch: (cause) => new Redis.RedisError({ cause })
86 })
87
88 const redis = yield* Redis.make({
89 send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) =>
90 Effect.tryPromise({
91 try: () => client.sendCommand([command, ...args]) as Promise<A>,
92 catch: (cause) => new Redis.RedisError({ cause })
93 }),
94 subscribe: (channel, onMessage) =>
95 Effect.gen(function*() {
96 const terminal = yield* Deferred.make<void, Redis.RedisError>()
97 yield* Effect.acquireRelease(
98 Effect.tryPromise({
99 try: async () => {
100 let subscriberReady = false
101 const subscriber = client.duplicate(
102 socket?.reconnectStrategy === undefined
103 ? {
104 socket: {
105 ...socket,
106 reconnectStrategy: (retries, cause) => {
107 if (!subscriberReady) return cause
108 if (cause instanceof SocketTimeoutError) return false
109 const jitter = Math.floor(Math.random() * 200)
110 const delay = Math.min(2 ** retries * 50, 2000)
111 return delay + jitter
112 }
113 }
114 }
115 : {}
116 )
117 subscriber.once("ready", () => {
118 subscriberReady = true
119 })
120 subscriber.on("error", (cause) => {
121 runSync(Effect.logWarning("NodeRedis subscriber error", cause))
122 if (!subscriber.isOpen) {
123 runSync(Deferred.fail(terminal, new Redis.RedisError({ cause })))
124 }
125 })
126 try {
127 await subscriber.connect()
128 await subscriber.subscribe(channel, (message, channel) => {
129 onMessage({ channel, message })
130 })
131 return subscriber
132 } catch (cause) {
133 if (subscriber.isOpen) subscriber.destroy()
134 throw cause
135 }
136 },
137 catch: (cause) => new Redis.RedisError({ cause })
138 }),
139 (subscriber) =>
140 Effect.sync(() => {
141 if (subscriber.isOpen) subscriber.destroy()
142 })
143 )
144 return Deferred.await(terminal)
145 })
146 })
147
148 const nodeRedis = Fn.identity<NodeRedis["Service"]>({
149 client,
150 use
151 })
152
153 return Context.make(NodeRedis, nodeRedis).pipe(
154 Context.add(Redis.Redis, redis)
155 )
156})
157
158/**
159 * Provides `Redis` and `NodeRedis` services backed by a `node-redis` client
160 * created with the supplied options, connected when the layer is built and
161 * closed when the layer scope ends.
162 *
163 * **Details**
164 *
165 * By default, the initial connection fails on its first connection error. A
166 * caller-supplied `socket.reconnectStrategy` is used instead when present. Once
167 * the client has emitted `ready`, the default reconnect strategy uses
168 * node-redis' exponential backoff and stops on socket timeouts.
169 *
170 * Scope finalization calls `close()`, which waits for in-flight commands,
171 * including blocking commands, and can therefore delay scope closure.
172 *
173 * @stability unstable
174 * @category layers
175 * @since 4.0.0
176 */
177export const layer = (
178 options?: NodeRedisClientOptions | undefined
179): Layer.Layer<Redis.Redis | NodeRedis, Redis.RedisError> => Layer.effectContext(make(options))
180
181/**
182 * Provides `Redis` and `NodeRedis` services from `Config`-backed node-redis
183 * client options, connecting the client when the layer is built and closing it
184 * when the layer scope ends.
185 *
186 * **Details**
187 *
188 * By default, the initial connection fails on its first connection error. A
189 * caller-supplied `socket.reconnectStrategy` is used instead when present. Once
190 * the client has emitted `ready`, the default reconnect strategy uses
191 * node-redis' exponential backoff and stops on socket timeouts.
192 *
193 * Scope finalization calls `close()`, which waits for in-flight commands,
194 * including blocking commands, and can therefore delay scope closure.
195 *
196 * @stability unstable
197 * @category layers
198 * @since 4.0.0
199 */
200export const layerConfig: (
201 options: Config.Wrap<NodeRedisClientOptions>
202) => Layer.Layer<Redis.Redis | NodeRedis, Redis.RedisError | Config.ConfigError> = (
203 options: Config.Wrap<NodeRedisClientOptions>
204): Layer.Layer<Redis.Redis | NodeRedis, Redis.RedisError | Config.ConfigError> =>
205 Layer.effectContext(
206 Config.unwrap(options).pipe(
207 Effect.flatMap(make)
208 )
209 )