Merge pull request #20 from p-society/bull

Author: zakhaev26

docker/docker-compose.dev.yml

@@ -1,34 +1,6 @@
 version: '2'
 
 services:
-  zookeeper:
-    image: zookeeper:latest
-    ports:
-      - "2181:2181"
-
-  kafka:
-    image: confluentinc/cp-kafka:latest
-    ports:
-      - "9092:9092"
-    environment:
-      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://$PRIVATE_IP:9092  # Use the dynamic private IP here
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-    depends_on:
-      - zookeeper
-
-  kafka-ui:
-    container_name: kafka-ui
-    image: provectuslabs/kafka-ui:latest
-    ports:
-      - 8080:8080
-    depends_on:
-      - kafka
-    environment:
-      DYNAMIC_CONFIG_ENABLED: 'true'
-      KAFKA_CLUSTERS_0_NAME: broadcast_cluster_1
-      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
-
   redis:
     image: redis/redis-stack-server:latest
     ports:

package.json

@@ -28,6 +28,7 @@
     "@faker-js/faker": "^8.3.1",
     "@kafkajs/confluent-schema-registry": "^2.0.1",
     "@nestjs-modules/mailer": "^2.0.2",
+    "@nestjs/bullmq": "^10.2.3",
     "@nestjs/common": "^10.4.3",
     "@nestjs/config": "^3.3.0",
     "@nestjs/core": "^10.4.3",
@@ -43,6 +44,7 @@
     "@socket.io/redis-adapter": "^8.3.0",
     "@types/nodemailer": "^6.4.17",
     "bcrypt": "^5.1.1",
+    "bullmq": "^5.34.3",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.14.1",
     "kafkajs": "^2.2.4",

src/app.module.ts

@@ -10,13 +10,10 @@ import { EventEmitterModule } from '@nestjs/event-emitter';
 import { RedisModule } from './services/redis/redis.module';
 import { UsersModule } from './services/apis/users/users.module';
 import { ConfigModule, ConfigService } from '@nestjs/config';
-import { SendOtpModule } from './services/apis/sendOtp/sendOtp.module';
 import { ReactionsModule } from './services/apis/reactions/reactions.module';
-import { KafkaModulev2 } from './services/kafka/kafka.module';
 import { ProfilesModule } from './services/apis/profiles/profiles.module';
-import { TOPIC_NAME } from './services/kafka/consumer/consumer.service';
-import fs from 'fs';
-import os from 'os';
+import { QueueModule } from './services/bullmq/queue.module';
+import { BullModule } from '@nestjs/bullmq';
 
 @Module({
   imports: [
@@ -25,6 +22,14 @@ import os from 'os';
       envFilePath: `.env`,
     }),
     RedisModule,
+    BullModule.forRoot({
+      connection: {
+        host: process.env.REDIS_HOST,
+        port: parseInt(process.env.REDIS_PORT),
+        password: process.env.REDIS_PASSWORD,
+      },
+    }),
+    QueueModule,
     EventEmitterModule.forRoot(),
     MongooseModule.forRootAsync({
       imports: [ConfigModule],
@@ -35,32 +40,9 @@ import os from 'os';
     }),
     AuthModule,
     UsersModule,
-    SendOtpModule,
     AdapterModule,
     ProfilesModule,
     ReactionsModule,
-    KafkaModulev2.register([
-      {
-        name: 'KAFKA_SERVICE_1',
-        options: {
-          client: {
-            clientId: 'gc-broadcast-server',
-            brokers: ['localhost:9092'],
-            retry: {
-              retries: 2,
-              initialRetryTime: 30,
-            },
-          },
-          consumer: {
-            groupId: 'notification',
-            allowAutoTopicCreation: true,
-          },
-          seek: {
-            [TOPIC_NAME]: new Date('2020-05-21'),
-          },
-        },
-      },
-    ]),
   ],
   controllers: [AppController],
   providers: [

src/services/apis/reactions/dto/reactions.dto.ts

@@ -1,5 +1,3 @@
-import { IsNotEmpty, IsString } from 'class-validator';
-
 /**
  * DTO for incoming reaction payload.
  * This ensures the payload structure and validates fields.
@@ -9,15 +7,11 @@ export class ReactionDto {
    * The emoji representing the reaction.
    * @example "👍"
    */
-  @IsNotEmpty({ message: 'Emoji must not be empty' })
-  @IsString({ message: 'Emoji must be a string' })
   emoji: string;
 
   /**
    * The sport associated with the reaction.
    * @example "football"
    */
-  @IsNotEmpty({ message: 'Sport must not be empty' })
-  @IsString({ message: 'Sport must be a string' })
   sport: string;
 }

src/services/apis/reactions/reactions.controller.ts

@@ -0,0 +1,13 @@
+import { Controller, Post } from '@nestjs/common';
+import { ModifyBody, setCreatedBy } from 'src/decorators/ModifyBody.decorator';
+import { ReactionService } from './reactions.service';
+import { ReactionDto } from './dto/reactions.dto';
+@Controller('reactions')
+export class ReactionController {
+  constructor(private readonly reactionService: ReactionService) {}
+
+  @Post()
+  async create(@ModifyBody(setCreatedBy()) createReactionsDto: ReactionDto) {
+    return await this.reactionService.enqueueReactions(createReactionsDto);
+  }
+}

src/services/apis/reactions/reactions.module.ts

@@ -1,8 +1,13 @@
 import { Module } from '@nestjs/common';
 import { ReactionGateway } from '../../gateways/reactions/reactions.gateway';
 import { ReactionService } from './reactions.service';
+import { PresenceModule } from 'src/services/gateways/presence/presence.module';
+import { ReactionController } from './reactions.controller';
 
 @Module({
+  imports: [PresenceModule],
   providers: [ReactionGateway, ReactionService],
+  exports: [ReactionService],
+  controllers: [ReactionController],
 })
 export class ReactionsModule {}

src/services/apis/reactions/reactions.service.ts

@@ -1,14 +1,21 @@
 import { Injectable } from '@nestjs/common';
+import { ReactionDto } from './dto/reactions.dto';
+import { processReaction } from './reaction.helper';
+import { ReactionStreamProducer } from 'src/services/bullmq/producers/reaction-stream.producer';
 
 @Injectable()
 export class ReactionService {
-  /**
-   * Save the reaction or perform other business logic.
-   * @param reaction - Processed reaction data (already validated and transformed).
-   */
-  saveReaction(reaction: { emoji: string; sport: string }): string {
-    // Simulate saving to a database or performing other business logic
-    console.log('Saving reaction to database:', reaction);
-    return 'Reaction saved successfully!';
+  constructor(
+    private readonly reactionStreamProducer: ReactionStreamProducer,
+  ) {}
+  async enqueueReactions(createReactionsDto: ReactionDto) {
+    const reaction = processReaction(createReactionsDto);
+    await this.reactionStreamProducer.pushForAsyncStream(
+      `process-reaction`,
+      reaction,
+      {
+        removeOnComplete: true,
+      },
+    );
   }
 }