1import { assert, describe, it } from "@effect/vitest"
2import { Duration, Effect, Exit, Layer, Queue, Scope } from "effect"
3import { Persistence, Redis } from "effect/persistence"
4
5describe("Redis", () => {
6 it("preserves __proto__ as an own script option", () => {
7 const value = { polluted: true }
8 const script = Redis.script(() => [], {
9 ["__proto__"]: value,
10 lua: "return nil",
11 numberOfKeys: 0
12 } as any)
13
14 assert.isTrue(Object.hasOwn(script, "__proto__"))
15 assert.strictEqual((script as any)["__proto__"], value)
16 })
17
18 it.effect("clearing an empty persistence store succeeds", () => {
19 const redis = Redis.Redis.of({
20 send: <A>(command: string, ...args: ReadonlyArray<string>) => {
21 if (command.toUpperCase() === "KEYS") return Effect.succeed([] as unknown as A)
22 if (command.toUpperCase() === "DEL" && args.length === 0) {
23 return Effect.fail(new Redis.RedisError({ cause: "ERR wrong number of arguments for 'del' command" }))
24 }
25 return Effect.succeed(undefined as unknown as A)
26 },
27 subscribe: () => Queue.unbounded<Redis.RedisMessage, Redis.RedisError>(),
28 eval: () => () => Effect.die("unused")
29 })
30 return Effect.gen(function*() {
31 const backing = yield* Persistence.BackingPersistence
32 const store = yield* backing.make("empty")
33 yield* store.clear
34 }).pipe(
35 Effect.provide(Persistence.layerBackingRedis.pipe(Layer.provide(Layer.succeed(Redis.Redis, redis))))
36 )
37 })
38
39 it.effect("rounds fractional persistence TTLs up to whole milliseconds", () => {
40 const commands: Array<readonly [command: string, args: ReadonlyArray<string>]> = []
41 const scripts: Array<unknown> = []
42 const redis = Redis.Redis.of({
43 send: <A>(command: string, ...args: ReadonlyArray<string>) => {
44 commands.push([command, args])
45 return Effect.succeed(undefined as unknown as A)
46 },
47 subscribe: () => Queue.unbounded<Redis.RedisMessage, Redis.RedisError>(),
48 eval:
49 <Config extends { readonly params: ReadonlyArray<unknown>; readonly result: unknown }>() =>
50 (...params: Config["params"]) => {
51 scripts.push(params[0])
52 return Effect.succeed(undefined as Config["result"])
53 }
54 })
55 return Effect.gen(function*() {
56 const backing = yield* Persistence.BackingPersistence
57 const store = yield* backing.make("ttl")
58 const ttl = Duration.nanos(1_500_000n)
59
60 yield* store.set("single", {}, ttl)
61 yield* store.setMany([["batch", {}, ttl]])
62
63 assert.deepStrictEqual(
64 commands.filter(([command]) => command === "SET"),
65 [["SET", ["ttl:single", "{}", "PX", "2"]]]
66 )
67 assert.deepStrictEqual(scripts, [{
68 sets: new Map([["ttl:batch", "{}"]]),
69 expires: new Map([["ttl:batch", 2]])
70 }])
71 }).pipe(
72 Effect.provide(Persistence.layerBackingRedis.pipe(Layer.provide(Layer.succeed(Redis.Redis, redis))))
73 )
74 })
75
76 it.effect("retries script loading when SCRIPT LOAD fails", () =>
77 Effect.gen(function*() {
78 const commands: Array<readonly [command: string, args: ReadonlyArray<string>]> = []
79 let scriptLoadAttempts = 0
80 const redis = yield* Redis.make({
81 send: (command, ...args) =>
82 Effect.suspend(() => {
83 commands.push([command, args])
84 if (command === "SCRIPT") {
85 scriptLoadAttempts += 1
86 if (scriptLoadAttempts === 1) {
87 return Effect.fail(new Redis.RedisError({ cause: new Error("ERR transient script load failure") }))
88 }
89 return Effect.succeed("sha" as any)
90 }
91 return Effect.succeed("ok")
92 }),
93 subscribe: () => Effect.succeed(Effect.never)
94 })
95 const evalScript = redis.eval(
96 Redis.script((key: string) => [key], {
97 lua: "return KEYS[1]",
98 numberOfKeys: 1
99 }).withReturnType<string>()
100 )
101
102 const first = yield* Effect.result(evalScript("key"))
103 const second = yield* evalScript("key")
104
105 assert.strictEqual(first._tag, "Failure")
106 assert.strictEqual(second, "ok")
107 assert.deepStrictEqual(commands.map(([command]) => command), [
108 "SCRIPT",
109 "SCRIPT",
110 "EVALSHA"
111 ])
112 }))
113
114 it.effect("reloads and retries scripts when EVALSHA returns NOSCRIPT", () =>
115 Effect.gen(function*() {
116 const commands: Array<readonly [command: string, args: ReadonlyArray<string>]> = []
117 let evalShaAttempts = 0
118 const redis = yield* Redis.make({
119 send: (command, ...args) =>
120 Effect.suspend(() => {
121 commands.push([command, args])
122 if (command === "EVALSHA") {
123 evalShaAttempts += 1
124 }
125 if (command === "SCRIPT") {
126 return Effect.succeed("sha" as any)
127 }
128 if (command === "EVALSHA" && evalShaAttempts === 1) {
129 return Effect.fail(new Redis.RedisError({ cause: new Error("NOSCRIPT No matching script") }))
130 }
131 return Effect.succeed("ok")
132 }),
133 subscribe: () => Effect.succeed(Effect.never)
134 })
135 const evalScript = redis.eval(
136 Redis.script((key: string) => [key], {
137 lua: "return KEYS[1]",
138 numberOfKeys: 1
139 }).withReturnType<string>()
140 )
141
142 const result = yield* evalScript("key")
143
144 assert.strictEqual(result, "ok")
145 assert.deepStrictEqual(commands.map(([command]) => command), [
146 "SCRIPT",
147 "EVALSHA",
148 "SCRIPT",
149 "EVALSHA"
150 ])
151 }))
152
153 it.effect("receives messages from a subscription", () =>
154 Effect.gen(function*() {
155 const redis = yield* Redis.make({
156 send: () => Effect.die("unused"),
157 subscribe: (_channel, onMessage) =>
158 Effect.sync(() => {
159 onMessage({ channel: "events", message: "hello" })
160 return Effect.never
161 })
162 })
163
164 const subscription = yield* redis.subscribe("events")
165
166 assert.deepStrictEqual(yield* Queue.take(subscription), {
167 channel: "events",
168 message: "hello"
169 })
170 }))
171
172 it.effect("reports subscription setup failures through the dequeue", () =>
173 Effect.gen(function*() {
174 const error = new Redis.RedisError({ cause: new Error("subscription failed") })
175 const redis = yield* Redis.make({
176 send: () => Effect.die("unused"),
177 subscribe: () => Effect.fail(error)
178 })
179
180 const subscription = yield* redis.subscribe("events")
181
182 assert.strictEqual(yield* Queue.take(subscription).pipe(Effect.flip), error)
183 }))
184
185 it.effect("reports listener failures through the dequeue", () =>
186 Effect.gen(function*() {
187 const error = new Redis.RedisError({ cause: new Error("listener failed") })
188 const redis = yield* Redis.make({
189 send: () => Effect.die("unused"),
190 subscribe: () => Effect.succeed(Effect.fail(error))
191 })
192
193 const subscription = yield* redis.subscribe("events")
194
195 assert.strictEqual(yield* Queue.take(subscription).pipe(Effect.flip), error)
196 }))
197
198 it.effect("shuts down the dequeue and releases the subscriber with its scope", () =>
199 Effect.gen(function*() {
200 let released = false
201 const redis = yield* Redis.make({
202 send: () => Effect.die("unused"),
203 subscribe: () =>
204 Effect.acquireRelease(
205 Effect.void,
206 () =>
207 Effect.sync(() => {
208 released = true
209 })
210 ).pipe(Effect.as(Effect.never))
211 })
212 const scope = yield* Scope.make()
213 const subscription = yield* redis.subscribe("events").pipe(Scope.provide(scope))
214
215 yield* Scope.close(scope, Exit.void)
216
217 assert.isTrue(released)
218 assert.isTrue(Exit.hasInterrupts(yield* Queue.take(subscription).pipe(Effect.exit)))
219 }))
220})