1/**
2 * Deno Redis integration backed by `@db/redis`.
3 *
4 * This module creates a scoped, native Deno Redis client and exposes it as
5 * both the portable `Redis` service and the Deno-specific {@link DenoRedis}
6 * service for direct access to the raw client. Unlike Bun's built-in client,
7 * `@db/redis` connects eagerly using RESP2, so layer construction can fail
8 * with a `RedisError`.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import { connect, parseURL, type Redis as RedisClient, type RedisConnectOptions } from "@db/redis"
14import * as Config from "effect/Config"
15import * as Context from "effect/Context"
16import * as Effect from "effect/Effect"
17import * as Fn from "effect/Function"
18import * as Layer from "effect/Layer"
19import * as Redis from "effect/persistence/Redis"
20import * as Predicate from "effect/Predicate"
21import * as Record from "effect/Record"
22
23/**
24 * Options for connecting to Redis, including a Redis URL or individual
25 * connection settings. Explicit settings override values from the URL.
26 *
27 * @stability unstable
28 * @category models
29 * @since 4.0.0
30 */
31export type RedisOptions = Omit<RedisConnectOptions, "hostname"> & {
32 readonly hostname?: string
33 readonly url?: string
34}
35
36/**
37 * Service tag for Deno Redis integration, exposing the raw `@db/redis` client
38 * and a `use` helper that maps client promise failures to `RedisError`.
39 *
40 * @stability unstable
41 * @category services
42 * @since 4.0.0
43 */
44export class DenoRedis extends Context.Service<DenoRedis, {
45 readonly client: RedisClient
46 readonly use: <A>(f: (client: RedisClient) => Promise<A>) => Effect.Effect<A, Redis.RedisError>
47}>()("@effect/platform-deno/DenoRedis") {}
48
49const make = Effect.fnUntraced(function*(options: RedisOptions = {}) {
50 const connectClient = () => {
51 const { url, ...connectOptions } = options
52 const parsedUrl = url === undefined ? undefined : new URL(url)
53 const { name, ...parsed } = url === undefined ? { hostname: "localhost" } : parseURL(url)
54 return connect({
55 ...parsed,
56 ...(name === undefined ? {} : { username: name }),
57 ...(parsedUrl?.username ? { username: decodeURIComponent(parsedUrl.username) } : {}),
58 ...(parsedUrl?.password ? { password: decodeURIComponent(parsedUrl.password) } : {}),
59 ...Record.filter(connectOptions, Predicate.isNotUndefined)
60 })
61 }
62
63 const client = yield* Effect.acquireRelease(
64 Effect.tryPromise({
65 try: connectClient,
66 catch: (cause) => new Redis.RedisError({ cause })
67 }),
68 (client) => Effect.sync(() => client.close())
69 )
70
71 const use = <A>(f: (client: RedisClient) => Promise<A>) =>
72 Effect.tryPromise({
73 try: () => f(client),
74 catch: (cause) => new Redis.RedisError({ cause })
75 })
76
77 const redis = yield* Redis.make({
78 send: <A = unknown>(command: string, ...args: ReadonlyArray<string>) =>
79 Effect.tryPromise({
80 try: () => client.sendCommand(command, args as Array<string>) as Promise<A>,
81 catch: (cause) => new Redis.RedisError({ cause })
82 }),
83 subscribe: (channel, onMessage) =>
84 Effect.acquireRelease(
85 Effect.tryPromise({
86 try: async () => {
87 const subscriber = await connectClient()
88 try {
89 const subscription = await subscriber.subscribe(channel)
90 // @db/redis resolves subscribe after writing the command, before
91 // reading its acknowledgement. This ordered reply proves that
92 // Redis processed SUBSCRIBE before the dequeue is returned.
93 await subscriber.ping()
94 return { subscriber, subscription }
95 } catch (cause) {
96 subscriber.close()
97 throw cause
98 }
99 },
100 catch: (cause) => new Redis.RedisError({ cause })
101 }),
102 ({ subscriber }) => Effect.sync(() => subscriber.close())
103 ).pipe(
104 Effect.map(({ subscription }) =>
105 Effect.tryPromise({
106 try: async () => {
107 for await (const message of subscription.receive()) {
108 onMessage(message)
109 }
110 },
111 catch: (cause) => new Redis.RedisError({ cause })
112 })
113 )
114 )
115 })
116
117 const denoRedis = Fn.identity<DenoRedis["Service"]>({
118 client,
119 use
120 })
121
122 return Context.make(DenoRedis, denoRedis).pipe(
123 Context.add(Redis.Redis, redis)
124 )
125})
126
127/**
128 * Provides `Redis` and `DenoRedis` services backed by an `@db/redis` client,
129 * closing the client when the layer scope ends. URL-derived options can be
130 * overridden by other supplied options.
131 *
132 * @stability unstable
133 * @category layers
134 * @since 4.0.0
135 */
136export const layer = (
137 options?: RedisOptions | undefined
138): Layer.Layer<Redis.Redis | DenoRedis, Redis.RedisError> => Layer.effectContext(make(options))
139
140/**
141 * Provides `Redis` and `DenoRedis` services from `Config`-backed options,
142 * closing the client when the layer scope ends.
143 *
144 * @stability unstable
145 * @category layers
146 * @since 4.0.0
147 */
148export const layerConfig = (
149 options: Config.Wrap<RedisOptions>
150): Layer.Layer<Redis.Redis | DenoRedis, Redis.RedisError | Config.ConfigError> =>
151 Layer.effectContext(
152 Config.unwrap(options).pipe(
153 Effect.flatMap(make)
154 )
155 )