1import { assert, describe, it } from "@effect/vitest"
2import { Cause, Context, Effect, Fiber, Option, Stream } from "effect"
3// @ts-ignore
4// oxlint-disable-next-line @typescript-eslint/no-unused-vars
5import { NodeInspectSymbol } from "effect/Inspectable"
6import * as Layer from "effect/Layer"
7import * as RpcClient from "effect/rpc/RpcClient"
8import type { RpcClientError } from "effect/rpc/RpcClientError"
9import type * as RpcGroup from "effect/rpc/RpcGroup"
10import * as RpcServer from "effect/rpc/RpcServer"
11import * as RpcTest from "effect/rpc/RpcTest"
12import { AuthClient, AuthLayer, TimingLayer, User, UserRpcs, UsersLayer } from "./rpc-schemas.ts"
13
14export class UsersClient extends Context.Service<
15 UsersClient,
16 RpcClient.RpcClient<RpcGroup.Rpcs<typeof UserRpcs>, RpcClientError>
17>()("UsersClient") {
18 static readonly layer = Layer.effect(UsersClient)(RpcClient.make(UserRpcs)).pipe(
19 Layer.provide(AuthClient)
20 )
21 static layerTest = Layer.effect(UsersClient)(RpcTest.makeClient(UserRpcs)).pipe(
22 Layer.provide([UsersLayer, AuthLayer, TimingLayer, AuthClient])
23 )
24}
25
26export const e2eSuite = <E>(
27 name: string,
28 layer: Layer.Layer<UsersClient | RpcServer.Protocol, E>,
29 concurrent = true
30) => {
31 describe(name, { concurrent, timeout: 30_000 }, () => {
32 it.effect("should get user", () =>
33 Effect.gen(function*() {
34 const client = yield* UsersClient
35 const user = yield* client.GetUser({ id: "1" })
36 assert.instanceOf(user, User)
37 assert.deepStrictEqual(user, new User({ id: "1", name: "Logged in user" }))
38 }).pipe(Effect.provide(layer)))
39
40 it.effect("should get deferred user", () =>
41 Effect.gen(function*() {
42 const client = yield* UsersClient
43 const user = yield* client.GetUserDeferred({ id: "1" })
44 assert.instanceOf(user, User)
45 assert.deepStrictEqual(user, new User({ id: "1", name: "John" }))
46 }).pipe(Effect.provide(layer)))
47
48 it.effect("nested method", () =>
49 Effect.gen(function*() {
50 const client = yield* UsersClient
51 yield* client["nested.test"]()
52 }).pipe(Effect.provide(layer)))
53
54 it.effect("should not flatten Option", () =>
55 Effect.gen(function*() {
56 const client = yield* UsersClient
57 const user = yield* client.GetUserOption({ id: "1" })
58 assert.deepStrictEqual(user, Option.some(new User({ id: "1", name: "John" })))
59 }).pipe(Effect.provide(layer)))
60
61 it.effect("headers", () =>
62 Effect.gen(function*() {
63 const client = yield* UsersClient
64 const user = yield* client.GetUser({ id: "1" })
65 assert.instanceOf(user, User)
66 assert.deepStrictEqual(user, new User({ id: "123", name: "Logged in user" }))
67 }).pipe(
68 RpcClient.withHeaders({ userId: "123" }),
69 Effect.provide(layer)
70 ))
71
72 it.live("Stream", () =>
73 Effect.gen(function*() {
74 const client = yield* UsersClient
75 const users: Array<User> = []
76 yield* client.StreamUsers({ id: "1" }).pipe(
77 Stream.take(5),
78 Stream.runForEach((user) =>
79 Effect.sync(() => {
80 users.push(user)
81 })
82 ),
83 Effect.forkChild
84 )
85
86 yield* Effect.sleep(2000)
87 assert.lengthOf(users, 5)
88
89 // test interrupts
90 const interrupts = yield* client.GetInterrupts()
91 assert.equal(interrupts, 1)
92
93 const { supportsAck } = yield* RpcServer.Protocol
94
95 // test backpressure
96 if (supportsAck) {
97 const emits = yield* client.GetEmits()
98 assert.equal(emits, 5)
99 }
100 }).pipe(Effect.provide(layer)), { timeout: 20000 })
101
102 it.effect("defect", () =>
103 Effect.gen(function*() {
104 const client = yield* UsersClient
105 const cause = yield* client.ProduceDefect().pipe(
106 Effect.sandbox,
107 Effect.flip
108 )
109 assert.deepStrictEqual(cause, Cause.die("boom"))
110 }).pipe(
111 RpcClient.withHeaders({ userId: "123" }),
112 Effect.provide(layer)
113 ))
114
115 it.live("never", () =>
116 Effect.gen(function*() {
117 const client = yield* UsersClient
118 const fiber = yield* client.Never().pipe(
119 Effect.forkChild
120 )
121 yield* Effect.sleep(500)
122 assert.isUndefined(fiber.pollUnsafe())
123
124 yield* Fiber.interrupt(fiber)
125 yield* Effect.sleep(100)
126
127 const { supportsAck } = yield* RpcServer.Protocol
128 if (supportsAck) {
129 const interrupts = yield* client.GetInterrupts()
130 assert.equal(interrupts, 1)
131 }
132 }).pipe(
133 RpcClient.withHeaders({ userId: "123" }),
134 Effect.provide(layer)
135 ))
136
137 it.effect("timing middleware", () =>
138 Effect.gen(function*() {
139 const client = yield* UsersClient
140 const result = yield* client.TimedMethod({ shouldFail: false })
141 assert.equal(result, 1)
142 yield* client.TimedMethod({ shouldFail: true }).pipe(Effect.exit)
143 const { count, defect, success } = yield* client.GetTimingMiddlewareMetrics()
144 assert.notEqual(count, 0)
145 assert.notEqual(defect, 0)
146 assert.notEqual(success, 0)
147 }).pipe(Effect.provide(layer)))
148 })
149}