refactor: redis 트랜젝션 방식 변경

Author: mssak

apps/backend/src/redis/redis.service.ts

@@ -18,9 +18,9 @@ export type RedisNode = {
 };
 
 export type RedisEdge = {
-  fromNode: number;
-  toNode: number;
-  type: 'add' | 'delete';
+  fromNode?: number;
+  toNode?: number;
+  type?: 'add' | 'delete';
 };
 
 @Injectable()

apps/backend/src/tasks/tasks.service.ts

@@ -11,12 +11,16 @@ import { InjectDataSource } from '@nestjs/typeorm';
 import { Page } from '../page/page.entity';
 import { Node } from '../node/node.entity';
 import { Edge } from '../edge/edge.entity';
+import { Inject } from '@nestjs/common';
+import Redis from 'ioredis';
+
+const REDIS_CLIENT_TOKEN = 'REDIS_CLIENT';
 
 @Injectable()
 export class TasksService {
   private readonly logger = new Logger(TasksService.name);
   constructor(
-    private readonly redisService: RedisService,
+    @Inject(REDIS_CLIENT_TOKEN) private readonly redisClient: Redis,
     @InjectDataSource() private readonly dataSource: DataSource,
   ) {}
 
@@ -25,9 +29,13 @@ export class TasksService {
     this.logger.log('스케줄러 시작');
     // 시작 시간
     const startTime = performance.now();
-    const pageKeys = await this.redisService.getAllKeys('page:*');
-    const nodeKeys = await this.redisService.getAllKeys('node:*');
-    const edgeKeys = await this.redisService.getAllKeys('edge:*');
+
+    const pageKeys = await this.redisClient.keys('page:*');
+    const nodeKeys = await this.redisClient.keys('node:*');
+    const edgeKeys = await this.redisClient.keys('edge:*');
+    // const pageKeys = await this.redisService.getAllKeys('page:*');
+    // const nodeKeys = await this.redisService.getAllKeys('node:*');
+    // const edgeKeys = await this.redisService.getAllKeys('edge:*');
 
     Promise.allSettled([
       ...pageKeys.map(this.migratePage.bind(this)),
@@ -51,7 +59,11 @@ export class TasksService {
   }
 
   async migratePage(key: string) {
-    const redisData = (await this.redisService.get(key)) as RedisPage;
+    const data = await this.redisClient.hgetall(key);
+    const redisData = Object.fromEntries(
+      Object.entries(data).map(([field, value]) => [field, value]),
+    ) as RedisPage;
+    // const redisData = (await this.redisClient.hgetall(key)) as RedisPage;
     // 데이터 없으면 오류
     if (!redisData) {
       throw new Error(`redis에 ${key}에 해당하는 데이터가 없습니다.`);
@@ -71,6 +83,7 @@ export class TasksService {
 
     // 트랜잭션 시작
     const queryRunner = this.dataSource.createQueryRunner();
+    const redisRunner = this.redisClient.multi();
     try {
       await queryRunner.startTransaction();
 
@@ -81,20 +94,23 @@ export class TasksService {
       await pageRepository.update(pageId, updateData);
 
       // redis에서 데이터 삭제
-      await this.redisService.delete(key);
+      redisRunner.del(key);
+      // await this.redisService.delete(key);
 
       // 트랜잭션 커밋
       await queryRunner.commitTransaction();
+      await redisRunner.exec();
     } catch (err) {
       // 실패하면 postgres는 roll back하고 redis의 값을 살린다.
       this.logger.error(err.stack);
       await queryRunner.rollbackTransaction();
-      updateData.title &&
-        (await this.redisService.setField(key, 'title', updateData.title));
-      updateData.content &&
-        (await this.redisService.setField(key, 'content', JSON.parse(content)));
-      updateData.emoji &&
-        (await this.redisService.setField(key, 'emoji', updateData.emoji));
+      redisRunner.discard();
+      // updateData.title &&
+      //   (await this.redisService.setField(key, 'title', updateData.title));
+      // updateData.content &&
+      //   (await this.redisService.setField(key, 'content', JSON.parse(content)));
+      // updateData.emoji &&
+      //   (await this.redisService.setField(key, 'emoji', updateData.emoji));
 
       // Promise.all에서 실패를 인식하기 위해 에러를 던진다.
       throw err;
@@ -105,9 +121,10 @@ export class TasksService {
   }
 
   async migrateNode(key: string) {
-    const redisData = (await this.redisService.get(
-      key,
-    )) as unknown as RedisNode;
+    const data = await this.redisClient.hgetall(key);
+    const redisData = Object.fromEntries(
+      Object.entries(data).map(([field, value]) => [field, value]),
+    ) as RedisNode;
     // 데이터 없으면 오류
     if (!redisData) {
       throw new Error(`redis에 ${key}에 해당하는 데이터가 없습니다.`);
@@ -126,6 +143,7 @@ export class TasksService {
 
     // 트랜잭션 시작
     const queryRunner = this.dataSource.createQueryRunner();
+    const redisRunner = this.redisClient.multi();
     try {
       await queryRunner.startTransaction();
 
@@ -136,20 +154,16 @@ export class TasksService {
       await nodeRepository.update(nodeId, updateData);
 
       // redis에서 데이터 삭제
-      await this.redisService.delete(key);
+      redisRunner.del(key);
 
       // 트랜잭션 커밋
       await queryRunner.commitTransaction();
+      await redisRunner.exec();
     } catch (err) {
       // 실패하면 postgres는 roll back하고 redis의 값을 살린다.
       this.logger.error(err.stack);
       await queryRunner.rollbackTransaction();
-      updateData.x &&
-        (await this.redisService.setField(key, 'x', updateData.x.toString()));
-      updateData.y &&
-        (await this.redisService.setField(key, 'y', updateData.y.toString()));
-      updateData.color &&
-        (await this.redisService.setField(key, 'color', updateData.color));
+      redisRunner.discard();
 
       // Promise.all에서 실패를 인식하기 위해 에러를 던진다.
       throw err;
@@ -160,16 +174,20 @@ export class TasksService {
   }
 
   async migrateEdge(key: string) {
-    const redisData = (await this.redisService.get(
-      key,
-    )) as unknown as RedisEdge;
+    const data = await this.redisClient.hgetall(key);
+    const redisData = Object.fromEntries(
+      Object.entries(data).map(([field, value]) => [field, value]),
+    ) as RedisEdge;
+
     // 데이터 없으면 오류
     if (!redisData) {
       throw new Error(`redis에 ${key}에 해당하는 데이터가 없습니다.`);
     }
 
     // 트랜잭션 시작
     const queryRunner = this.dataSource.createQueryRunner();
+    const redisRunner = this.redisClient.multi();
+
     try {
       await queryRunner.startTransaction();
 
@@ -203,25 +221,28 @@ export class TasksService {
       }
 
       // redis에서 데이터 삭제
-      await this.redisService.delete(key);
+      redisRunner.del(key);
+      // await this.redisService.delete(key);
 
       // 트랜잭션 커밋
       await queryRunner.commitTransaction();
     } catch (err) {
       // 실패하면 postgres는 roll back하고 redis의 값을 살린다.
       this.logger.error(err.stack);
       await queryRunner.rollbackTransaction();
-      await this.redisService.setField(
-        key,
-        'fromNode',
-        redisData.fromNode.toString(),
-      );
-      await this.redisService.setField(
-        key,
-        'toNode',
-        redisData.toNode.toString(),
-      );
-      await this.redisService.setField(key, 'type', redisData.type);
+      redisRunner.discard();
+
+      // await this.redisService.setField(
+      //   key,
+      //   'fromNode',
+      //   redisData.fromNode.toString(),
+      // );
+      // await this.redisService.setField(
+      //   key,
+      //   'toNode',
+      //   redisData.toNode.toString(),
+      // );
+      // await this.redisService.setField(key, 'type', redisData.type);
 
       // Promise.all에서 실패를 인식하기 위해 에러를 던진다.
       throw err;

apps/websocket/src/redis/redis.service.ts

@@ -17,9 +17,9 @@ export class RedisService {
     @Inject(RED_LOCK_TOKEN) private readonly redisLock: Redlock,
   ) {}
 
-  async getAllKeys(pattern) {
-    return await this.redisClient.keys(pattern);
-  }
+  // async getAllKeys(pattern) {
+  //   return await this.redisClient.keys(pattern);
+  // }
 
   createStream() {
     return this.redisClient.scanStream();