1import { assert, describe, it } from "@effect/vitest"
2import { Duration, Effect, Exit, Layer, Ref, Schema, Stream } from "effect"
3import { Sse } from "effect/encoding"
4import { HttpClient, HttpClientResponse } from "effect/http"
5import { HttpApi, HttpApiEndpoint, HttpApiGroup, HttpApiSchema } from "effect/http-api"
6import type * as HttpClientError from "effect/http/HttpClientError"
7import { Atom, AtomHttpApi, AtomRegistry, Hydration } from "effect/reactivity"
8
9const Api = HttpApi.make("api").add(
10 HttpApiGroup.make("group").add(
11 HttpApiEndpoint.get("get", "/users/:id", {
12 params: {
13 id: Schema.FiniteFromString
14 },
15 query: {
16 page: Schema.FiniteFromString
17 }
18 })
19 )
20)
21
22const TopLevelApi = HttpApi.make("top-level-api").add(
23 HttpApiGroup.make("group", { topLevel: true }).add(
24 HttpApiEndpoint.get("query", "/query"),
25 HttpApiEndpoint.post("mutation", "/mutation")
26 )
27)
28
29const StreamApi = HttpApi.make("StreamApi").add(
30 HttpApiGroup.make("events").add(
31 HttpApiEndpoint.get("watch", "/watch", { success: HttpApiSchema.StreamSse({ data: Schema.String }) })
32 )
33)
34
35describe("AtomHttpApi", () => {
36 it.effect("query keeps SSE decode options in the cache key and forwards them", () =>
37 Effect.gen(function*() {
38 const Client = makeStreamingClient()
39 const registry = AtomRegistry.make()
40 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()))
41 const limited = Client.query("group", "events", { sseOptions: { maxEventSize: 4 } })
42 const defaults = Client.query("group", "events", {})
43 assert.notStrictEqual(limited, defaults)
44 assert.strictEqual(limited, Client.query("group", "events", { sseOptions: { maxEventSize: 4 } }))
45
46 const limitedStream = yield* AtomRegistry.getResult(registry, limited)
47 const defaultStream = yield* AtomRegistry.getResult(registry, defaults)
48 const [error, values] = yield* Effect.all([
49 limitedStream.pipe(Stream.runCollect, Effect.flip),
50 Stream.runCollect(defaultStream)
51 ], { concurrency: "unbounded" })
52
53 assertEventTooLarge(error)
54 assert.deepStrictEqual(values, ["hello"])
55 }).pipe(Effect.scoped))
56
57 it.effect("mutation forwards SSE decode options independently for each call", () =>
58 Effect.gen(function*() {
59 const Client = makeStreamingClient()
60 const registry = AtomRegistry.make()
61 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()))
62 const mutation = Client.mutation("group", "events", { responseMode: "decoded-only" })
63 yield* AtomRegistry.mount(registry, mutation)
64
65 registry.set(mutation, { sseOptions: { maxEventSize: 4 } })
66 const limitedStream = yield* AtomRegistry.getResult(registry, mutation, { suspendOnWaiting: true })
67 const error = yield* limitedStream.pipe(Stream.runCollect, Effect.flip)
68 assertEventTooLarge(error)
69
70 registry.set(mutation, {})
71 const defaultStream = yield* AtomRegistry.getResult(registry, mutation, { suspendOnWaiting: true })
72 assert.deepStrictEqual(yield* Stream.runCollect(defaultStream), ["hello"])
73 }).pipe(Effect.scoped))
74
75 it.effect("dispatches queries and mutations for top-level groups", () =>
76 Effect.gen(function*() {
77 const requests: Array<string> = []
78 const httpClient = HttpClient.makeWith(
79 Effect.fnUntraced(function*(requestEffect) {
80 const request = yield* requestEffect
81 requests.push(request.url)
82 return HttpClientResponse.fromWeb(request, new Response(null, { status: 204 }))
83 }),
84 Effect.succeed as HttpClient.HttpClient.Preprocess<HttpClientError.HttpClientError, never>
85 )
86 const Client = AtomHttpApi.Service()("TopLevelClient", {
87 api: TopLevelApi,
88 httpClient: Layer.succeed(HttpClient.HttpClient, httpClient)
89 })
90 const registry = AtomRegistry.make()
91 const query = Client.query("group", "query", {})
92 const mutation = Client.mutation("group", "mutation")
93
94 yield* AtomRegistry.mount(registry, query)
95 yield* AtomRegistry.mount(registry, mutation)
96 registry.set(mutation, {})
97
98 const queryExit = yield* Effect.exit(AtomRegistry.getResult(registry, query, { suspendOnWaiting: true }))
99 const mutationExit = yield* Effect.exit(AtomRegistry.getResult(registry, mutation, { suspendOnWaiting: true }))
100
101 assert.deepStrictEqual(
102 { mutation: mutationExit, query: queryExit, requests },
103 { mutation: Exit.succeed(undefined), query: Exit.succeed(undefined), requests: ["/query", "/mutation"] }
104 )
105 }).pipe(Effect.scoped))
106
107 it.effect("query creates a serializable atom with reactivity and retention that encodes the request", () =>
108 Effect.gen(function*() {
109 const requestRef = yield* Ref.make<
110 {
111 readonly url: string
112 readonly urlParams: ReadonlyArray<readonly [string, string]>
113 } | undefined
114 >(undefined)
115
116 const httpClient = HttpClient.makeWith(
117 Effect.fnUntraced(function*(requestEffect) {
118 const request = yield* requestEffect
119 yield* Ref.set(requestRef, {
120 url: request.url,
121 urlParams: request.urlParams.params
122 })
123 return HttpClientResponse.fromWeb(request, new Response(null, { status: 204 }))
124 }),
125 Effect.succeed as HttpClient.HttpClient.Preprocess<HttpClientError.HttpClientError, never>
126 )
127
128 const Client = AtomHttpApi.Service()("Client", {
129 api: Api,
130 httpClient: Layer.succeed(HttpClient.HttpClient, httpClient)
131 })
132
133 const atom = Client.query("group", "get", {
134 params: { id: 1 },
135 query: { page: 2 },
136 reactivityKeys: ["users"],
137 timeToLive: "1 minute",
138 serializationKey: `1:2`
139 })
140
141 assert.deepStrictEqual(
142 {
143 idleTTL: atom.idleTTL,
144 serializable: Atom.isSerializable(atom)
145 },
146 {
147 idleTTL: 60_000,
148 serializable: true
149 }
150 )
151 assert.strictEqual(
152 Client.query("group", "get", { params: { id: 2 }, query: { page: 3 }, timeToLive: 0 }).idleTTL,
153 0
154 )
155 assert.strictEqual(
156 Client.query("group", "get", { params: { id: 3 }, query: { page: 4 }, timeToLive: 0n }).idleTTL,
157 0
158 )
159 const keepAliveAtom = Client.query("group", "get", {
160 params: { id: 2 },
161 query: { page: 3 },
162 reactivityKeys: ["users"],
163 timeToLive: "Infinity",
164 serializationKey: "keep-alive"
165 })
166 assert.deepStrictEqual(
167 {
168 keepAlive: keepAliveAtom.keepAlive,
169 serializable: Atom.isSerializable(keepAliveAtom)
170 },
171 {
172 keepAlive: true,
173 serializable: true
174 }
175 )
176 if (!Atom.isSerializable(atom)) {
177 assert.fail("expected query atom to be serializable")
178 }
179 const key = atom[Atom.SerializableTypeId].key
180
181 const atomFromEncodedInput = Client.query("group", "get", {
182 params: { id: 1 },
183 query: { page: 2 },
184 reactivityKeys: ["users"],
185 timeToLive: "1 minute",
186 serializationKey: `1:2`
187 })
188 if (!Atom.isSerializable(atomFromEncodedInput)) {
189 assert.fail("expected query atom from encoded input to be serializable")
190 }
191 assert.strictEqual(atomFromEncodedInput[Atom.SerializableTypeId].key, key)
192
193 const registry = AtomRegistry.make()
194 const unmount = registry.mount(atom)
195 yield* Effect.yieldNow
196 yield* Effect.yieldNow
197 yield* Effect.yieldNow
198
199 const request = yield* Ref.get(requestRef)
200 assert(request !== undefined)
201 assert.strictEqual(request.url, "/users/1")
202 assert.deepStrictEqual(request.urlParams, [["page", "2"]])
203
204 const dehydrated = Hydration.toValues(Hydration.dehydrate(registry))
205 assert.lengthOf(dehydrated, 1)
206 assert.strictEqual(dehydrated[0]!.key, key)
207
208 unmount()
209 }))
210
211 it.effect("returns generated SSE stream failures from queries", () =>
212 Effect.gen(function*() {
213 const Client = AtomHttpApi.Service()("StreamClient", {
214 api: StreamApi,
215 baseUrl: "https://example.test",
216 httpClient: Layer.succeed(
217 HttpClient.HttpClient,
218 HttpClient.make((request) =>
219 Effect.succeed(HttpClientResponse.fromWeb(
220 request,
221 new Response("retry: 1000\n\n", {
222 headers: { "content-type": "text/event-stream" }
223 })
224 ))
225 )
226 )
227 })
228 const registry = AtomRegistry.make()
229 const atom = Client.query("events", "watch", {})
230 yield* AtomRegistry.mount(registry, atom)
231 const stream = yield* AtomRegistry.getResult(registry, atom)
232 const error = yield* Effect.flip(Stream.runCollect(stream))
233
234 assert.deepStrictEqual(
235 error,
236 new Sse.Retry({ duration: Duration.millis(1000), lastEventId: undefined })
237 )
238 }).pipe(Effect.scoped))
239})
240
241const makeStreamingClient = () =>
242 AtomHttpApi.Service()("StreamingClient", {
243 baseUrl: "http://test",
244 api: HttpApi.make("StreamingApi").add(
245 HttpApiGroup.make("group").add(
246 HttpApiEndpoint.get("events", "/events", {
247 success: HttpApiSchema.StreamSse({ data: Schema.String })
248 })
249 )
250 ),
251 httpClient: Layer.succeed(
252 HttpClient.HttpClient,
253 HttpClient.make((request) =>
254 Effect.succeed(HttpClientResponse.fromWeb(
255 request,
256 new Response(
257 new ReadableStream<Uint8Array>({
258 start(controller) {
259 const encoder = new TextEncoder()
260 controller.enqueue(encoder.encode("data: "))
261 controller.enqueue(encoder.encode("\"hello\"\n\n"))
262 controller.close()
263 }
264 }),
265 { headers: { "content-type": "text/event-stream" } }
266 )
267 ))
268 )
269 )
270 })
271
272const assertEventTooLarge = (error: unknown) => {
273 assert.instanceOf(error, Sse.SseError)
274 assert.instanceOf(error.reason, Sse.EventTooLarge)
275 assert.strictEqual(error.reason.maxEventSize, 4)
276}