feat: implement session management with SessionService and SessionPlugin; refactor query execution to handle session cookies; update token and query handling for improved session tracking

src/query/executer/query.executer.service.ts

@@ -15,6 +15,7 @@ import { InjectQueue } from "@nestjs/bullmq";
 import { QUEUE_NAMES } from "src/queue/constants";
 import { Queue, QueueEvents } from "bullmq";
 import { FunctionService } from "src/query/function/function.service";
+import { SessionService } from "../session/session.service";
 
 @Injectable()
 export class QueryExecuterService {
@@ -27,6 +28,7 @@ export class QueryExecuterService {
     private readonly functionService: FunctionService,
     readonly databaseManagerService: DatabaseManagerService,
     readonly redisNodeService: RedisNodeService,
+    readonly sessionService: SessionService,
     @InjectQueue(QUEUE_NAMES.QUERY) private queryQueue: Queue
   ) {
     this.queueEvents = new QueueEvents(this.queryQueue.name);
@@ -54,7 +56,8 @@ export class QueryExecuterService {
   async runQueryQueued(
     token: string,
     queryData: any,
-    headers: Record<string, any> = {}
+    headers: Record<string, any> = {},
+    cookies: Record<string, any> = {}
   ): Promise<QueryResponse> {
     const job = await this.queryQueue.add(
       `${new Date().getTime()}_${token}`,
@@ -62,6 +65,7 @@ export class QueryExecuterService {
         token,
         queryData,
         headers,
+        cookies,
       },
       {
         removeOnComplete: true,
@@ -77,7 +81,8 @@ export class QueryExecuterService {
   async runQuery(
     token: string,
     queryData: any,
-    headers: Record<string, any> = {}
+    headers: Record<string, any> = {},
+    cookies: Record<string, any> = {}
   ): Promise<QueryResponse> {
     const query = await this.queryRepository.findOne({
       where: { id: token },
@@ -88,7 +93,16 @@ export class QueryExecuterService {
       throw new Error("Query not found");
     }
 
-    const vm = await this.createVm(query);
+    const sessionId = cookies["x-session-id"] || null;
+
+    console.log("Session ID:", sessionId);
+
+    if (!sessionId) {
+      const session = await this.sessionService.create(query.project.id);
+      cookies["x-session-id"] = session.sessionId;
+    }
+
+    const vm = await this.createVm(query, cookies["x-session-id"]);
     const result = await vm.runScript(
       this.clearImports(query.source),
       queryData,
@@ -99,10 +113,20 @@ export class QueryExecuterService {
       throw new Error(`Error initializing VM: ${JSON.stringify(result)}`);
     }
 
+    if (!result?.cookies || !result?.cookies["x-session-id"]) {
+      if (result.cookies === undefined) {
+        result.cookies = {};
+      }
+
+      result.cookies["x-session-id"] = (
+        await this.sessionService.get(cookies["x-session-id"], query.project.id)
+      ).sessionId;
+    }
+
     return result;
   }
 
-  private async createVm(query: Query) {
+  private async createVm(query: Query, sessionId: string = null) {
     const imports = this.parseImports(query.source);
     const importsParsed = imports.map((imp) => {
       const item = imp.split("/");
@@ -140,7 +164,9 @@ export class QueryExecuterService {
         throw new Error(`Plugin ${plugin.name} not found`);
       }
 
-      plugins.push(await registeredPlugins[plugin.name](this, query));
+      plugins.push(
+        await registeredPlugins[plugin.name](this, query, sessionId)
+      );
     }
 
     const vm = new Vm({