Merge pull request #36 from game-node-app/dev

Dev

.env.example

@@ -8,3 +8,4 @@ DOMAIN_WEBSITE=localhost:3000
 SUPERTOKENS_CORE_URI=http://localhost:3567
 REDIS_URL=redis://localhost:6379
 SERVER_PORT=5000
+RABBITMQ_URI=amqp://rabbitmq:rabbitmq@localhost:5672

docker-compose.prod.yml

@@ -1,47 +1,6 @@
 version: '3'
 
 services:
-  supertokens-db:
-    hostname: supertokens-db
-    image: mysql:latest
-    environment:
-      MYSQL_ROOT_PASSWORD: ${SUPERTOKENS_DB_PASS}
-      MYSQL_USER: ${SUPERTOKENS_DB_USER}
-      MYSQL_PASSWORD: ${SUPERTOKENS_DB_PASS}
-      MYSQL_DATABASE: supertokens
-
-    networks:
-      - game_node_app
-    restart: always
-    volumes:
-      - supertokens-db:/var/lib/mysql
-    healthcheck:
-      test: [ "CMD", "mysqladmin", "ping", "-h", "localhost" ]
-      timeout: 20s
-      retries: 10
-
-  supertokens:
-    hostname: supertokens
-    image: registry.supertokens.io/supertokens/supertokens-mysql:7.0
-
-    depends_on:
-      - supertokens-db
-
-    environment:
-      MYSQL_CONNECTION_URI: mysql://${SUPERTOKENS_DB_USER}:${SUPERTOKENS_DB_PASS}@supertokens-db:3306/supertokens
-
-    networks:
-      - game_node_app
-    restart: always
-
-    healthcheck:
-      test: >
-        bash -c 'exec 3<>/dev/tcp/127.0.0.1/3567 && echo -e "GET /hello HTTP/1.1\r\nhost: 127.0.0.1:3567\r\nConnection: close\r\n\r\n" >&3 && cat <&3 | grep "Hello"'
-      interval: 10s
-      timeout: 5s
-      retries: 5
-
-
   db:
     image: mysql:latest
     environment:
@@ -88,6 +47,7 @@ services:
       DB_USER: ${DB_USER}
       DB_PORT: 3306
       SUPERTOKENS_CORE_URI: http://@supertokens:3567
+      RABBITMQ_URI: amqp://${RABBITMQ_USERNAME}:${RABBITMQ_PASSWORD}@rabbitmq:5672
       REDIS_URL: redis://redis:6379
       SERVER_PORT: 5000
       DOMAIN_API: https://server.gamenode.app
@@ -120,5 +80,4 @@ volumes:
   redis:
   db:
   gamenode:
-  supertokens-db:
 

docker-compose.yml

@@ -64,6 +64,21 @@ services:
     networks:
       - game_node_app
 
+  rabbitmq:
+    image: rabbitmq:3.13-management
+    ports:
+      # These will be available at your localhost:
+      - "5672:5672"
+      - "25672:25672" # (erlang) communication between the nodes and CLI tool
+      - "15672:15672" # communication with the web management API
+
+    volumes:
+      - rabbitmq:/var/lib/rabbitmq/
+
+    environment:
+      RABBITMQ_DEFAULT_USER: gamenode
+      RABBITMQ_DEFAULT_PASS: gamenode
+
 
 networks:
   game_node_app:
@@ -73,4 +88,5 @@ volumes:
   redis:
   db:
   manticore:
+  rabbitmq:
 

package.json

@@ -11,7 +11,7 @@
     "start": "nest start",
     "start:dev": "nest start --watch",
     "start:debug": "nest start --debug --watch",
-    "start:prod": "node --max-old-space-size=4096 dist/src/main",
+    "start:prod": "node --max-old-space-size=2048 dist/src/main",
     "start:dev:docker": "docker compose up --build -d && yarn start:dev",
     "typeorm": "typeorm-ts-node-commonjs",
     "migration:generate": "yarn run typeorm migration:generate ./src/migrations/all -d ./data-source.ts",
