TRPCExpressRouteLive

src/ImportError.ts

@@ -0,0 +1,11 @@
+import { Data } from "effect"
+
+
+export class ImportError extends Data.TaggedError("ImportError")<{
+    path:   string
+    cause:  unknown
+}> {
+    toString(): string {
+        return `Could not import '${ this.path }'`
+    }
+}

src/Layers/JSONWebToken.ts

@@ -1,5 +1,6 @@
 import { Context, Effect, Layer } from "effect"
 import type * as JWT from "jsonwebtoken"
+import { ImportError } from "../ImportError"
 
 
 export class JSONWebToken extends Context.Tag("JSONWebToken")<JSONWebToken, {
@@ -27,7 +28,7 @@ export class JSONWebToken extends Context.Tag("JSONWebToken")<JSONWebToken, {
 
 const importJWT = Effect.tryPromise({
     try: () => import("jsonwebtoken"),
-    catch: cause => new Error("Could not import 'jsonwebtoken'. Make sure it is installed.", { cause }),
+    catch: cause => new ImportError({ path: "jsonwebtoken", cause }),
 })
 
 export const JSONWebTokenLive = Layer.effect(JSONWebToken, importJWT.pipe(

src/Layers/OpenAIClient.ts

@@ -1,5 +1,6 @@
 import { Config, Context, Effect, Layer } from "effect"
 import type { OpenAI } from "openai"
+import { ImportError } from "../ImportError"
 
 
 export class OpenAIClient extends Context.Tag("OpenAIClient")<OpenAIClient, OpenAIClientService>() {}
@@ -28,7 +29,7 @@ export class OpenAIClientService {
 
 const importOpenAI = Effect.tryPromise({
     try: () => import("openai"),
-    catch: cause => new Error("Could not import 'openai'. Make sure it is installed.", { cause }),
+    catch: cause => new ImportError({ path: "openai", cause }),
 })
 
 export const OpenAIClientLive = (

src/Layers/express/ExpressApp.ts

@@ -1,5 +1,6 @@
 import { Config, Context, Effect, Layer } from "effect"
 import type { Express } from "express"
+import { ImportError } from "../../ImportError"
 
 
 export class ExpressApp extends Context.Tag("ExpressApp")<ExpressApp, Express>() {}
@@ -7,7 +8,7 @@ export class ExpressApp extends Context.Tag("ExpressApp")<ExpressApp, Express>()
 
 const importExpress = Effect.tryPromise({
     try: () => import("express"),
-    catch: cause => new Error("Could not import 'express'. Make sure it is installed.", { cause }),
+    catch: cause => new ImportError({ path: "express", cause }),
 })
 
 export const ExpressAppLive = (

src/Layers/express/ExpressNodeHTTPServer.ts

@@ -1,6 +1,7 @@
 import { Config, Context, Effect, Layer, Match } from "effect"
 import type { Server } from "node:http"
 import type { AddressInfo } from "node:net"
+import { ImportError } from "../../ImportError"
 import { ExpressApp } from "./ExpressApp"
 
 
@@ -9,7 +10,7 @@ export class ExpressNodeHTTPServer extends Context.Tag("ExpressNodeHTTPServer")<
 
 const importNodeHTTP = Effect.tryPromise({
     try: () => import("node:http"),
-    catch: cause => new Error("Could not import 'node:http'. Make sure you are using a runtime that implements Node APIs.", { cause }),
+    catch: cause => new ImportError({ path: "node:http", cause }),
 })
 
 const serverListeningMessage = Match.type<AddressInfo | string | null>().pipe(

src/Layers/trpc/TRPCExpressRoute.ts

@@ -1,15 +1,21 @@
-import { createExpressMiddleware } from "@trpc/server/adapters/express"
 import { Config, Effect, Layer } from "effect"
+import { ImportError } from "../../ImportError"
 import { ExpressApp } from "../express"
 import { TRPCUnknownContextCreator } from "./TRPCContextCreator"
 import { TRPCAnyRouter } from "./TRPCRouter"
 
 
+const importTRPCServerExpressAdapter = Effect.tryPromise({
+    try: () => import("@trpc/server/adapters/express"),
+    catch: cause => new ImportError({ path: "@trpc/server/adapters/express", cause }),
+})
+
 export const TRPCExpressRouteLive = (
     config: {
         readonly path: Config.Config<string>
     }
 ) => Layer.effectDiscard(Effect.gen(function*() {
+    const { createExpressMiddleware } = yield* importTRPCServerExpressAdapter
     const app = yield* ExpressApp.ExpressApp
 
     app.use(yield* config.path,

src/Layers/trpc/TRPCWebSocketServer.ts

@@ -0,0 +1,44 @@
+import { applyWSSHandler } from "@trpc/server/adapters/ws"
+import { Context, Effect, Layer } from "effect"
+import ws from "ws"
+import { ExpressHTTPServer } from "../http/ExpressHTTPServer.service"
+import { ServerConfig } from "../ServerConfig"
+import { TRPCContextCreator } from "../trpc/TRPCContextCreator.service"
+import { RPCRouter } from "./RPCRouter.service"
+
+
+export class TRPCWebSocketServer extends Context.Tag("TRPCWebSocketServer")<TRPCWebSocketServer, {
+    wss:      ws.Server
+    handler:  ReturnType<typeof applyWSSHandler>
+}>() {}
+
+export const RPCWebSocketServerLive = Layer.effect(RPCWebSocketServer, ServerConfig.rpcHTTPRoot.pipe(
+    Effect.flatMap(rpcHTTPRoot => Effect.acquireRelease(
+        Effect.gen(function*() {
+            yield* Effect.logInfo(`WebSocket server started on ${ rpcHTTPRoot }`)
+
+            const wss = new ws.WebSocketServer({
+                server:  yield* ExpressHTTPServer,
+                host:    rpcHTTPRoot,
+            })
+
+            return {
+                wss,
+                handler: applyWSSHandler({
+                    wss,
+                    router:         yield* RPCRouter,
+                    createContext:  (yield* TRPCContextCreator).createWebSocketContext,
+                }),
+            }
+        }),
+
+        ({ wss, handler }) => Effect.gen(function*() {
+            yield* Effect.logInfo(`WebSocket server on ${ rpcHTTPRoot } is stopping. Waiting for existing connections to end...`)
+
+            handler.broadcastReconnectNotification()
+            yield* Effect.async(resume => {
+                wss.close(() => resume(Effect.logInfo("WebSocket server closed")))
+            })
+        }),
+    ))
+))

src/Layers/trpc/createTRCPErrorMapper.ts

@@ -1,9 +1,10 @@
 import { Effect, type Cause } from "effect"
+import { ImportError } from "../../ImportError"
 
 
 const importTRPCServer = Effect.tryPromise({
     try: () => import("@trpc/server"),
-    catch: cause => new Error("Could not import '@trpc/server'. Make sure it is installed.", { cause }),
+    catch: cause => new ImportError({ path: "@trpc/server", cause }),
 })
 
 export const createTRCPErrorMapper = importTRPCServer.pipe(Effect.map(({ TRPCError }) =>

src/index.ts

@@ -1,3 +1,4 @@
+export * from "./ImportError"
 export * as Layers from "./Layers"
 export * as Schema from "./Schema"
 export * as Types from "./Types"