feat: add queue for online users

einsbym · Jun 23, 2024 · 3bc4c1d · 3bc4c1d
1 parent cf54872
commit 3bc4c1d
Show file tree

Hide file tree

Showing 6 changed files with 69 additions and 25 deletions.
diff --git a/src/auth/auth.service.ts b/src/auth/auth.service.ts
@@ -21,6 +21,8 @@ export class AuthService {
         const { password, ...userWithoutPassword } = user;
         const payload = { sub: user.id };
 
+        await this.usersService.createOnlineUserJob(userWithoutPassword.username);
+
         return {
             accessToken: await this.jwtService.signAsync(payload),
             user: userWithoutPassword,

diff --git a/src/consumers/user-activity.consumer.ts b/src/consumers/user-activity.consumer.ts
@@ -0,0 +1,22 @@
+import { Process, Processor } from '@nestjs/bull';
+import { Injectable } from '@nestjs/common';
+import { InjectRepository } from '@nestjs/typeorm';
+import { Job } from 'bull';
+import { UserActivity } from 'src/entities/user-activity.entity';
+import { CreateUserActivityInput } from 'src/models/dtos/create-user-activity.input';
+import { Repository } from 'typeorm';
+
+@Injectable()
+@Processor('user-activity')
+export class UserActivityConsumer {
+    constructor(
+        @InjectRepository(UserActivity)
+        private readonly userActivityRepository: Repository<UserActivity>,
+    ) {}
+
+    @Process('user-activity')
+    async processUserActivityJob(job: Job<CreateUserActivityInput>) {
+        const { data } = job;
+        await this.userActivityRepository.save(data);
+    }
+}
diff --git a/src/modules/user.module.ts b/src/modules/user.module.ts
@@ -1,23 +1,27 @@
+import { BullModule } from '@nestjs/bull';
 import { Module } from '@nestjs/common';
 import { TypeOrmModule } from '@nestjs/typeorm';
+import { UserActivityConsumer } from 'src/consumers/user-activity.consumer';
+import { UserController } from 'src/controllers/user.controller';
+import { UserActivity } from 'src/entities/user-activity.entity';
 import { User } from 'src/entities/user.entity';
 import { UserStatsView } from 'src/entities/views/user-stats.view';
+import { StorageClientService } from 'src/providers/storage-client.service';
 import { UserService } from '../providers/user.service';
 import { UserResolver } from '../resolvers/user.resolver';
-import { UserController } from 'src/controllers/user.controller';
-import { StorageClientService } from 'src/providers/storage-client.service';
-import { BullModule } from '@nestjs/bull';
-import { UserActivity } from 'src/entities/user-activity.entity';
 
 @Module({
     imports: [
         TypeOrmModule.forFeature([User, UserStatsView, UserActivity]),
         BullModule.registerQueue({
             name: 'user-activity',
         }),
+        BullModule.registerQueue({
+            name: 'online-users',
+        }),
     ],
     controllers: [UserController],
-    providers: [UserResolver, UserService, StorageClientService],
+    providers: [UserResolver, UserService, StorageClientService, UserActivityConsumer],
     exports: [UserService, StorageClientService],
 })
 export class UserModule {}
diff --git a/src/providers/post.service.ts b/src/providers/post.service.ts
@@ -48,7 +48,7 @@ export class PostService {
         post.user = user;
         post.files = savedFiles;
 
-        await this.userService.createJob({
+        await this.userService.createUserActivityJob({
             user: user,
             description: `${user.firstName} created a post.`,
         });
@@ -103,7 +103,7 @@ export class PostService {
 
             const likedPostId = (await this.postRepository.save(post)).id;
 
-            await this.userService.createJob({
+            await this.userService.createUserActivityJob({
                 user: user,
                 description: `${user.firstName} liked a post.`,
             });
@@ -122,7 +122,7 @@ export class PostService {
 
         const unlikedPostId = (await this.postRepository.save(post)).id;
 
-        await this.userService.createJob({
+        await this.userService.createUserActivityJob({
             user: user,
             description: `${user.firstName} unliked a post.`,
         });
@@ -171,7 +171,7 @@ export class PostService {
 
         const updatedPost = await this.postRepository.save(post);
 
-        await this.userService.createJob({
+        await this.userService.createUserActivityJob({
             user: user,
             description: `${user.firstName} updated their post.`,
         });
@@ -198,7 +198,7 @@ export class PostService {
         }
 
         await this.postRepository.remove(post);
-        await this.userService.createJob({
+        await this.userService.createUserActivityJob({
             user: user,
             description: `${user.firstName} removed their post.`,
         });

diff --git a/src/providers/user.service.ts b/src/providers/user.service.ts
@@ -1,4 +1,4 @@
-import { InjectQueue, Process, Processor } from '@nestjs/bull';
+import { InjectQueue } from '@nestjs/bull';
 import {
     BadRequestException,
     ConflictException,
@@ -8,7 +8,7 @@ import {
 } from '@nestjs/common';
 import { InjectEntityManager, InjectRepository } from '@nestjs/typeorm';
 import * as bcrypt from 'bcrypt';
-import { Job, Queue } from 'bull';
+import { Queue } from 'bull';
 import { UserActivity } from 'src/entities/user-activity.entity';
 import { User } from 'src/entities/user.entity';
 import { UserStatsView } from 'src/entities/views/user-stats.view';
@@ -21,7 +21,6 @@ import { EntityManager, Repository } from 'typeorm';
 import { StorageClientService } from './storage-client.service';
 
 @Injectable()
-@Processor('user-activity')
 export class UserService {
     constructor(
         @InjectRepository(User)
@@ -34,19 +33,31 @@ export class UserService {
         private entityManager: EntityManager,
 
         @InjectQueue('user-activity')
-        private readonly queue: Queue,
+        private readonly userActivityQueue: Queue,
+
+        @InjectQueue('online-users')
+        private readonly onlineUsersQueue: Queue,
 
         private storageClientService: StorageClientService,
     ) {}
 
-    @Process('user-activity')
-    async processJob(job: Job<CreateUserActivityInput>) {
-        const { data } = job;
-        await this.userActivityRepository.save(data);
+    async createOnlineUserJob(username: string) {
+        await this.onlineUsersQueue.add('online-users', username, {
+            attempts: 3,
+            backoff: 5000,
+            removeOnComplete: true,
+        });
+    }
+
+    async isCurrentlyOnline(username: string) {
+        const jobs = await this.onlineUsersQueue.getWaiting();
+        const isOnline = !!jobs.find((job) => job.data === username);
+
+        return isOnline;
     }
 
-    async createJob(createUserActivityInput: CreateUserActivityInput) {
-        await this.queue.add('user-activity', createUserActivityInput, {
+    async createUserActivityJob(createUserActivityInput: CreateUserActivityInput) {
+        await this.userActivityQueue.add('user-activity', createUserActivityInput, {
             attempts: 3,
             backoff: 5000,
             removeOnComplete: true,
@@ -92,7 +103,7 @@ export class UserService {
 
             const { password, ...userWithoutPassword } = user;
 
-            await this.createJob({
+            await this.createUserActivityJob({
                 user: user,
                 description: `${user.firstName} changed their profile picture.`,
             });
@@ -121,7 +132,7 @@ export class UserService {
 
             const { password, ...userWithoutPassword } = user;
 
-            await this.createJob({
+            await this.createUserActivityJob({
                 user: user,
                 description: `${user.firstName} changed their cover image.`,
             });
@@ -139,7 +150,7 @@ export class UserService {
             bio: updateBioInput.bio,
         });
 
-        await this.createJob({
+        await this.createUserActivityJob({
             user: user,
             description: `${user.firstName} changed their bio.`,
         });
@@ -154,7 +165,7 @@ export class UserService {
             isPrivate: isPrivate,
         });
 
-        await this.createJob({
+        await this.createUserActivityJob({
             user: user,
             description: `${user.firstName} changed their visibility settings.`,
         });
@@ -177,7 +188,7 @@ export class UserService {
             });
         }
 
-        await this.createJob({
+        await this.createUserActivityJob({
             user: user,
             description: `${user.firstName}'s role was updated.`,
         });

diff --git a/src/resolvers/user.resolver.ts b/src/resolvers/user.resolver.ts
@@ -106,4 +106,9 @@ export class UserResolver {
     ) {
         return this.userService.unfollow(context.req, userToUnfollowId);
     }
+
+    @Query(() => Boolean)
+    isCurrentlyOnline(@Args('username') username: string) {
+        return this.userService.isCurrentlyOnline(username);
+    }
 }