@@ -25,9 +25,10 @@
     "test:e2e": "jest --config ./test/jest-e2e.json"
   },
   "dependencies": {
+    "@golevelup/nestjs-rabbitmq": "^5.1.0",
     "@liaoliaots/nestjs-redis-health": "^9.0.4",
     "@nestjs/axios": "^3.0.0",
-    "@nestjs/bull": "^10.0.1",
+    "@nestjs/bullmq": "^10.1.0",
     "@nestjs/cache-manager": "^2.1.0",
     "@nestjs/common": "^10.2.7",
     "@nestjs/config": "^3.1.1",
@@ -45,7 +46,7 @@
     "async-retry": "^1.3.3",
     "axios": "^1.4.0",
     "bad-words": "^3.0.4",
-    "bull": "^4.10.4",
+    "bullmq": "^5.4.2",
     "cache-manager": "^5.2.2",
     "cache-manager-redis-yet": "^4.1.1",
     "class-transformer": "^0.5.1",

src/achievements/achievements-queue/achievements-queue.processor.ts

@@ -1,21 +1,26 @@
-import { Process, Processor } from "@nestjs/bull";
+import { Processor, WorkerHost } from "@nestjs/bullmq";
 import {
     ACHIEVEMENTS_QUEUE_NAME,
     ACHIEVEMENTS_QUEUE_TRACKING_JOB_NAME,
     AchievementsQueueJob,
 } from "./achievements-queue.constants";
-import { Job } from "bull";
+import { Job } from "bullmq";
 import { AchievementsService } from "../achievements.service";
+import { WorkerHostProcessor } from "../../utils/WorkerHostProcessor";
 
 @Processor(ACHIEVEMENTS_QUEUE_NAME)
-export class AchievementsQueueProcessor {
-    constructor(private readonly achievementsService: AchievementsService) {}
+export class AchievementsQueueProcessor extends WorkerHostProcessor {
+    constructor(private readonly achievementsService: AchievementsService) {
+        super();
+    }
 
-    @Process(ACHIEVEMENTS_QUEUE_TRACKING_JOB_NAME)
-    async processTrackingJobs(job: Job<AchievementsQueueJob>) {
-        return this.achievementsService.trackAchievementsProgress(
-            job.data.targetUserId,
-            job.data.category,
-        );
+    async process(job: Job<AchievementsQueueJob>): Promise<void> {
+        switch (job.name) {
+            case ACHIEVEMENTS_QUEUE_TRACKING_JOB_NAME:
+                return this.achievementsService.trackAchievementsProgress(
+                    job.data.targetUserId,
+                    job.data.category,
+                );
+        }
     }
 }

src/achievements/achievements-queue/achievements-queue.service.spec.ts

@@ -1,14 +1,13 @@
 import { Test, TestingModule } from "@nestjs/testing";
 import { AchievementsQueueService } from "./achievements-queue.service";
 import { ACHIEVEMENTS_QUEUE_NAME } from "./achievements-queue.constants";
-import { BullModule, getQueueToken } from "@nestjs/bull";
+import { BullModule, getQueueToken } from "@nestjs/bullmq";
 
 describe("AchievementsQueueService", () => {
     let service: AchievementsQueueService;
 
     beforeEach(async () => {
         const module: TestingModule = await Test.createTestingModule({
-            imports: [BullModule.forRoot({})],
             providers: [
                 AchievementsQueueService,
                 {

src/achievements/achievements-queue/achievements-queue.service.ts

@@ -1,11 +1,11 @@
 import { Injectable } from "@nestjs/common";
-import { InjectQueue } from "@nestjs/bull";
+import { InjectQueue } from "@nestjs/bullmq";
 import {
     ACHIEVEMENTS_QUEUE_NAME,
     ACHIEVEMENTS_QUEUE_TRACKING_JOB_NAME,
     AchievementsQueueJob,
 } from "./achievements-queue.constants";
-import { Queue } from "bull";
+import { Queue } from "bullmq";
 
 @Injectable()
 export class AchievementsQueueService {

src/achievements/achievements.module.ts

@@ -5,7 +5,7 @@ import { AchievementsQueueService } from "./achievements-queue/achievements-queu
 import { TypeOrmModule } from "@nestjs/typeorm";
 import { ObtainedAchievement } from "./entities/obtained-achievement.entity";
 import { AchievementsQueueProcessor } from "./achievements-queue/achievements-queue.processor";
-import { BullModule } from "@nestjs/bull";
+import { BullModule } from "@nestjs/bullmq";
 import { ACHIEVEMENTS_QUEUE_NAME } from "./achievements-queue/achievements-queue.constants";
 import { LevelModule } from "../level/level.module";
 

src/activities/activities-queue/activities-queue-processor.ts

@@ -1,36 +1,41 @@
-import { Process, Processor } from "@nestjs/bull";
+import { Processor, WorkerHost } from "@nestjs/bullmq";
 import { Activity } from "../activities-repository/entities/activity.entity";
-import { Job } from "bull";
+import { Job } from "bullmq";
 import { Injectable, Logger } from "@nestjs/common";
 import { ActivitiesRepositoryService } from "../activities-repository/activities-repository.service";
+import { WorkerHostProcessor } from "../../utils/WorkerHostProcessor";
 
 @Processor("activities")
 @Injectable()
-export class ActivitiesQueueProcessor {
-    private readonly logger = new Logger(ActivitiesQueueProcessor.name);
+export class ActivitiesQueueProcessor extends WorkerHostProcessor {
+    logger = new Logger(ActivitiesQueueProcessor.name);
 
     constructor(
         private activitiesRepositoryService: ActivitiesRepositoryService,
-    ) {}
-
-    @Process("addActivity")
-    async addActivity(job: Job<Activity>) {
-        try {
-            await this.activitiesRepositoryService.create(job.data);
-        } catch (e) {
-            this.logger.error("Error while processing activity: ", e);
-            this.logger.log("This error happened for data: ", job.data);
-        }
+    ) {
+        super();
     }
 
-    @Process("deleteActivity")
-    async removeActivity(job: Job<string>) {
-        try {
-            const sourceId = job.data;
-            await this.activitiesRepositoryService.deleteBySourceId(sourceId);
-        } catch (e) {
-            this.logger.error("Error while deleting activity: ", e);
-            this.logger.log("This error happened for data: ", job.data);
+    async process(job: Job<Activity | string>) {
+        if (job.name === "addActivity") {
+            try {
+                await this.activitiesRepositoryService.create(
+                    job.data as Activity,
+                );
+            } catch (e) {
+                this.logger.error("Error while processing activity: ", e);
+                this.logger.log("This error happened for data: ", job.data);
+            }
+        } else if (job.name === "deleteActivity") {
+            try {
+                const sourceId = job.data;
+                await this.activitiesRepositoryService.deleteBySourceId(
+                    sourceId as string,
+                );
+            } catch (e) {
+                this.logger.error("Error while deleting activity: ", e);
+                this.logger.log("This error happened for data: ", job.data);
+            }
         }
     }
 }

src/activities/activities-queue/activities-queue.module.ts

@@ -1,4 +1,4 @@
-import { BullModule } from "@nestjs/bull";
+import { BullModule } from "@nestjs/bullmq";
 import { Module } from "@nestjs/common";
 import { ProfileModule } from "../../profile/profile.module";
 import { ActivitiesQueueProcessor } from "./activities-queue-processor";

src/activities/activities-queue/activities-queue.service.spec.ts

@@ -1,6 +1,6 @@
 import { Test, TestingModule } from "@nestjs/testing";
 import { ActivitiesQueueService } from "./activities-queue.service";
-import { BullModule, getQueueToken } from "@nestjs/bull";
+import { getQueueToken } from "@nestjs/bullmq";
 
 describe("ActivitiesFeedService", () => {
     let service: ActivitiesQueueService;

src/activities/activities-queue/activities-queue.service.ts

@@ -1,11 +1,12 @@
-import { InjectQueue } from "@nestjs/bull";
+import { InjectQueue } from "@nestjs/bullmq";
 import { Injectable, Logger } from "@nestjs/common";
-import { Queue } from "bull";
+import { Queue } from "bullmq";
 import { ActivityCreate } from "./activities-queue.constants";
 
 @Injectable()
 export class ActivitiesQueueService {
     private readonly logger = new Logger(ActivitiesQueueService.name);
+
     constructor(
         @InjectQueue("activities") private readonly activitiesQueue: Queue,
     ) {}

src/app.module.ts

@@ -5,7 +5,7 @@ import { TypeOrmModule } from "@nestjs/typeorm";
 import { CacheModule } from "@nestjs/cache-manager";
 import { ScheduleModule } from "@nestjs/schedule";
 import { redisStore } from "cache-manager-redis-yet";
-import { BullModule } from "@nestjs/bull";
+import { BullModule } from "@nestjs/bullmq";
 import { LoggerMiddleware } from "./app.logger.middlewhare";
 import { GlobalModule } from "./global/global.module";
 import { CollectionsModule } from "./collections/collections.module";
@@ -72,7 +72,7 @@ import { NotificationsModule } from "./notifications/notifications.module";
                 const redisPort = new URL(redisUrl!).port;
 
                 return {
-                    redis: {
+                    connection: {
                         host: redisHost,
                         port: parseInt(redisPort as string) as any,
                         autoResubscribe: true,
@@ -84,8 +84,8 @@ import { NotificationsModule } from "./notifications/notifications.module";
                     defaultJobOptions: {
                         removeOnComplete: true,
                         removeOnFail: true,
-                        attempts: 10,
-                        backoff: 5000,
+                        attempts: 3,
+                        backoff: 10000,
                     },
                 };
             },

src/auth/auth.guard.ts

@@ -1,4 +1,9 @@
-import { CanActivate, ExecutionContext, Injectable } from "@nestjs/common";
+import {
+    CanActivate,
+    ExecutionContext,
+    Injectable,
+    Logger,
+} from "@nestjs/common";
 import { Error as STError } from "supertokens-node";
 
 import { verifySession } from "supertokens-node/recipe/session/framework/express";
@@ -15,9 +20,20 @@ import UserRoles from "supertokens-node/recipe/userroles";
  */
 @Injectable()
 export class AuthGuard implements CanActivate {
+    private readonly logger = new Logger(AuthGuard.name);
     constructor(private readonly reflector: Reflector) {}
 
     async canActivate(context: ExecutionContext): Promise<boolean> {
+        const ctxType = context.getType<"http" | "rmq">();
+
+        if (ctxType === "rmq") {
+            this.logger.warn(
+                `Warning: AuthGuard can't be used in a RabbitMQ service/handler`,
+            );
+
+            return true;
+        }
+
         const ctx = context.switchToHttp();
 
         let err = undefined;

src/auth/auth.service.ts

@@ -4,7 +4,6 @@ import Session from "supertokens-node/recipe/session";
 import Dashboard from "supertokens-node/recipe/dashboard";
 import ThirdPartyPasswordless from "supertokens-node/recipe/thirdpartypasswordless";
 import UserRoles from "supertokens-node/recipe/userroles";
-import jwt from "supertokens-node/recipe/jwt";
 import {
     SupertokensConfigInjectionToken,
     AuthModuleConfig,
@@ -31,7 +30,6 @@ export class AuthService {
                 apiKey: this.config.apiKey,
             },
             recipeList: [
-                jwt.init(),
                 ThirdPartyPasswordless.init({
                     flowType: "USER_INPUT_CODE",
                     contactMethod: "EMAIL",