1import { NodeHttpServer, NodeSocket, NodeSocketServer } from "@effect/platform-node"
2import { assert, describe, it } from "@effect/vitest"
3import { Cause, Deferred, Effect, Exit, Fiber, Layer, Ref, Schedule, Schema, Stream } from "effect"
4import { Entity, EntityProxy, EntityProxyServer, Sharding } from "effect/cluster"
5import { HttpClient, HttpClientRequest, HttpRouter, HttpServer } from "effect/http"
6import type * as NetAddress from "effect/net/NetAddress"
7import { Rpc, RpcClient, RpcGroup, RpcSerialization, RpcServer, RpcTest } from "effect/rpc"
8import { SocketServer } from "effect/socket"
9import { e2eSuite, UsersClient } from "./fixtures/rpc-e2e.ts"
10import { RpcLayer, User } from "./fixtures/rpc-schemas.ts"
11
12describe("RpcServer", () => {
13 // http ndjson
14 const HttpProtocol = RpcServer.layerProtocolHttp({ path: "/rpc" })
15 const HttpNdjsonServer = HttpRouter.serve(
16 RpcLayer.pipe(Layer.provideMerge(HttpProtocol)),
17 { disableListenLog: true, disableLogger: true }
18 )
19 const HttpNdjsonClient = UsersClient.layer.pipe(
20 Layer.provide(
21 RpcClient.layerProtocolHttp({
22 url: "",
23 transformClient: HttpClient.mapRequest(HttpClientRequest.appendUrl("/rpc"))
24 })
25 )
26 )
27 const CustomDefectLayer = HttpNdjsonClient.pipe(
28 Layer.provideMerge(HttpNdjsonServer),
29 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerNdjson])
30 )
31 e2eSuite(
32 "e2e http ndjson",
33 HttpNdjsonClient.pipe(
34 Layer.provideMerge(HttpNdjsonServer),
35 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerNdjson])
36 )
37 )
38 e2eSuite(
39 "e2e http SchemaBinary",
40 HttpNdjsonClient.pipe(
41 Layer.provideMerge(HttpNdjsonServer),
42 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerSchemaBinary()])
43 )
44 )
45 e2eSuite(
46 "e2e http jsonrpc",
47 HttpNdjsonClient.pipe(
48 Layer.provideMerge(HttpNdjsonServer),
49 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerNdJsonRpc()])
50 )
51 )
52
53 // websocket
54 const WsProtocol = RpcServer.layerProtocolWebsocket({ path: "/rpc" })
55 const HttpWsServer = HttpRouter.serve(
56 RpcLayer.pipe(Layer.provideMerge(WsProtocol)),
57 { disableListenLog: true, disableLogger: true }
58 )
59 const HttpWsClient = UsersClient.layer.pipe(
60 Layer.provide(RpcClient.layerProtocolSocket()),
61 Layer.provide(
62 Effect.gen(function*() {
63 const server = yield* HttpServer.HttpServer
64 const address = server.address as NetAddress.InetAddress
65 return NodeSocket.layerWebSocket(`http://127.0.0.1:${address.port}/rpc`)
66 }).pipe(Layer.unwrap)
67 )
68 )
69 e2eSuite(
70 "e2e ws ndjson",
71 HttpWsClient.pipe(
72 Layer.provideMerge(HttpWsServer),
73 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerNdjson])
74 )
75 )
76 e2eSuite(
77 "e2e ws json",
78 HttpWsClient.pipe(
79 Layer.provideMerge(HttpWsServer),
80 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerJson])
81 )
82 )
83 e2eSuite(
84 "e2e ws SchemaBinary",
85 HttpWsClient.pipe(
86 Layer.provideMerge(HttpWsServer),
87 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerSchemaBinary()])
88 )
89 )
90 e2eSuite(
91 "e2e ws jsonrpc",
92 HttpWsClient.pipe(
93 Layer.provideMerge(HttpWsServer),
94 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerJsonRpc()])
95 )
96 )
97
98 // tcp
99 const TcpServer = RpcLayer.pipe(
100 Layer.provideMerge(RpcServer.layerProtocolSocketServer),
101 Layer.provideMerge(NodeSocketServer.layer({ port: 0 }))
102 )
103 const TcpClient = UsersClient.layer.pipe(
104 Layer.provide(RpcClient.layerProtocolSocket()),
105 Layer.provide(
106 Effect.gen(function*() {
107 const server = yield* SocketServer.SocketServer
108 const address = server.address as NetAddress.InetAddress
109 return NodeSocket.layerNet({ port: address.port })
110 }).pipe(Layer.unwrap)
111 )
112 )
113 e2eSuite(
114 "e2e tcp ndjson",
115 TcpClient.pipe(
116 Layer.provideMerge(TcpServer),
117 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerNdjson])
118 )
119 )
120 e2eSuite(
121 "e2e tcp SchemaBinary",
122 TcpClient.pipe(
123 Layer.provideMerge(TcpServer),
124 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerSchemaBinary()])
125 )
126 )
127 e2eSuite(
128 "e2e tcp jsonrpc",
129 TcpClient.pipe(
130 Layer.provideMerge(TcpServer),
131 Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerNdJsonRpc()])
132 )
133 )
134
135 // worker
136 // const WorkerClient = UsersClient.layer.pipe(
137 // Layer.provide(RpcClient.layerProtocolWorker({ size: 1 })),
138 // Layer.provide(
139 // NodeWorker.layerPlatform(() =>
140 // CP.fork(new URL("./fixtures/rpc-worker.ts", import.meta.url), {
141 // execPath: "node"
142 // })
143 // )
144 // ),
145 // Layer.merge(Layer.succeed(RpcServer.Protocol, {
146 // supportsAck: true
147 // } as any))
148 // )
149 // e2eSuite("e2e worker", WorkerClient)
150
151 describe("RpcTest", () => {
152 it.effect("works", () =>
153 Effect.gen(function*() {
154 const client = yield* UsersClient
155 const user = yield* client.GetUser({ id: "1" })
156 assert.deepStrictEqual(user, new User({ id: "1", name: "Logged in user" }))
157 }).pipe(Effect.provide(UsersClient.layerTest)))
158 })
159
160 describe("custom defect schema", () => {
161 it.effect("preserves full defect with custom schema", () =>
162 Effect.gen(function*() {
163 const client = yield* UsersClient
164 const cause = yield* client.ProduceDefectCustom().pipe(
165 Effect.sandbox,
166 Effect.flip
167 )
168 const defect = Cause.squash(cause)
169 assert.instanceOf(defect, Error)
170 assert.strictEqual(defect.name, "CustomDefect")
171 assert.strictEqual(defect.message, "detailed error")
172 assert.strictEqual(defect.stack, "Error: detailed error\n at handler.ts:1")
173 }).pipe(Effect.provide(CustomDefectLayer)))
174 })
175
176 describe("entity proxy", () => {
177 it.effect("provides handler context for generated rpc handlers", () =>
178 Effect.gen(function*() {
179 const TestEntity = Entity.make("TestEntity", [Rpc.make("NoPayload")])
180 const TestEntityRpcs = EntityProxy.toRpcGroup(TestEntity)
181 const called = yield* Deferred.make<void>()
182 const testClient = (entityId: string) => ({
183 NoPayload: (payload: void, options?: { readonly discard?: boolean }) =>
184 Effect.gen(function*() {
185 assert.strictEqual(entityId, "id")
186 assert.strictEqual(payload, undefined)
187 assert.strictEqual(options?.discard, true)
188 yield* Deferred.succeed(called, undefined)
189 })
190 })
191 const sharding = Sharding.Sharding.of({
192 ...({} as Sharding.Sharding["Service"]),
193 isShutdown: Effect.succeed(false),
194 makeClient: () => Effect.succeed(testClient) as never,
195 pollStorage: Effect.void
196 })
197
198 const client = yield* RpcTest.makeClient(TestEntityRpcs).pipe(
199 Effect.provide(EntityProxyServer.layerRpcHandlers(TestEntity)),
200 Effect.provideService(Sharding.Sharding, sharding)
201 )
202
203 yield* client["TestEntity.NoPayloadDiscard"]({
204 entityId: "id",
205 payload: undefined
206 })
207 yield* Deferred.await(called)
208 }))
209 })
210
211 // Asserts a failing sibling call does not affect an in-flight Ticker stream on the same connection
212 const Ticker = Rpc.make("Ticker", {
213 success: Schema.Number,
214 stream: true
215 })
216
217 const IsolationClient = RpcClient.layerProtocolSocket().pipe(
218 Layer.provide(
219 Effect.gen(function*() {
220 const server = yield* SocketServer.SocketServer
221 const address = server.address as NetAddress.InetAddress
222 return NodeSocket.layerNet({ port: address.port })
223 }).pipe(Layer.unwrap)
224 ),
225 Layer.provide(RpcSerialization.layerNdjson)
226 )
227
228 const assertTickerSurvives = <E, R>(
229 setup: Effect.Effect<
230 {
231 readonly ticker: Stream.Stream<number, E>
232 readonly failingCall: Effect.Effect<void>
233 },
234 never,
235 R
236 >,
237 label: string
238 ) =>
239 Effect.gen(function*() {
240 const { failingCall, ticker } = yield* setup
241
242 const received = yield* Ref.make<Array<number>>([])
243
244 const tickerFiber = yield* ticker.pipe(
245 Stream.runForEach((value) => Ref.update(received, (xs) => [...xs, value])),
246 Effect.forkChild
247 )
248
249 yield* Effect.retry(
250 Effect.flatMap(
251 Ref.get(received),
252 (xs) => xs.length >= 2 ? Effect.void : Effect.fail("not enough ticks yet")
253 ),
254 { schedule: Schedule.spaced("50 millis"), times: 200 }
255 )
256
257 const ticksBefore = (yield* Ref.get(received)).length
258 assert.isAtLeast(ticksBefore, 2)
259
260 yield* failingCall
261
262 yield* Effect.sleep("300 millis")
263
264 const ticksAfter = (yield* Ref.get(received)).length
265 const tickerStatus = tickerFiber.pollUnsafe()
266
267 yield* Fiber.interrupt(tickerFiber)
268
269 assert.isUndefined(tickerStatus, `Ticker stream must still be running after ${label}`)
270 assert.isAbove(ticksAfter, ticksBefore, `Ticker stream must keep emitting after ${label}`)
271 })
272
273 describe("unknown-tag isolation", () => {
274 const Ghost = Rpc.make("Ghost", {
275 payload: { value: Schema.String },
276 success: Schema.String
277 })
278
279 const serverGroup = RpcGroup.make(Ticker)
280 const clientGroup = RpcGroup.make(Ticker, Ghost)
281
282 const TickerHandlers = serverGroup.toLayer({
283 Ticker: () => Stream.fromSchedule(Schedule.spaced("60 millis"))
284 })
285
286 const IsolationServer = RpcServer.layer(serverGroup).pipe(
287 Layer.provide(TickerHandlers),
288 Layer.provideMerge(RpcServer.layerProtocolSocketServer),
289 Layer.provideMerge(NodeSocketServer.layer({ port: 0 })),
290 Layer.provide(RpcSerialization.layerNdjson)
291 )
292
293 it.live(
294 "an unknown request tag fails only its own request, not other in-flight streams on the same connection",
295 () =>
296 assertTickerSurvives(
297 Effect.map(RpcClient.make(clientGroup), (client) => ({
298 ticker: client.Ticker(),
299 failingCall: Effect.gen(function*() {
300 const ghostExit = yield* client.Ghost({ value: "boo" }).pipe(Effect.exit)
301 assert.isTrue(Exit.isFailure(ghostExit), "Ghost call should fail with the routing miss")
302 })
303 })),
304 "the unknown-tag failure"
305 ).pipe(Effect.provide(IsolationClient.pipe(Layer.provideMerge(IsolationServer)))),
306 { timeout: 30_000 }
307 )
308 })
309
310 describe("fatal-defect isolation", () => {
311 const Boom = Rpc.make("Boom", {
312 success: Schema.String
313 })
314
315 const group = RpcGroup.make(Ticker, Boom)
316
317 const Handlers = group.toLayer({
318 Ticker: () => Stream.fromSchedule(Schedule.spaced("60 millis")),
319 Boom: () => Effect.die("boom")
320 })
321
322 const DefectServer = RpcServer.layer(group, { disableFatalDefects: true }).pipe(
323 Layer.provide(Handlers),
324 Layer.provideMerge(RpcServer.layerProtocolSocketServer),
325 Layer.provideMerge(NodeSocketServer.layer({ port: 0 })),
326 Layer.provide(RpcSerialization.layerNdjson)
327 )
328
329 it.live(
330 "with disableFatalDefects a handler defect fails only its own request, not other in-flight streams",
331 () =>
332 assertTickerSurvives(
333 Effect.map(RpcClient.make(group), (client) => ({
334 ticker: client.Ticker(),
335 failingCall: Effect.gen(function*() {
336 const boomExit = yield* client.Boom().pipe(Effect.exit)
337 if (!Exit.isFailure(boomExit)) {
338 return assert.fail("Boom call must fail with the handler defect")
339 }
340 assert.include(
341 String(Cause.squash(boomExit.cause)),
342 "boom",
343 "the caller must receive the handler defect"
344 )
345 })
346 })),
347 "the handler defect"
348 ).pipe(Effect.provide(IsolationClient.pipe(Layer.provideMerge(DefectServer)))),
349 { timeout: 30_000 }
350 )
351 })
352})