1import * as DenoRedis from "@effect/platform-deno/DenoRedis"
2import { assert, it } from "@effect/vitest"
3import { RedisContainer } from "@testcontainers/redis"
4import { Deferred, Effect, Fiber, Layer, Queue, Schema } from "effect"
5import { PersistedQueue, Persistence, Redis } from "effect/persistence"
6import * as PersistedCacheTest from "../../../effect/test/persistence/PersistedCacheTest.ts"
7import * as PersistedQueueTest from "../../../effect/test/persistence/PersistedQueueTest.ts"
8
9const RedisLayer = Layer.unwrap(
10 Effect.gen(function*() {
11 const container = yield* Effect.acquireRelease(
12 Effect.promise(() => new RedisContainer("redis:alpine").start()),
13 (container) => Effect.promise(() => container.stop())
14 )
15 return DenoRedis.layer({
16 url: `redis://${container.getHost()}:${container.getMappedPort(6379)}`
17 })
18 }).pipe(
19 Effect.catchCause(() => Effect.fail(new PersistedCacheTest.TransientError()))
20 )
21)
22
23PersistedCacheTest.suite(
24 "DenoRedis",
25 Persistence.layerRedis.pipe(Layer.provide(RedisLayer))
26)
27
28PersistedQueueTest.suite(
29 "DenoRedis",
30 // short intervals so the periodic reset runs while the suite's takes are
31 // in flight
32 PersistedQueue.layerStoreRedis({
33 pollInterval: "50 millis",
34 lockRefreshInterval: "100 millis"
35 }).pipe(Layer.provide(RedisLayer))
36)
37
38const PersistedQueueRedisLayer = Layer.mergeAll(
39 RedisLayer,
40 PersistedQueue.layer.pipe(
41 Layer.provideMerge(
42 PersistedQueue.layerStoreRedis().pipe(Layer.provide(RedisLayer))
43 )
44 )
45)
46
47it.layer(PersistedQueueRedisLayer, { timeout: "30 seconds" })(
48 "PersistedQueue (DenoRedis)",
49 (it) => {
50 it.effect("receives published messages", () =>
51 Effect.gen(function*() {
52 const redis = yield* Redis.Redis
53 const subscription = yield* redis.subscribe("effect-test")
54
55 const subscribers = yield* redis.send<number>("PUBLISH", "effect-test", "hello")
56 assert.strictEqual(subscribers, 1)
57
58 assert.deepStrictEqual(yield* Queue.take(subscription), {
59 channel: "effect-test",
60 message: "hello"
61 })
62 }))
63
64 it.effect("moves exhausted elements to the failed list", () =>
65 Effect.gen(function*() {
66 const redis = yield* DenoRedis.DenoRedis
67 const queueName = "test-redis-failed"
68
69 const queue = yield* PersistedQueue.make({
70 name: queueName,
71 schema: RedisItem,
72 maxAttempts: 1
73 })
74 const id = yield* queue.offer({ n: 42 })
75 const error = yield* queue.take(() => Effect.fail("boom")).pipe(Effect.flip)
76 assert.strictEqual(error, "boom")
77
78 const failed = yield* redis.use((client) => client.lrange(`effectq:${queueName}:failed`, 0, -1))
79 assert.strictEqual(failed.length, 1)
80 const failedItem = JSON.parse(failed[0])
81 assert.strictEqual(failedItem.id, id)
82 assert.deepStrictEqual(failedItem.element, { n: 42 })
83 assert.strictEqual(failedItem.attempts, 1)
84
85 const pending = yield* redis.use((client) => client.hlen(`effectq:${queueName}:pending`))
86 assert.strictEqual(pending, 0)
87 }))
88 }
89)
90
91it.effect("closes the connection when interrupted during acquisition", () =>
92 Effect.gen(function*() {
93 const listener = yield* makeListener
94 const authReceived = yield* Deferred.make<void>()
95 const sendAuthReply = yield* Deferred.make<void>()
96 const server = yield* Effect.gen(function*() {
97 const connection = yield* accept(listener)
98 yield* read(connection)
99 yield* Deferred.succeed(authReceived, void 0)
100 yield* Deferred.await(sendAuthReply)
101 yield* write(connection, "+OK\r\n")
102 return yield* read(connection)
103 }).pipe(Effect.forkChild)
104
105 const port = (listener.addr as Deno.NetAddr).port
106 const layerFiber = yield* Layer.build(DenoRedis.layer({
107 url: `redis://:secret@127.0.0.1:${port}`
108 })).pipe(Effect.forkChild({ startImmediately: true }))
109
110 yield* Deferred.await(authReceived)
111 const interruptFiber = yield* Fiber.interrupt(layerFiber).pipe(
112 Effect.forkChild({ startImmediately: true })
113 )
114 yield* Deferred.succeed(sendAuthReply, void 0)
115 yield* Fiber.join(interruptFiber)
116
117 assert.isNull(yield* Fiber.join(server))
118 }))
119
120it.effect("uses the URL username for two-argument AUTH", () =>
121 Effect.gen(function*() {
122 const listener = yield* makeListener
123 const server = yield* Effect.gen(function*() {
124 const connection = yield* accept(listener)
125 const request = yield* read(connection)
126 yield* write(connection, "+OK\r\n")
127 return request
128 }).pipe(Effect.forkChild)
129
130 const port = (listener.addr as Deno.NetAddr).port
131 yield* Layer.build(DenoRedis.layer({
132 url: `redis://alice:secret@127.0.0.1:${port}`,
133 password: undefined
134 }))
135
136 assert.strictEqual(
137 yield* Fiber.join(server),
138 "*3\r\n$4\r\nAUTH\r\n$5\r\nalice\r\n$6\r\nsecret\r\n"
139 )
140 }))
141
142it.effect("decodes URL authority credentials once", () =>
143 Effect.gen(function*() {
144 const listener = yield* makeListener
145 const server = yield* Effect.gen(function*() {
146 const connection = yield* accept(listener)
147 const request = yield* read(connection)
148 yield* write(connection, "+OK\r\n")
149 return request
150 }).pipe(Effect.forkChild)
151
152 const port = (listener.addr as Deno.NetAddr).port
153 yield* Layer.build(DenoRedis.layer({
154 url: `redis://app%3A%2540:p%40ss%2525@127.0.0.1:${port}`
155 }))
156
157 assert.strictEqual(
158 yield* Fiber.join(server),
159 "*3\r\n$4\r\nAUTH\r\n$7\r\napp:%40\r\n$7\r\np@ss%25\r\n"
160 )
161 }))
162
163it.effect("does not decode a query-string password twice", () =>
164 Effect.gen(function*() {
165 const listener = yield* makeListener
166 const server = yield* Effect.gen(function*() {
167 const connection = yield* accept(listener)
168 const request = yield* read(connection)
169 yield* write(connection, "+OK\r\n")
170 return request
171 }).pipe(Effect.forkChild)
172
173 const port = (listener.addr as Deno.NetAddr).port
174 yield* Layer.build(DenoRedis.layer({
175 url: `redis://alice@127.0.0.1:${port}/?password=p%2540ss`
176 }))
177
178 assert.strictEqual(
179 yield* Fiber.join(server),
180 "*3\r\n$4\r\nAUTH\r\n$5\r\nalice\r\n$6\r\np%40ss\r\n"
181 )
182 }))
183
184it.effect("prefers explicit credentials over URL credentials", () =>
185 Effect.gen(function*() {
186 const listener = yield* makeListener
187 const server = yield* Effect.gen(function*() {
188 const connection = yield* accept(listener)
189 const request = yield* read(connection)
190 yield* write(connection, "+OK\r\n")
191 return request
192 }).pipe(Effect.forkChild)
193
194 const port = (listener.addr as Deno.NetAddr).port
195 yield* Layer.build(DenoRedis.layer({
196 url: `redis://alice:secret@127.0.0.1:${port}`,
197 username: "bob",
198 password: "other"
199 }))
200
201 assert.strictEqual(
202 yield* Fiber.join(server),
203 "*3\r\n$4\r\nAUTH\r\n$3\r\nbob\r\n$5\r\nother\r\n"
204 )
205 }))
206
207const RedisItem = Schema.Struct({
208 n: Schema.Number
209})
210
211const encoder = new TextEncoder()
212const decoder = new TextDecoder()
213
214const makeListener = Effect.acquireRelease(
215 Effect.sync(() => Deno.listen({ hostname: "127.0.0.1", port: 0 })),
216 (listener) => Effect.sync(() => listener.close())
217)
218
219const accept = (listener: Deno.TcpListener) =>
220 Effect.acquireRelease(
221 Effect.promise(() => listener.accept()),
222 (connection) => Effect.sync(() => connection.close())
223 )
224
225const read = (connection: Deno.TcpConn) =>
226 Effect.promise(() => {
227 const buffer = new Uint8Array(256)
228 return connection.read(buffer).then((size) => size === null ? null : decoder.decode(buffer.subarray(0, size)))
229 })
230
231const write = (connection: Deno.TcpConn, value: string) => Effect.promise(() => connection.write(encoder.encode(value)))