feat: use socket.io

close #65 close #66
golemfactory · May 20, 2024 · 9776c34 · 9776c34
1 parent 5fe165f
commit 9776c34
Show file tree

Hide file tree

Showing 29 changed files with 363 additions and 116 deletions.
diff --git a/.vscode/settings.json b/.vscode/settings.json
@@ -1,7 +1,3 @@
 {
-  "cSpell.words": [
-    "Fastify",
-    "Siwe",
-    "viem"
-  ]
-}
+  "cSpell.words": ["Fastify", "Siwe", "socketio", "viem"]
+}
diff --git a/backend/package.json b/backend/package.json
@@ -24,7 +24,7 @@
     "@golem-sdk/task-executor": "workspace:*",
     "@types/jsonwebtoken": "^9.0.6",
     "@types/uuid": "^9.0.8",
-        "@types/ws": "^8.5.10",
+    "@types/ws": "^8.5.10",
     "awilix": "^10.0.1",
     "chalk": "^5.3.0",
     "dayjs": "^1.11.10",
@@ -34,10 +34,12 @@
     "ftp-srv": "^4.6.3",
     "js-big-decimal": "^2.0.7",
     "jsonwebtoken": "^9.0.2",
+    "jwt-decode": "^4.0.0",
     "loginWithCrypto": "workspace:*",
     "mongodb": "^6.3.0",
     "mongoose": "^8.2.0",
     "rxjs": "^7.8.1",
+    "socket.io": "^4.7.5",
     "uuid": "^9.0.1",
     "viem": "^2.7.22",
     "ya-ts-client": "workspace:*"

diff --git a/backend/src/fastify.ts b/backend/src/fastify.ts
@@ -10,7 +10,7 @@ import { FastifySSEPlugin } from "fastify-sse-v2";
 import fastifyMultipart from "@fastify/multipart";
 import { Yagna } from "./services/yagna/routes.js";
 import websocket from "@fastify/websocket";
