1import { Context, Effect, Layer, Metric, Option, Queue, Schema } from "effect"
2import * as Headers from "effect/http/Headers"
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("GetUserOption", {
44 success: Schema.Option(User),
45 payload: { id: Schema.String }
46 }),
47 StreamUsers,
48 Rpc.make("GetInterrupts", {
49 success: Schema.Number
50 }),
51 Rpc.make("GetEmits", {
52 success: Schema.Number
53 }),
54 Rpc.make("ProduceDefect"),
55 Rpc.make("Never"),
56 Rpc.make("nested.test"),
57 Rpc.make("TimedMethod", {
58 payload: {
59 shouldFail: Schema.Boolean
60 },
61 success: Schema.Number
62 }).middleware(TimingMiddleware),
63 Rpc.make("GetTimingMiddlewareMetrics", {
64 success: Schema.Struct({
65 success: Schema.Number,
66 defect: Schema.Number,
67 count: Schema.Number
68 })
69 })
70).middleware(AuthMiddleware)
71
72export const AuthLayer = Layer.succeed(AuthMiddleware)(
73 AuthMiddleware.of((effect, options) =>
74 Effect.provideService(
75 effect,
76 CurrentUser,
77 new User({ id: options.headers.userid ?? "1", name: options.headers.name ?? "Fallback name" })
78 )
79 )
80)
81
82const rpcSuccesses = Metric.counter("rpc_middleware_success")
83const rpcDefects = Metric.counter("rpc_middleware_defects")
84const rpcCount = Metric.counter("rpc_middleware_count")
85export const TimingLayer = Layer.succeed(TimingMiddleware)(
86 TimingMiddleware.of((effect) =>
87 effect.pipe(
88 Effect.tap(Metric.update(rpcSuccesses, 1)),
89 Effect.tapDefect(() => Metric.update(rpcDefects, 1)),
90 Effect.ensuring(Metric.update(rpcCount, 1))
91 )
92 )
93)
94
95export const UsersLayer = UserRpcs.toLayer(Effect.gen(function*() {
96 let interrupts = 0
97 let emits = 0
98 return UserRpcs.of({
99 GetUser: (_) =>
100 CurrentUser.pipe(
101 Rpc.fork
102 ),
103 GetUserOption: Effect.fnUntraced(function*(req) {
104 return Option.some(new User({ id: req.id, name: "John" }))
105 }),
106 StreamUsers: Effect.fnUntraced(function*(req, _) {
107 const mailbox = yield* Queue.bounded<User>(0)
108
109 yield* Effect.addFinalizer(() =>
110 Effect.sync(() => {
111 interrupts++
112 })
113 )
114
115 yield* Queue.offer(mailbox, new User({ id: req.id, name: "John" })).pipe(
116 Effect.tap(() =>
117 Effect.sync(() => {
118 emits++
119 })
120 ),
121 Effect.delay(100),
122 Effect.forever,
123 Effect.forkScoped
124 )
125
126 return mailbox
127 }),
128 GetInterrupts: () => Effect.sync(() => interrupts),
129 GetEmits: () => Effect.sync(() => emits),
130 ProduceDefect: () => Effect.die("boom"),
131 Never: () => Effect.never.pipe(Effect.onInterrupt(() => Effect.sync(() => interrupts++))),
132 "nested.test": () => Effect.void,
133 TimedMethod: (_) => _.shouldFail ? Effect.die("boom") : Effect.succeed(1),
134 GetTimingMiddlewareMetrics: () =>
135 Effect.all({
136 defect: Metric.value(rpcDefects).pipe(Effect.map((_) => _.count)),
137 success: Metric.value(rpcSuccesses).pipe(Effect.map((_) => _.count)),
138 count: Metric.value(rpcCount).pipe(Effect.map((_) => _.count))
139 })
140 })
141}))
142
143export const RpcLayer = RpcServer.layer(UserRpcs, {
144 disableFatalDefects: true
145}).pipe(
146 Layer.provide([
147 UsersLayer,
148 AuthLayer,
149 TimingLayer
150 ])
151)
152
153export const AuthClient = RpcMiddleware.layerClient(AuthMiddleware, ({ next, request }) =>
154 next({
155 ...request,
156 headers: Headers.set(request.headers, "name", "Logged in user")
157 }))