import { Context, Deferred, Effect, Layer, Metric, Option, Queue, Schema } from "effect" import { Headers } from "effect/unstable/http" import * as Rpc from "effect/unstable/rpc/Rpc" import * as RpcGroup from "effect/unstable/rpc/RpcGroup" import * as RpcMiddleware from "effect/unstable/rpc/RpcMiddleware" import * as RpcServer from "effect/unstable/rpc/RpcServer" export class User extends Schema.Class("User")({ id: Schema.String, name: Schema.String }) {} class StreamUsers extends Rpc.make("StreamUsers", { success: User, payload: { id: Schema.String }, stream: true }) {} class CurrentUser extends Context.Service()("CurrentUser") {} class Unauthorized extends Schema.ErrorClass("Unauthorized")({ _tag: Schema.tag("Unauthorized") }) {} class AuthMiddleware extends RpcMiddleware.Service()("AuthMiddleware", { error: Unauthorized, requiredForClient: true }) {} class TimingMiddleware extends RpcMiddleware.Service()("TimingMiddleware") {} class GetUser extends Rpc.make("GetUser", { success: User, payload: { id: Schema.String } }) {} export const UserRpcs = RpcGroup.make( GetUser, Rpc.make("GetUserDeferred", { success: User, payload: { id: Schema.String } }), Rpc.make("GetUserOption", { success: Schema.Option(User), payload: { id: Schema.String } }), StreamUsers, Rpc.make("GetInterrupts", { success: Schema.Number }), Rpc.make("GetEmits", { success: Schema.Number }), Rpc.make("ProduceDefect"), Rpc.make("ProduceDefectCustom", { defect: Schema.Defect({ includeStack: true }) }), Rpc.make("Never"), Rpc.make("nested.test"), Rpc.make("TimedMethod", { payload: { shouldFail: Schema.Boolean }, success: Schema.Number }).middleware(TimingMiddleware), Rpc.make("GetTimingMiddlewareMetrics", { success: Schema.Struct({ success: Schema.Number, defect: Schema.Number, count: Schema.Number }) }) ).middleware(AuthMiddleware) export const AuthLive = Layer.succeed(AuthMiddleware)( AuthMiddleware.of((effect, options) => Effect.provideService( effect, CurrentUser, new User({ id: options.headers.userid ?? "1", name: options.headers.name ?? "Fallback name" }) ) ) ) const rpcSuccesses = Metric.counter("rpc_middleware_success") const rpcDefects = Metric.counter("rpc_middleware_defects") const rpcCount = Metric.counter("rpc_middleware_count") export const TimingLive = Layer.succeed(TimingMiddleware)( TimingMiddleware.of((effect) => effect.pipe( Effect.tap(Metric.update(rpcSuccesses, 1)), Effect.tapDefect(() => Metric.update(rpcDefects, 1)), Effect.ensuring(Metric.update(rpcCount, 1)) ) ) ) export const UsersLive = UserRpcs.toLayer(Effect.gen(function*() { let interrupts = 0 let emits = 0 return UserRpcs.of({ GetUser: (_) => CurrentUser.pipe( Rpc.fork ), GetUserDeferred(_) { const deferred = Deferred.makeUnsafe() Deferred.doneUnsafe(deferred, Effect.succeed(new User({ id: "1", name: "John" }))) return Effect.succeed(deferred) }, GetUserOption: Effect.fnUntraced(function*(req) { return Option.some(new User({ id: req.id, name: "John" })) }), StreamUsers: Effect.fnUntraced(function*(req, _) { const mailbox = yield* Queue.bounded(0) yield* Effect.addFinalizer(() => Effect.sync(() => { interrupts++ }) ) yield* Queue.offer(mailbox, new User({ id: req.id, name: "John" })).pipe( Effect.tap(() => Effect.sync(() => { emits++ }) ), Effect.delay(100), Effect.forever, Effect.forkScoped ) return mailbox }), GetInterrupts: () => Effect.sync(() => interrupts), GetEmits: () => Effect.sync(() => emits), ProduceDefect: () => Effect.die("boom"), ProduceDefectCustom: () => Effect.die({ message: "detailed error", stack: "Error: detailed error\n at handler.ts:1", name: "CustomDefect" }), Never: () => Effect.never.pipe(Effect.onInterrupt(() => Effect.sync(() => interrupts++))), "nested.test": () => Effect.void, TimedMethod: (_) => _.shouldFail ? Effect.die("boom") : Effect.succeed(1), GetTimingMiddlewareMetrics: () => Effect.all({ defect: Metric.value(rpcDefects).pipe(Effect.map((_) => _.count)), success: Metric.value(rpcSuccesses).pipe(Effect.map((_) => _.count)), count: Metric.value(rpcCount).pipe(Effect.map((_) => _.count)) }) }) })) export const RpcLive = RpcServer.layer(UserRpcs, { disableFatalDefects: true }).pipe( Layer.provide([ UsersLive, AuthLive, TimingLive ]) ) export const AuthClient = RpcMiddleware.layerClient(AuthMiddleware, ({ next, request }) => next({ ...request, headers: Headers.set(request.headers, "name", "Logged in user") }))