-
+import socketioServer from "./socket.io.js";
 export const startupFastifyServer = async (): Promise<FastifyInstance> => {
   const fastify = Fastify({
     logger: false,
@@ -24,6 +24,11 @@ export const startupFastifyServer = async (): Promise<FastifyInstance> => {
   fastify.register(fastifyMultipart);
   fastify.register(FastifySSEPlugin);
   fastify.register(websocket);
+  fastify.register(socketioServer, {
+    cors: {
+      origin: "*",
+    },
+  });
   fastify.register(cors, {
     origin: "*",
   });

diff --git a/backend/src/services/payment/routes.ts b/backend/src/services/payment/routes.ts
@@ -18,7 +18,6 @@ export const paymentService = fastifyPlugin(
       //@ts-ignore TODO: add declaration in auth module so ts-ignore is not needed
       onRequest: [fastify.authenticate],
       handler: async (request, reply) => {
-        console.log("wtf creating deposit"); 
         const paymentService = container.cradle.paymentService;
         // @ts-ignore
         // TODO: make sure request.body is the right type
@@ -28,7 +27,7 @@ export const paymentService = fastifyPlugin(
           request.body.nonce
         );
 
-        console.log("res",res); 
+        console.log("res", res);
         return res;
       },
     });

diff --git a/backend/src/services/user/routes.ts b/backend/src/services/user/routes.ts
@@ -1,38 +1,75 @@
 import fastify, { FastifyInstance, FastifyRequest } from "fastify";
 import { container } from "../../di.js";
 import fastifyPlugin from "fastify-plugin";
-
+import { userModel } from "./model.js";
+import { jwtDecode } from "jwt-decode";
+import { set } from "mongoose";
 export const userService = fastifyPlugin(
   (fastify: FastifyInstance, opts, done) => {
-    fastify.get("/me", {
-      onRequest: [fastify.authenticate],
-      handler: async (request: FastifyRequest, reply) => {
-        const userService = container.cradle.userService;
-        const requestUser = request.user;
-        const user = await userService.findById(requestUser._id);
-        if (!user) {
-          throw new Error(`User not found with id ${requestUser._id}`);
-        }
-        return {
-          walletAddress: user.walletAddress,
-          _id: user._id.toString(),
-          nonce: user.nonce.toString(),
-          currentAllocation: {
-            id: user.currentAllocationId,
-          },
-          currentActivity: {
-            id: user.currentActivityId,
-          },
-          deposits: user.deposits.map((d) => {
-            return {
-              isCurrent: d.isCurrent,
-              isValid: d.isValid,
-              nonce: d.nonce.toString(),
-            };
-          }),
-        };
-      },
+    fastify.io.of("/me").use((socket, next) => {
+      const token = socket.handshake.auth.token;
+      if (!token) {
+        next(new Error("Authentication error"));
+      }
+      next();
     });
+
+    fastify.io.of("/me").on("connection", async (socket) => {
+      console.log("socket", socket.handshake.auth.token);
+
+      const user = jwtDecode<{
+        _id: string;
+      }>(socket.handshake.auth.token);
+
+      if (!user._id) {
+        throw new Error(`Wrong token`);
+      }
+      if (!user) {
+        throw new Error(
+          `User not found with id ${socket.handshake.auth.token}`
+        );
+      }
+
+      const userDTO = await container.cradle.userService.getUserDTO(user._id);
+      socket.emit("user", userDTO);
+      userModel
+        .watch(
+          [
+            {
+              $match: {
+                _id: user._id,
+              },
+            },
+          ],
+          {
+            fullDocument: "updateLookup",
+          }
+        )
+        .on("change", async () => {
+          const userDTO = await container.cradle.userService.getUserDTO(
+            user._id
+          );
+          socket.emit("user", userDTO);
+        });
+    });
+    // fastify.get(
+    //   "/me",
+    //   {
+    //     websocket: true,
+    //     onRequest: [fastify.authenticate],
+    //   },
+    //   async (socket, request) => {
+    //     console.log("socket", request.user._id);
+    //     const userService = container.cradle.userService;
+    //     const requestUser = request.user;
+    //     const user = await userService.findById(requestUser._id);
+    //     console.log("user", user);
+    //     if (!user) {
+    //       throw new Error(`User not found with id ${requestUser._id}`);
+    //     }
+
+    //   }
+    // );
     done();
   }
 );
diff --git a/backend/src/services/user/service.ts b/backend/src/services/user/service.ts
@@ -100,7 +100,7 @@ export const userService: IUserService = {
   },
 
   addDeposit: async (userId: string, deposit: Deposit) => {
-    console.log("adding deposit"); 
+    console.log("adding deposit");
     await userModel.updateOne(
       { _id: userId },
       {

diff --git a/backend/src/services/yagna/routes.ts b/backend/src/services/yagna/routes.ts
@@ -8,15 +8,16 @@ export const Yagna = fastifyPlugin((fastify: FastifyInstance, opts, done) => {
     handler: async (request, reply) => {
       const requestUser = request.user;
       const Yagna = container.cradle.Yagna;
-      try {
-        await Yagna.createExecutor(requestUser._id);
-      } catch (e) {
-        console.log("error", e);
-        reply.code(500).send({
-          message: "Unable to create executor",
-        });
-      }
-      const allocation = await Yagna.getUserAllocation(requestUser._id);
+      // try {
+      //   await Yagna.createExecutor(requestUser._id);
+      // } catch (e) {
+      //   console.log("error", e);
+      //   reply.code(500).send({
+      //     message: "Unable to create executor",
+      //   });
+      // }
+      const allocation = await Yagna.createUserAllocation(requestUser._id);
+      // const allocation = await Yagna.getUserAllocation(requestUser._id);
       if (!allocation) {
         reply.code(500).send({
           message: "Unable to create allocation",

diff --git a/backend/src/services/yagna/service.ts b/backend/src/services/yagna/service.ts
@@ -10,9 +10,9 @@ import bigDecimal from "js-big-decimal";
 const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
 
 import { Worker } from "./worker.js";
-import { WorkContext } from "@golem-sdk/golem-js"
+import { WorkContext } from "@golem-sdk/golem-js";
 import { TaskExecutor } from "@golem-sdk/task-executor";
-import { formatEther } from "viem";
+import { formatEther, parseEther } from "viem";
 export class Yagna {
   public debitNoteEvents: Subject<any>;
   private paymentService: YaTsClient.PaymentApi.RequestorService;
@@ -80,6 +80,43 @@ export class Yagna {
     }).default;
   }
 
+  async createUserAllocation(userId: string) {
+    debugLog("payments", "creating user allocation", userId);
+    const userService = container.cradle.userService;
+    const userDeposit = await userService.getCurrentDeposit(userId);
+
+    if (!userDeposit) {
+      throw new Error({ code: ErrorCode.NO_DEPOSIT });
+    }
+    console.log("userDeposit", userDeposit);
+    try {
+      console.log("trying");
+      //@ts-ignore
+
+      const allocation = await this.paymentService.createAllocation({
+        totalAmount: formatEther(userDeposit.amount),
+        makeDeposit: false,
+        deposit: {
+          contract: container.cradle.contractAddress,
+          id: userDeposit.id.toString(16),
+          validate: {
+            flatFeeAmount: parseEther("23").toString(),
+          },
+        },
+      });
+
+      console.log("allocation", allocation);
+
+      container.cradle.userService.setCurrentAllocationId(
+        userId,
+        allocation.allocationId
+      );
+      return allocation;
+    } catch (e) {
+      console.log("error", e);
+    }
+    // @ts-ignore
+  }
   async getUserAllocation(userId: string) {
     debugLog("payments", "getting user allocation", userId);
     const user = await container.cradle.userService.getUserById(userId);
@@ -169,8 +206,8 @@ export class Yagna {
       agreementMaxPoolSize: 1,
       //@ts-ignore
       budget: formatEther(userDeposit.amount),
-      enableLogging : true, 
-      subnetTag : 'pociej_own'
+      enableLogging: true,
+      subnetTag: "pociej_own",
     });
 
     this.userContext.setExecutor(userId, executor);
@@ -239,7 +276,7 @@ export class Yagna {
           })
           .catch((e: any) => {
             console.log("Error", e);
-            reject(); 
+            reject();
           });
       }
     });

diff --git a/backend/src/socket.io.ts b/backend/src/socket.io.ts
@@ -0,0 +1,42 @@
+import { FastifyInstance, FastifyPluginAsync } from "fastify";
+import fp from "fastify-plugin";
+import { Server, ServerOptions } from "socket.io";
+import { IUser } from "./services/user/types.js";
+
+const fastifySocketIO: FastifyPluginAsync<Partial<ServerOptions>> = fp(
+  async function (fastify, opts) {
+    const server = new Server<
+      {},
+      {
+        user: () => void;
+      }
+    >({
+      cors: {
+        origin: "*",
+      },
+    });
+
+    server.listen(Number(process.env.PORT) || 5174);
+
+    fastify.decorate("io", server);
+
+    fastify.addHook("onClose", (fastify: FastifyInstance, done) => {
+      (fastify as any).io.close();
+      done();
+    });
+  },
+  { fastify: ">=4.x.x", name: "fastify-socket.io" }
+);
+
+declare module "fastify" {
+  interface FastifyInstance {
+    io: Server<
+      {},
+      {
+        user: (data: IUser) => void;
+      }
+    >;
+  }
+}
+
+export default fastifySocketIO;
diff --git a/frontend/.env.development b/frontend/.env.development
@@ -1 +1,2 @@
-VITE_BACKEND_URL=http://127.0.0.1:5174/api
+VITE_BACKEND_HTTP_URL=http://127.0.0.1:5174/api
+VITE_BACKEND_WS_URL=127.0.0.1:5174
diff --git a/frontend/.env.production b/frontend/.env.production
@@ -1 +1,2 @@
-VITE_BACKEND_URL=http://0.0.0.0:5174/api
+VITE_BACKEND_HTTP_URL=https://0.0.0.0:5174/api
+VITE_BACKEND_WS_URL=wss://0.0.0.0:5174/api
diff --git a/frontend/package.json b/frontend/package.json
@@ -28,6 +28,7 @@
     "react-daisyui": "^5.0.0",
     "react-dom": "^18.2.0",
     "siwe": "^2.3.2",
+    "socket.io-client": "^4.7.5",
     "swr": "^2.2.5",
     "ts-pattern": "^5.1.1",
     "usehooks-ts": "^3.1.0",