feat: add trpc websocket (#205)

packages/api/package.json

@@ -10,7 +10,9 @@
   "main": "./index.ts",
   "types": "./index.ts",
   "license": "MIT",
+  "type": "module",
   "scripts": {
+    "dev": "pnpm tsx ./src/wssDevServer.ts",
     "clean": "rm -rf .turbo node_modules",
     "lint": "eslint .",
     "format": "prettier --check . --ignore-path ../../.gitignore",
@@ -18,17 +20,20 @@
   },
   "dependencies": {
     "@homarr/auth": "workspace:^0.1.0",
-    "@homarr/definitions": "workspace:^0.1.0",
     "@homarr/db": "workspace:^0.1.0",
+    "@homarr/definitions": "workspace:^0.1.0",
+    "@homarr/log": "workspace:^",
     "@homarr/validation": "workspace:^0.1.0",
     "@trpc/client": "next",
     "@trpc/server": "next",
-    "superjson": "2.2.1"
+    "superjson": "2.2.1",
+    "ws": "^8.16.0"
   },
   "devDependencies": {
     "@homarr/eslint-config": "workspace:^0.2.0",
     "@homarr/prettier-config": "workspace:^0.1.0",
     "@homarr/tsconfig": "workspace:^0.1.0",
+    "@types/ws": "^8.5.10",
     "eslint": "^8.57.0",
     "prettier": "^3.2.5",
     "typescript": "^5.4.2"

packages/api/src/router/user.ts

@@ -1,6 +1,5 @@
-import "server-only";
-
 import { TRPCError } from "@trpc/server";
+import { observable } from "@trpc/server/observable";
 
 import { createSalt, hashPassword } from "@homarr/auth";
 import type { Database } from "@homarr/db";
@@ -99,6 +98,15 @@ export const userRouter = createTRPCRouter({
         })
         .where(eq(users.id, input.userId));
     }),
+  test: publicProcedure.subscription(() => {
+    return observable<number>((emit) => {
+      let counter = 0;
+      setInterval(() => {
+        counter = counter + 1;
+        emit.next(counter);
+      }, 1000);
+    });
+  }),
 });
 
 const createUser = async (

packages/api/src/trpc.ts

@@ -12,6 +12,7 @@ import superjson from "superjson";
 import type { Session } from "@homarr/auth";
 import { auth } from "@homarr/auth";
 import { db } from "@homarr/db";
+import { logger } from "@homarr/log";
 import { ZodError } from "@homarr/validation";
 
 /**
@@ -33,7 +34,10 @@ export const createTRPCContext = async (opts: {
   const session = opts.session ?? (await auth());
   const source = opts.headers.get("x-trpc-source") ?? "unknown";
 
-  console.log(">>> tRPC Request from", source, "by", session?.user);
+  logger.info(
+    `tRPC request from ${source} by user '${session?.user.id}'`,
+    session?.user,
+  );
 
   return {
     session,

packages/api/src/wssDevServer.ts

@@ -0,0 +1,53 @@
+import { applyWSSHandler } from "@trpc/server/adapters/ws";
+import { WebSocketServer } from "ws";
+
+import { logger } from "@homarr/log";
+
+import { appRouter } from "./root";
+import { createTRPCContext } from "./trpc";
+
+const wss = new WebSocketServer({
+  port: 3001,
+});
+const handler = applyWSSHandler({
+  wss,
+  router: appRouter,
+  createContext: ({ req }) => {
+    return createTRPCContext({
+      headers: {
+        ...req.headers,
+        get(key: string) {
+          const item = req.headers[key];
+          return typeof item === "string" ? item ?? null : item?.at(0) ?? null;
+        },
+      } as Headers,
+      session: {
+        // TODO: replace with actual session
+        user: {
+          id: "1",
+          name: "Test User",
+          email: "",
+        },
+        expires: new Date().toISOString(),
+      },
+    });
+  },
+});
+
+wss.on("connection", (ws, incomingMessage) => {
+  logger.info(
+    `➕ Connection (${wss.clients.size}) ${incomingMessage.method} ${incomingMessage.url}`,
+  );
+  ws.once("close", (code, reason) => {
+    logger.info(
+      `➖ Connection (${wss.clients.size}) ${code} ${reason.toString()}`,
+    );
+  });
+});
+logger.info("✅ WebSocket Server listening on ws://localhost:3001");
+
+process.on("SIGTERM", () => {
+  logger.info("SIGTERM");
+  handler.broadcastReconnectNotification();
+  wss.close();
+});