1import { Context, Deferred, Effect, Layer, Metric, Option, Queue, Schema } from "effect"
2import { Headers } from "effect/http"
3import * as Rpc from "effect/rpc/Rpc"
4import * as RpcGroup from "effect/rpc/RpcGroup"
5import * as RpcMiddleware from "effect/rpc/RpcMiddleware"
6import * as RpcServer from "effect/rpc/RpcServer"
7
8export class User extends Schema.Class<User>("User")({
9 id: Schema.String,
10 name: Schema.String
11}) {}
12
13class StreamUsers extends Rpc.make("StreamUsers", {
14 success: User,
15 payload: {
16 id: Schema.String
17 },
18 stream: true
19}) {}
20
21class CurrentUser extends Context.Service<CurrentUser, User>()("CurrentUser") {}
22
23class Unauthorized extends Schema.Error<Unauthorized>("Unauthorized")({
24 _tag: Schema.tag("Unauthorized")
25}) {}
26
27class AuthMiddleware extends RpcMiddleware.Service<AuthMiddleware, {
28 provides: CurrentUser
29}>()("AuthMiddleware", {
30 error: Unauthorized,
31 requiredForClient: true
32}) {}
33
34class TimingMiddleware extends RpcMiddleware.Service<TimingMiddleware>()("TimingMiddleware") {}
35
36class GetUser extends Rpc.make("GetUser", {
37 success: User,
38 payload: { id: Schema.String }
39}) {}
40
41export const UserRpcs = RpcGroup.make(
42 GetUser,
43 Rpc.make("GetUserDeferred", {
44 success: User,
45 payload: { id: Schema.String }
46 }),
47 Rpc.make("GetUserOption", {
48 success: Schema.Option(User),
49 payload: { id: Schema.String }
50 }),
51 StreamUsers,
52 Rpc.make("GetInterrupts", {
53 success: Schema.Number
54 }),
55 Rpc.make("GetEmits", {
56 success: Schema.Number
57 }),
58 Rpc.make("ProduceDefect"),
59 Rpc.make("ProduceDefectCustom", {
60 defect: Schema.Defect({ includeStack: true })
61 }),
62 Rpc.make("Never"),
63 Rpc.make("nested.test"),
64 Rpc.make("TimedMethod", {
65 payload: {
66 shouldFail: Schema.Boolean
67 },
68 success: Schema.Number
69 }).middleware(TimingMiddleware),
70 Rpc.make("GetTimingMiddlewareMetrics", {
71 success: Schema.Struct({
72 success: Schema.Number,
73 defect: Schema.Number,
74 count: Schema.Number
75 })
76 })
77).middleware(AuthMiddleware)
78
79export const AuthLayer = Layer.succeed(AuthMiddleware)(
80 AuthMiddleware.of((effect, options) =>
81 Effect.provideService(
82 effect,
83 CurrentUser,
84 new User({ id: options.headers.userid ?? "1", name: options.headers.name ?? "Fallback name" })
85 )
86 )
87)
88
89const rpcSuccesses = Metric.counter("rpc_middleware_success")
90const rpcDefects = Metric.counter("rpc_middleware_defects")
91const rpcCount = Metric.counter("rpc_middleware_count")
92export const TimingLayer = Layer.succeed(TimingMiddleware)(
93 TimingMiddleware.of((effect) =>
94 effect.pipe(
95 Effect.tap(Metric.update(rpcSuccesses, 1)),
96 Effect.tapDefect(() => Metric.update(rpcDefects, 1)),
97 Effect.ensuring(Metric.update(rpcCount, 1))
98 )
99 )
100)
101
102export const UsersLayer = UserRpcs.toLayer(Effect.gen(function*() {
103 let interrupts = 0
104 let emits = 0
105 return UserRpcs.of({
106 GetUser: (_) =>
107 CurrentUser.pipe(
108 Rpc.fork
109 ),
110 GetUserDeferred(_) {
111 const deferred = Deferred.makeUnsafe<User>()
112 Deferred.doneUnsafe(deferred, Effect.succeed(new User({ id: "1", name: "John" })))
113 return Effect.succeed(deferred)
114 },
115 GetUserOption: Effect.fnUntraced(function*(req) {
116 return Option.some(new User({ id: req.id, name: "John" }))
117 }),
118 StreamUsers: Effect.fnUntraced(function*(req, _) {
119 const mailbox = yield* Queue.bounded<User>(0)
120
121 yield* Effect.addFinalizer(() =>
122 Effect.sync(() => {
123 interrupts++
124 })
125 )
126
127 yield* Queue.offer(mailbox, new User({ id: req.id, name: "John" })).pipe(
128 Effect.tap(() =>
129 Effect.sync(() => {
130 emits++
131 })
132 ),
133 Effect.delay(100),
134 Effect.forever,
135 Effect.forkScoped
136 )
137
138 return mailbox
139 }),
140 GetInterrupts: () => Effect.sync(() => interrupts),
141 GetEmits: () => Effect.sync(() => emits),
142 ProduceDefect: () => Effect.die("boom"),
143 ProduceDefectCustom: () =>
144 Effect.die({
145 message: "detailed error",
146 stack: "Error: detailed error\n at handler.ts:1",
147 name: "CustomDefect"
148 }),
149 Never: () => Effect.never.pipe(Effect.onInterrupt(() => Effect.sync(() => interrupts++))),
150 "nested.test": () => Effect.void,
151 TimedMethod: (_) => _.shouldFail ? Effect.die("boom") : Effect.succeed(1),
152 GetTimingMiddlewareMetrics: () =>
153 Effect.all({
154 defect: Metric.value(rpcDefects).pipe(Effect.map((_) => _.count)),
155 success: Metric.value(rpcSuccesses).pipe(Effect.map((_) => _.count)),
156 count: Metric.value(rpcCount).pipe(Effect.map((_) => _.count))
157 })
158 })
159}))
160
161export const RpcLayer = RpcServer.layer(UserRpcs, {
162 disableFatalDefects: true
163}).pipe(
164 Layer.provide([
165 UsersLayer,
166 AuthLayer,
167 TimingLayer
168 ])
169)
170
171export const AuthClient = RpcMiddleware.layerClient(AuthMiddleware, ({ next, request }) =>
172 next({
173 ...request,
174 headers: Headers.set(request.headers, "name", "Logged in user")
175 }))