1import { assert, describe, expectTypeOf, it } from "@effect/vitest"
2import { type Cause, Deferred, Effect, Layer, Schema } from "effect"
3import { AsyncResult, Atom, AtomRegistry, AtomRpc, Hydration } from "effect/reactivity"
4import { Rpc, RpcClient, type RpcClientError, RpcGroup, RpcMiddleware, type RpcSerialization } from "effect/rpc"
5
6const Group = RpcGroup.make(
7 Rpc.make("getUser", {
8 payload: Schema.Struct({
9 id: Schema.FiniteFromString
10 }),
11 success: Schema.Struct({
12 id: Schema.Number,
13 name: Schema.String
14 })
15 })
16)
17
18class ClientFailure extends Schema.Error<ClientFailure>("ClientFailure")({
19 _tag: Schema.tag("ClientFailure")
20}) {}
21
22class ClientMiddleware extends RpcMiddleware.Service<ClientMiddleware, {
23 clientError: ClientFailure
24}>()("ClientMiddleware", {
25 requiredForClient: true
26}) {}
27
28const ClientMiddlewareGroup = RpcGroup.make(
29 Rpc.make("getUser", {
30 success: Schema.String
31 }).middleware(ClientMiddleware),
32 Rpc.make("stream", {
33 success: Schema.String,
34 stream: true
35 }).middleware(ClientMiddleware)
36)
37
38const makeClientMiddlewareProtocol = (
39 middleware: RpcMiddleware.RpcMiddlewareClient<never, ClientFailure, never>
40) =>
41 Layer.mergeAll(
42 Layer.effect(
43 RpcClient.Protocol,
44 RpcClient.Protocol.make(() =>
45 Effect.succeed({
46 send: () => Effect.die("unexpected request"),
47 supportsAck: false,
48 supportsTransferables: false,
49 codecFor: Schema.toCodecJson as RpcSerialization.CodecFor
50 })
51 )
52 ),
53 RpcMiddleware.layerClient(ClientMiddleware, middleware)
54 )
55
56describe("AtomRpc", () => {
57 it("includes client middleware errors in atom failure types", () => {
58 const Client = AtomRpc.Service()("MiddlewareClient", {
59 group: ClientMiddlewareGroup,
60 protocol: Layer.empty,
61 makeEffect: Effect.die("unused")
62 })
63
64 const mutation = Client.mutation("getUser")
65 const query = Client.query("getUser", undefined)
66 const stream = Client.query("stream", undefined)
67
68 expectTypeOf<Atom.Failure<typeof mutation>>().toEqualTypeOf<RpcClientError.RpcClientError | ClientFailure>()
69 expectTypeOf<Atom.Failure<typeof query>>().toEqualTypeOf<RpcClientError.RpcClientError | ClientFailure>()
70 expectTypeOf<Atom.Failure<typeof stream>>().toEqualTypeOf<
71 RpcClientError.RpcClientError | ClientFailure | Cause.NoSuchElementError
72 >()
73 })
74
75 it.effect("query creates a serializable atom with reactivity and retention", () =>
76 Effect.gen(function*() {
77 const Client = AtomRpc.Service()("Client", {
78 group: Group,
79 protocol: Layer.empty,
80 makeEffect: Effect.succeed(
81 ((tag: string, payload: { readonly id: number }) => {
82 if (tag !== "getUser") {
83 return Effect.die(`unexpected tag: ${tag}`)
84 }
85 return Effect.succeed({
86 id: payload.id,
87 name: `user-${payload.id}`
88 })
89 }) as any
90 )
91 })
92
93 const atom = Client.query("getUser", { id: 1 }, {
94 headers: {
95 "x-id": "abc"
96 },
97 reactivityKeys: ["users"],
98 timeToLive: "1 minute",
99 serializationKey: "1"
100 })
101
102 assert.deepStrictEqual(
103 {
104 idleTTL: atom.idleTTL,
105 serializable: Atom.isSerializable(atom)
106 },
107 {
108 idleTTL: 60_000,
109 serializable: true
110 }
111 )
112 assert.strictEqual(Client.query("getUser", { id: 2 }, { timeToLive: 0 }).idleTTL, 0)
113 assert.strictEqual(Client.query("getUser", { id: 3 }, { timeToLive: 0n }).idleTTL, 0)
114 const keepAliveAtom = Client.query("getUser", { id: 2 }, {
115 reactivityKeys: ["users"],
116 timeToLive: "Infinity",
117 serializationKey: "keep-alive"
118 })
119 assert.deepStrictEqual(
120 {
121 keepAlive: keepAliveAtom.keepAlive,
122 serializable: Atom.isSerializable(keepAliveAtom)
123 },
124 {
125 keepAlive: true,
126 serializable: true
127 }
128 )
129 if (!Atom.isSerializable(atom)) {
130 assert.fail("expected query atom to be serializable")
131 }
132 const key = atom[Atom.SerializableTypeId].key
133
134 const atomFromEncodedPayload = Client.query("getUser", { id: 1 }, {
135 headers: {
136 "x-id": "abc"
137 },
138 reactivityKeys: ["users"],
139 timeToLive: "1 minute",
140 serializationKey: "1"
141 })
142 assert(Atom.isSerializable(atomFromEncodedPayload), "expected query atom from encoded payload to be serializable")
143 assert(atom === atomFromEncodedPayload)
144 assert.strictEqual(atomFromEncodedPayload[Atom.SerializableTypeId].key, key)
145
146 const registry = AtomRegistry.make()
147 const unmount = registry.mount(atom)
148 yield* Effect.yieldNow
149 yield* Effect.yieldNow
150
151 const dehydrated = Hydration.toValues(Hydration.dehydrate(registry))
152 assert.lengthOf(dehydrated, 1)
153 assert.strictEqual(dehydrated[0]!.key, key)
154
155 unmount()
156 }))
157
158 it.effect("dehydrates the rest of the registry when client middleware fails", () =>
159 Effect.gen(function*() {
160 const Client = AtomRpc.Service()("ClientWithMiddleware", {
161 group: ClientMiddlewareGroup,
162 protocol: makeClientMiddlewareProtocol(() => Effect.fail(new ClientFailure({})))
163 })
164 const failedAtom = Client.query("getUser", undefined, {
165 serializationKey: "failed"
166 })
167 const unaffectedAtom = Atom.make(42).pipe(
168 Atom.serializable({
169 key: "unaffected",
170 schema: Schema.Number
171 })
172 )
173 const registry = AtomRegistry.make()
174 const unmountFailed = registry.mount(failedAtom)
175 const unmountUnaffected = registry.mount(unaffectedAtom)
176 yield* Effect.yieldNow
177 yield* Effect.yieldNow
178
179 assert(AsyncResult.isFailure(registry.get(failedAtom)))
180 const dehydrated = Hydration.toValues(Hydration.dehydrate(registry))
181 assert.isUndefined(dehydrated.find((entry) => entry.key === "AtomRpc:getUser:failed"))
182 assert.strictEqual(dehydrated.find((entry) => entry.key === "unaffected")?.value, 42)
183
184 unmountUnaffected()
185 unmountFailed()
186 }))
187
188 it.effect("settles promise dehydration when client middleware later fails", () =>
189 Effect.gen(function*() {
190 const failMiddleware = yield* Deferred.make<void>()
191 const Client = AtomRpc.Service()("ClientWithDeferredMiddleware", {
192 group: ClientMiddlewareGroup,
193 protocol: makeClientMiddlewareProtocol(() =>
194 Deferred.await(failMiddleware).pipe(
195 Effect.andThen(Effect.fail(new ClientFailure({})))
196 )
197 )
198 })
199 const failedAtom = Client.query("getUser", undefined, {
200 serializationKey: "deferred-failure"
201 })
202 const registry = AtomRegistry.make()
203 const unmountFailed = registry.mount(failedAtom)
204
205 assert(AsyncResult.isInitial(registry.get(failedAtom)))
206 const dehydrated = Hydration.toValues(Hydration.dehydrate(registry, {
207 encodeInitialAs: "promise"
208 }))
209 const resultPromise = dehydrated.find((entry) => entry.key === "AtomRpc:getUser:deferred-failure")
210 ?.resultPromise
211 assert.isDefined(resultPromise)
212
213 let notifications = 0
214 const unsubscribe = registry.subscribe(failedAtom, () => notifications++)
215 const hydratedRegistry = AtomRegistry.make()
216 Hydration.hydrate(hydratedRegistry, dehydrated)
217
218 yield* Deferred.succeed(failMiddleware, undefined)
219 yield* Effect.promise(() => resultPromise).pipe(Effect.timeout("1 second"))
220
221 assert.strictEqual(notifications, 1)
222 const unmountHydrated = hydratedRegistry.mount(failedAtom)
223
224 unmountHydrated()
225 unsubscribe()
226 unmountFailed()
227 }))
228})