diff --git a/package-lock.json b/package-lock.json index b6d9d1b..e08ee1e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -22,6 +22,7 @@ "@nestjs/swagger": "^11.2.0", "@nestjs/typeorm": "^11.0.0", "@nestjs/websockets": "^11.1.6", + "ioredis": "^5.10.0", "jsonwebtoken": "^9.0.2", "mysql2": "^3.14.4", "reflect-metadata": "^0.2.2", @@ -1436,6 +1437,12 @@ } } }, + "node_modules/@ioredis/commands": { + "version": "1.5.1", + "resolved": "https://registry.npmjs.org/@ioredis/commands/-/commands-1.5.1.tgz", + "integrity": "sha512-JH8ZL/ywcJyR9MmJ5BNqZllXNZQqQbnVZOqpPQqE1vHiFgAw4NHbvE0FOduNU8IX9babitBT46571OnPTT0Zcw==", + "license": "MIT" + }, "node_modules/@isaacs/balanced-match": { "version": "4.0.1", "resolved": "https://registry.npmjs.org/@isaacs/balanced-match/-/balanced-match-4.0.1.tgz", @@ -4942,6 +4949,15 @@ "node": ">=0.8" } }, + "node_modules/cluster-key-slot": { + "version": "1.1.2", + "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.2.tgz", + "integrity": "sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/co": { "version": "4.6.0", "resolved": "https://registry.npmjs.org/co/-/co-4.6.0.tgz", @@ -6797,6 +6813,30 @@ "integrity": "sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==", "license": "ISC" }, + "node_modules/ioredis": { + "version": "5.10.0", + "resolved": "https://registry.npmjs.org/ioredis/-/ioredis-5.10.0.tgz", + "integrity": "sha512-HVBe9OFuqs+Z6n64q09PQvP1/R4Bm+30PAyyD4wIEqssh3v9L21QjCVk4kRLucMBcDokJTcLjsGeVRlq/nH6DA==", + "license": "MIT", + "dependencies": { + "@ioredis/commands": "1.5.1", + "cluster-key-slot": "^1.1.0", + "debug": "^4.3.4", + "denque": "^2.1.0", + "lodash.defaults": "^4.2.0", + "lodash.isarguments": "^3.1.0", + "redis-errors": "^1.2.0", + "redis-parser": "^3.0.0", + "standard-as-callback": "^2.1.0" + }, + "engines": { + "node": ">=12.22.0" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/ioredis" + } + }, "node_modules/ipaddr.js": { "version": "1.9.1", "resolved": "https://registry.npmjs.org/ipaddr.js/-/ipaddr.js-1.9.1.tgz", @@ -8085,11 +8125,23 @@ "integrity": "sha512-TwuEnCnxbc3rAvhf/LbG7tJUDzhqXyFnv3dtzLOPgCG/hODL7WFnsbwktkD7yUV0RrreP/l1PALq/YSg6VvjlA==", "license": "MIT" }, + "node_modules/lodash.defaults": { + "version": "4.2.0", + "resolved": "https://registry.npmjs.org/lodash.defaults/-/lodash.defaults-4.2.0.tgz", + "integrity": "sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ==", + "license": "MIT" + }, "node_modules/lodash.includes": { "version": "4.3.0", "resolved": "https://registry.npmjs.org/lodash.includes/-/lodash.includes-4.3.0.tgz", "integrity": "sha512-W3Bx6mdkRTGtlJISOvVD/lbqjTlPPUDTMnlXZFnVwi9NKJ6tiAk6LVdlhZMm17VZisqhKcgzpO5Wz91PCt5b0w==" }, + "node_modules/lodash.isarguments": { + "version": "3.1.0", + "resolved": "https://registry.npmjs.org/lodash.isarguments/-/lodash.isarguments-3.1.0.tgz", + "integrity": "sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg==", + "license": "MIT" + }, "node_modules/lodash.isboolean": { "version": "3.0.3", "resolved": "https://registry.npmjs.org/lodash.isboolean/-/lodash.isboolean-3.0.3.tgz", @@ -9298,6 +9350,27 @@ "url": "https://paulmillr.com/funding/" } }, + "node_modules/redis-errors": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/redis-errors/-/redis-errors-1.2.0.tgz", + "integrity": "sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==", + "license": "MIT", + "engines": { + "node": ">=4" + } + }, + "node_modules/redis-parser": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/redis-parser/-/redis-parser-3.0.0.tgz", + "integrity": "sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==", + "license": "MIT", + "dependencies": { + "redis-errors": "^1.0.0" + }, + "engines": { + "node": ">=4" + } + }, "node_modules/reflect-metadata": { "version": "0.2.2", "resolved": "https://registry.npmjs.org/reflect-metadata/-/reflect-metadata-0.2.2.tgz", @@ -9962,6 +10035,12 @@ "node": ">=8" } }, + "node_modules/standard-as-callback": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/standard-as-callback/-/standard-as-callback-2.1.0.tgz", + "integrity": "sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==", + "license": "MIT" + }, "node_modules/statuses": { "version": "2.0.2", "resolved": "https://registry.npmjs.org/statuses/-/statuses-2.0.2.tgz", diff --git a/package.json b/package.json index 5c727a5..742d956 100644 --- a/package.json +++ b/package.json @@ -33,6 +33,7 @@ "@nestjs/swagger": "^11.2.0", "@nestjs/typeorm": "^11.0.0", "@nestjs/websockets": "^11.1.6", + "ioredis": "^5.10.0", "jsonwebtoken": "^9.0.2", "mysql2": "^3.14.4", "reflect-metadata": "^0.2.2", diff --git a/src/app.module.ts b/src/app.module.ts index 57b1f10..2f2a530 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -10,7 +10,8 @@ import { CardsetOrmEntity } from './cardset/infrastructure/persistence/orm/cards import { CardOrmEntity } from './cardset/infrastructure/persistence/orm/card.orm-entity'; import { CardsetManagerOrmEntity } from './cardset/infrastructure/persistence/orm/cardset-manager.orm-entity'; import { CardSetMetadataOrmEntity } from './cardset/infrastructure/persistence/orm/cardset-metadata.orm-entity'; -import { YjsDocumentOrmEntity } from './collaboration/infrastructure/persistence/orm/yjs-document.orm-entity'; +import { CardsetContentOrmEntity } from './collaboration/infrastructure/persistence/orm/cardset-content.orm-entity'; +import { CardsetIncrementalOrmEntity } from './collaboration/infrastructure/persistence/orm/cardset-incremental.orm-entity'; @Module({ imports: [ @@ -27,7 +28,8 @@ import { YjsDocumentOrmEntity } from './collaboration/infrastructure/persistence CardOrmEntity, CardsetManagerOrmEntity, CardSetMetadataOrmEntity, - YjsDocumentOrmEntity, + CardsetContentOrmEntity, + CardsetIncrementalOrmEntity, ], synchronize: true, }), diff --git a/src/auth/infrastructure/guard/ws-auth.guard.ts b/src/auth/infrastructure/guard/ws-auth.guard.ts index 656a2f3..cdeeea0 100644 --- a/src/auth/infrastructure/guard/ws-auth.guard.ts +++ b/src/auth/infrastructure/guard/ws-auth.guard.ts @@ -11,11 +11,23 @@ import { Socket } from 'socket.io'; export class WsAuthGuard implements CanActivate { private readonly logger = new Logger(WsAuthGuard.name); - constructor(private readonly authService: AuthService) {} + constructor(private readonly authService: AuthService) { } canActivate(context: ExecutionContext): boolean { const client: Socket = context.switchToWs().getClient(); + const SKIP_AUTH = process.env.SKIP_WS_AUTH === 'true'; + if (SKIP_AUTH) { + (client.data as { user: unknown }).user = { + userId: 'test-user', + email: 'test@example.com', + }; + this.logger.warn( + `⚠️ 테스트 모드: 인증을 건너뛰고 있습니다 (client ${client.id})`, + ); + return true; + } + const bearer = (client.handshake.auth?.token as string | undefined) ?? client.handshake.headers?.authorization; diff --git a/src/cardset/application/cardset.use-case.ts b/src/cardset/application/cardset.use-case.ts index 8a9bcff..d85714d 100644 --- a/src/cardset/application/cardset.use-case.ts +++ b/src/cardset/application/cardset.use-case.ts @@ -21,6 +21,7 @@ import { UserGrpcClient } from '../infrastructure/grpc/user-grpc.client'; import type { UserInfo } from '../infrastructure/grpc/user-grpc.client'; import { CreateCardsetRequest } from './dto/request/create-cardset.request'; import { UpdateCardsetRequest } from './dto/request/update-cardset.request'; +import { CollaborationUseCase } from '../../collaboration/application/collaboration.use-case'; @Injectable() export class CardsetUseCase { @@ -39,6 +40,7 @@ export class CardsetUseCase { private readonly dataSource: DataSource, @Inject(CARDSET_METADATA_REPOSITORY) private readonly metadataRepository: ICardSetMetadataRepository, + private readonly collaborationUseCase: CollaborationUseCase, ) {} private async checkIsManager( @@ -58,7 +60,6 @@ export class CardsetUseCase { async create(userId: number, dto: CreateCardsetRequest): Promise { await this.groupGrpcClient.checkUserInGroup(dto.groupId, userId); - // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment const additionalManagerIds: number[] = dto.managerIds ?? []; for (const managerId of additionalManagerIds) { await this.groupGrpcClient.checkUserInGroup(dto.groupId, managerId); @@ -110,11 +111,11 @@ export class CardsetUseCase { cardSetIds: number[], ): Promise> { if (cardSetIds.length === 0 || this.skipUserGrpc) return new Map(); - /* eslint-disable @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-call */ + const managers: CardsetManager[] = await this.cardsetManagerRepository.findByCardSetIds(cardSetIds); const userIds: number[] = [...new Set(managers.map((m) => m.userId))]; - /* eslint-enable @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-call */ + const users = await this.userGrpcClient.getUsersByIds(userIds); const userMap = new Map(users.map((u) => [Number(u.id), u])); const result = new Map(); @@ -249,11 +250,11 @@ export class CardsetUseCase { }[] > { await this.groupGrpcClient.checkUserInGroup(groupId, userId); - /* eslint-disable @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-call */ + const cardsets: Cardset[] = await this.cardsetRepository.findByGroupId(groupId); const ids: number[] = cardsets.map((c) => c.id); - /* eslint-enable @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-call */ + const [metadataMap, likedMap, bookmarkedMap, managersMap] = await Promise.all([ this.metadataRepository.findByCardSetIds(ids), @@ -308,7 +309,6 @@ export class CardsetUseCase { return this.dataSource.transaction(async (manager) => { if (dto.managerIds !== undefined) { - // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment const newManagerIds: number[] = dto.managerIds; for (const managerId of newManagerIds) { await this.groupGrpcClient.checkUserInGroup( @@ -404,4 +404,15 @@ export class CardsetUseCase { return this.cardsetRepository.update(id, updatedCardset); }); } + + async saveCards(cardSetId: number, userId: number): Promise { + await this.checkIsManager(cardSetId, userId); + await this.collaborationUseCase.saveCardsetContent(cardSetId); + } + + async findCardsFromYjs( + cardSetId: number, + ): Promise<{ id: string; question: string; answer: string }[]> { + return this.collaborationUseCase.getCards(cardSetId); + } } diff --git a/src/cardset/application/dto/response/card.response.ts b/src/cardset/application/dto/response/card.response.ts index 8a73533..9522d81 100644 --- a/src/cardset/application/dto/response/card.response.ts +++ b/src/cardset/application/dto/response/card.response.ts @@ -30,4 +30,19 @@ export class CardResponse { res.updatedAt = card.updatedAt; return res; } + + static fromYjs(card: { + id: string; + question: string; + answer: string; + }): CardResponse { + const res = new CardResponse(); + res.id = Number(card.id); + res.content = card.question; + res.order = 0; + res.cardsetId = 0; + res.createdAt = new Date(); + res.updatedAt = new Date(); + return res; + } } diff --git a/src/cardset/application/dto/response/yjs-card.response.ts b/src/cardset/application/dto/response/yjs-card.response.ts new file mode 100644 index 0000000..d2f12dc --- /dev/null +++ b/src/cardset/application/dto/response/yjs-card.response.ts @@ -0,0 +1,24 @@ +import { ApiProperty } from '@nestjs/swagger'; + +export class YjsCardResponse { + @ApiProperty({ example: '1' }) + id!: string; + + @ApiProperty({ example: '질문 내용' }) + question!: string; + + @ApiProperty({ example: '답변 내용' }) + answer!: string; + + static from(card: { + id: string; + question: string; + answer: string; + }): YjsCardResponse { + const res = new YjsCardResponse(); + res.id = card.id; + res.question = card.question; + res.answer = card.answer; + return res; + } +} diff --git a/src/cardset/cardset.module.ts b/src/cardset/cardset.module.ts index 1771f7d..d076ba1 100644 --- a/src/cardset/cardset.module.ts +++ b/src/cardset/cardset.module.ts @@ -30,6 +30,7 @@ import { ImageGrpcClient } from './infrastructure/grpc/image-grpc.client'; import { ReactionGrpcClient } from './infrastructure/grpc/reaction-grpc.client'; import { UserGrpcClient } from './infrastructure/grpc/user-grpc.client'; import { GrpcClientModule } from '../shared/grpc/grpc-client.module'; +import { CollaborationModule } from '../collaboration/collaboration.module'; @Module({ imports: [ @@ -40,8 +41,14 @@ import { GrpcClientModule } from '../shared/grpc/grpc-client.module'; CardSetMetadataOrmEntity, ]), GrpcClientModule, + CollaborationModule, + ], + controllers: [ + CardsetController, + CardController, + GroupCardsetController, + CardsetGrpcController, ], - controllers: [CardsetController, CardController, GroupCardsetController, CardsetGrpcController], providers: [ { provide: CARDSET_REPOSITORY, useClass: CardsetRepositoryImpl }, { provide: CARD_REPOSITORY, useClass: CardRepositoryImpl }, diff --git a/src/cardset/domain/repository/card.repository.ts b/src/cardset/domain/repository/card.repository.ts index 1a481b1..8ccdec5 100644 --- a/src/cardset/domain/repository/card.repository.ts +++ b/src/cardset/domain/repository/card.repository.ts @@ -7,7 +7,11 @@ export interface ICardRepository { findById(id: number): Promise; findAllByCardsetId(cardsetId: number): Promise; save(card: Card, manager?: EntityManager): Promise; - update(id: number, card: Partial): Promise; + update( + id: number, + card: Partial, + manager?: EntityManager, + ): Promise; delete(id: number, manager?: EntityManager): Promise; updateOrder( cardId: number, diff --git a/src/cardset/domain/repository/cardset-metadata.repository.ts b/src/cardset/domain/repository/cardset-metadata.repository.ts index daa1a89..d6da245 100644 --- a/src/cardset/domain/repository/cardset-metadata.repository.ts +++ b/src/cardset/domain/repository/cardset-metadata.repository.ts @@ -1,8 +1,14 @@ -export const CARDSET_METADATA_REPOSITORY = Symbol('CARDSET_METADATA_REPOSITORY'); +export const CARDSET_METADATA_REPOSITORY = Symbol( + 'CARDSET_METADATA_REPOSITORY', +); export interface ICardSetMetadataRepository { - findByCardSetId(cardSetId: number): Promise<{ likeCount: number; bookmarkCount: number } | null>; - findByCardSetIds(cardSetIds: number[]): Promise>; + findByCardSetId( + cardSetId: number, + ): Promise<{ likeCount: number; bookmarkCount: number } | null>; + findByCardSetIds( + cardSetIds: number[], + ): Promise>; upsertAndIncrementLike(cardSetId: number): Promise; upsertAndDecrementLike(cardSetId: number): Promise; upsertAndIncrementBookmark(cardSetId: number): Promise; diff --git a/src/cardset/infrastructure/grpc/reaction-grpc.client.ts b/src/cardset/infrastructure/grpc/reaction-grpc.client.ts index cfd05bb..6259131 100644 --- a/src/cardset/infrastructure/grpc/reaction-grpc.client.ts +++ b/src/cardset/infrastructure/grpc/reaction-grpc.client.ts @@ -71,7 +71,9 @@ export class ReactionGrpcClient implements OnModuleInit { userId, }), ); - return new Map(Object.entries(result.results).map(([k, v]) => [Number(k), v])); + return new Map( + Object.entries(result.results).map(([k, v]) => [Number(k), v]), + ); } async areBookmarked( @@ -85,6 +87,8 @@ export class ReactionGrpcClient implements OnModuleInit { userId, }), ); - return new Map(Object.entries(result.results).map(([k, v]) => [Number(k), v])); + return new Map( + Object.entries(result.results).map(([k, v]) => [Number(k), v]), + ); } } diff --git a/src/cardset/infrastructure/grpc/user-grpc.client.ts b/src/cardset/infrastructure/grpc/user-grpc.client.ts index 2d5c137..0562d10 100644 --- a/src/cardset/infrastructure/grpc/user-grpc.client.ts +++ b/src/cardset/infrastructure/grpc/user-grpc.client.ts @@ -10,9 +10,7 @@ export interface UserInfo { } interface UserQueryService { - getUsers(data: { - userIds: number[]; - }): Observable<{ users: UserInfo[] }>; + getUsers(data: { userIds: number[] }): Observable<{ users: UserInfo[] }>; } @Injectable() @@ -30,9 +28,7 @@ export class UserGrpcClient implements OnModuleInit { async getUsersByIds(userIds: number[]): Promise { if (userIds.length === 0) return []; - const result = await firstValueFrom( - this.userService.getUsers({ userIds }), - ); + const result = await firstValueFrom(this.userService.getUsers({ userIds })); return result.users; } } diff --git a/src/cardset/infrastructure/http/card.controller.ts b/src/cardset/infrastructure/http/card.controller.ts index 4ddbf0b..337c76c 100644 --- a/src/cardset/infrastructure/http/card.controller.ts +++ b/src/cardset/infrastructure/http/card.controller.ts @@ -5,7 +5,7 @@ import { CardUseCase } from '../../application/card.use-case'; @ApiTags('cards') @Controller('cards') export class CardController { - constructor(private readonly cardUseCase: CardUseCase) { } + constructor(private readonly cardUseCase: CardUseCase) {} // @Post() // @ApiOperation({ summary: '카드 생성' }) diff --git a/src/cardset/infrastructure/http/cardset.controller.ts b/src/cardset/infrastructure/http/cardset.controller.ts index 439dda0..9752ba7 100644 --- a/src/cardset/infrastructure/http/cardset.controller.ts +++ b/src/cardset/infrastructure/http/cardset.controller.ts @@ -8,6 +8,7 @@ import { Param, Headers, } from '@nestjs/common'; + import { ApiTags, ApiOperation, @@ -21,13 +22,14 @@ import { UpdateCardsetRequest } from '../../application/dto/request/update-cards import { CardsetCreateResponse } from '../../application/dto/response/cardset-create.response'; import { CardsetResponse } from '../../application/dto/response/cardset.response'; import { ManagerInfoResponse } from '../../application/dto/response/manager-info.response'; +import { YjsCardResponse } from '../../application/dto/response/yjs-card.response'; import { ApiResponse } from '../../../shared/common/api-response'; @ApiExtraModels(CardsetResponse, ManagerInfoResponse) @ApiTags('card-sets') @Controller('card-sets') export class CardsetController { - constructor(private readonly cardsetUseCase: CardsetUseCase) { } + constructor(private readonly cardsetUseCase: CardsetUseCase) {} @Post() @ApiOperation({ summary: '카드셋 생성' }) @@ -147,6 +149,37 @@ export class CardsetController { return ApiResponse.success(null, '삭제되었습니다.'); } + @Get(':cardsetId/cards') + @ApiOperation({ summary: '카드셋의 카드 목록 조회' }) + @ApiParam({ name: 'cardsetId', type: Number }) + @SwaggerApiResponse({ + status: 200, + description: '조회 성공', + type: [YjsCardResponse], + }) + async findCards( + @Headers('X-USER-ID') _userId: string, + @Param('cardsetId') cardsetId: string, + ): Promise> { + const cards = await this.cardsetUseCase.findCardsFromYjs( + parseInt(cardsetId), + ); + return ApiResponse.success(cards.map((c) => YjsCardResponse.from(c))); + } + + @Post(':cardsetId') + @ApiOperation({ summary: '카드 편집 저장' }) + @ApiParam({ name: 'cardsetId', type: Number }) + @SwaggerApiResponse({ status: 200, description: '저장 성공' }) + @SwaggerApiResponse({ status: 403, description: '매니저 권한 없음' }) + async saveCards( + @Headers('X-USER-ID') userId: string, + @Param('cardsetId') cardsetId: string, + ): Promise> { + await this.cardsetUseCase.saveCards(parseInt(cardsetId), parseInt(userId)); + return ApiResponse.success(null); + } + // @Put(':cardsetId/card-count') // @ApiOperation({ summary: '카드 수 업데이트' }) // @ApiParam({ name: 'cardsetId', type: Number }) diff --git a/src/cardset/infrastructure/http/group-cardset.controller.ts b/src/cardset/infrastructure/http/group-cardset.controller.ts index 82536fb..8827567 100644 --- a/src/cardset/infrastructure/http/group-cardset.controller.ts +++ b/src/cardset/infrastructure/http/group-cardset.controller.ts @@ -36,7 +36,15 @@ export class GroupCardsetController { parseInt(userId), ); const content = items.map( - ({ cardset, imageUrl, liked, bookmarked, managers, likeCount, bookmarkCount }) => + ({ + cardset, + imageUrl, + liked, + bookmarked, + managers, + likeCount, + bookmarkCount, + }) => CardsetListItemResponse.from( cardset, imageUrl, diff --git a/src/cardset/infrastructure/persistence/card.repository.impl.ts b/src/cardset/infrastructure/persistence/card.repository.impl.ts index b52d877..f9bf3da 100644 --- a/src/cardset/infrastructure/persistence/card.repository.impl.ts +++ b/src/cardset/infrastructure/persistence/card.repository.impl.ts @@ -36,8 +36,15 @@ export class CardRepositoryImpl implements ICardRepository { return CardMapper.toDomain(saved); } - async update(id: number, card: Partial): Promise { - await this.ormRepository.update(id, CardMapper.toOrm(card as Card)); + async update( + id: number, + card: Partial, + manager?: EntityManager, + ): Promise { + const repo = manager + ? manager.getRepository(CardOrmEntity) + : this.ormRepository; + await repo.update(id, CardMapper.toOrm(card as Card)); return this.findById(id); } diff --git a/src/cardset/infrastructure/persistence/cardset-metadata.repository.impl.ts b/src/cardset/infrastructure/persistence/cardset-metadata.repository.impl.ts index 237d685..1a06392 100644 --- a/src/cardset/infrastructure/persistence/cardset-metadata.repository.impl.ts +++ b/src/cardset/infrastructure/persistence/cardset-metadata.repository.impl.ts @@ -5,7 +5,9 @@ import { ICardSetMetadataRepository } from '../../domain/repository/cardset-meta import { CardSetMetadataOrmEntity } from './orm/cardset-metadata.orm-entity'; @Injectable() -export class CardSetMetadataRepositoryImpl implements ICardSetMetadataRepository { +export class CardSetMetadataRepositoryImpl + implements ICardSetMetadataRepository +{ constructor( @InjectRepository(CardSetMetadataOrmEntity) private readonly ormRepository: Repository, diff --git a/src/cardset/infrastructure/persistence/cardset.repository.impl.ts b/src/cardset/infrastructure/persistence/cardset.repository.impl.ts index 360cf85..0a8632d 100644 --- a/src/cardset/infrastructure/persistence/cardset.repository.impl.ts +++ b/src/cardset/infrastructure/persistence/cardset.repository.impl.ts @@ -25,7 +25,14 @@ export class CardsetRepositoryImpl implements ICardsetRepository { } async findAllPaged(options: CardsetPageOptions): Promise { - const { page, size, sortBy = 'createdAt', order = 'DESC', keyword, category } = options; + const { + page, + size, + sortBy = 'createdAt', + order = 'DESC', + keyword, + category, + } = options; const qb = this.ormRepository.createQueryBuilder('cs'); @@ -42,7 +49,9 @@ export class CardsetRepositoryImpl implements ICardsetRepository { cardCount: 'cs.cardCount', }; const sortField = allowedSortFields[sortBy] ?? 'cs.createdAt'; - qb.orderBy(sortField, order).skip(page * size).take(size); + qb.orderBy(sortField, order) + .skip(page * size) + .take(size); const [orms, total] = await qb.getManyAndCount(); return { items: orms.map((orm) => CardsetMapper.toDomain(orm)), total }; diff --git a/src/collaboration/application/collaboration.use-case.ts b/src/collaboration/application/collaboration.use-case.ts index f530234..c8ded29 100644 --- a/src/collaboration/application/collaboration.use-case.ts +++ b/src/collaboration/application/collaboration.use-case.ts @@ -1,57 +1,42 @@ -import { Inject, Injectable, Logger } from '@nestjs/common'; +import { Injectable, Logger, NotFoundException } from '@nestjs/common'; +import { InjectRepository } from '@nestjs/typeorm'; +import { Repository } from 'typeorm'; import * as Y from 'yjs'; -import { YJS_DOCUMENT_REPOSITORY } from '../domain/repository/yjs-document.repository'; -import type { IYjsDocumentRepository } from '../domain/repository/yjs-document.repository'; -import { YjsDocument } from '../domain/model/yjs-document'; +import { YjsDocumentService } from '../infrastructure/redis/yjs-document.service'; +import { CardsetContentOrmEntity } from '../infrastructure/persistence/orm/cardset-content.orm-entity'; @Injectable() export class CollaborationUseCase { private readonly logger = new Logger(CollaborationUseCase.name); - private documentCache = new Map(); - private persistDebounceMap = new Map(); constructor( - @Inject(YJS_DOCUMENT_REPOSITORY) - private readonly yjsDocumentRepository: IYjsDocumentRepository, + private readonly yjsDocumentService: YjsDocumentService, + @InjectRepository(CardsetContentOrmEntity) + private readonly cardsetContentRepository: Repository, ) {} async getOrCreateDocument(cardsetId: number): Promise { - if (this.documentCache.has(cardsetId)) { - return this.documentCache.get(cardsetId)!; - } + const fromRedis = await this.yjsDocumentService.loadDocument( + cardsetId.toString(), + ); + if (fromRedis) return fromRedis; - const stored = await this.yjsDocumentRepository.findByCardsetId(cardsetId); - let doc: Y.Doc; - - if (stored) { - doc = new Y.Doc(); - Y.applyUpdate(doc, stored.documentData); - this.logger.log(`Loaded Yjs document for cardset ${cardsetId} from DB`); - } else { - doc = new Y.Doc(); - await this.persistDocument(cardsetId, doc); - this.logger.log(`Created new Yjs document for cardset ${cardsetId}`); + const fromDb = await this.loadCardsetContentFromDB(cardsetId); + if (fromDb) { + await this.yjsDocumentService.saveDocument(cardsetId.toString(), fromDb); + return fromDb; } - doc.on('update', () => { - const existing = this.persistDebounceMap.get(cardsetId); - if (existing) clearTimeout(existing); - this.persistDebounceMap.set( - cardsetId, - setTimeout(() => { - void this.persistDocument(cardsetId, doc); - this.persistDebounceMap.delete(cardsetId); - }, 500), - ); - }); - - this.documentCache.set(cardsetId, doc); + const doc = new Y.Doc(); + await this.yjsDocumentService.saveDocument(cardsetId.toString(), doc); return doc; } async applyUpdate(cardsetId: number, update: number[]): Promise { const doc = await this.getOrCreateDocument(cardsetId); - Y.applyUpdate(doc, new Uint8Array(update)); + const updateArr = new Uint8Array(update); + Y.applyUpdate(doc, updateArr); + await this.yjsDocumentService.saveUpdate(cardsetId.toString(), updateArr); return Y.encodeStateAsUpdate(doc); } @@ -60,6 +45,15 @@ export class CollaborationUseCase { return Y.encodeStateAsUpdate(doc); } + async getCards( + cardsetId: number, + ): Promise<{ id: string; question: string; answer: string }[]> { + const doc = await this.getOrCreateDocument(cardsetId); + return doc + .getArray<{ id: string; question: string; answer: string }>('cards') + .toArray(); + } + async syncCardsFromDB( cardsetId: number, cards: { @@ -86,38 +80,69 @@ export class CollaborationUseCase { })); cardsArray.insert(0, yjsCards); + await this.yjsDocumentService.saveDocument(cardsetId.toString(), doc); this.logger.log( `Synced ${cards.length} cards to Yjs for cardset ${cardsetId}`, ); } - async deleteDocument(cardsetId: number): Promise { - await this.yjsDocumentRepository.deleteByCardsetId(cardsetId); - this.documentCache.delete(cardsetId); - this.logger.log(`Deleted Yjs document for cardset ${cardsetId}`); + async saveCardsetContent(cardSetId: number): Promise { + const doc = await this.yjsDocumentService.loadDocument( + cardSetId.toString(), + ); + if (!doc) { + throw new NotFoundException('Cardset snapshot not found in Redis'); + } + + const jsonContent = JSON.stringify(doc.toJSON() ?? {}); + + let content = await this.cardsetContentRepository.findOne({ + where: { cardsetId: cardSetId }, + }); + if (!content) { + content = this.cardsetContentRepository.create({ + cardsetId: cardSetId, + content: '', + }); + } + content.content = jsonContent; + await this.cardsetContentRepository.save(content); + + await this.yjsDocumentService.flushIncrementalHistory(cardSetId.toString()); + this.logger.log(`Saved cardset ${cardSetId} content to DB`); } - private async persistDocument(cardsetId: number, doc: Y.Doc): Promise { + async loadCardsetContentFromDB(cardSetId: number): Promise { try { - const state = Y.encodeStateAsUpdate(doc); - const stored = - await this.yjsDocumentRepository.findByCardsetId(cardsetId); - - if (stored) { - const updated = stored.withNewData(Buffer.from(state)); - await this.yjsDocumentRepository.update(stored.id, updated); - } else { - const newDoc = YjsDocument.create({ - cardsetId, - documentData: Buffer.from(state), - }); - await this.yjsDocumentRepository.save(newDoc); + const cardsetContent = await this.cardsetContentRepository.findOne({ + where: { cardsetId: cardSetId }, + }); + + if (!cardsetContent || !cardsetContent.content) return null; + + const jsonContent = JSON.parse(cardsetContent.content) as Record< + string, + unknown + >; + const doc = new Y.Doc(); + if (jsonContent && typeof jsonContent === 'object') { + const yMap = doc.getMap('content'); + for (const [key, value] of Object.entries(jsonContent)) { + yMap.set(key, value); + } } - } catch (error) { - this.logger.error( - `Failed to persist Yjs document for cardset ${cardsetId}:`, - error, - ); + return doc; + } catch { + return null; } } + + async deleteDocument(cardsetId: number): Promise { + await this.cardsetContentRepository.delete({ cardsetId }); + await this.yjsDocumentService.saveDocument( + cardsetId.toString(), + new Y.Doc(), + ); + this.logger.log(`Deleted Yjs document for cardset ${cardsetId}`); + } } diff --git a/src/collaboration/collaboration.module.ts b/src/collaboration/collaboration.module.ts index 718909e..1d70afb 100644 --- a/src/collaboration/collaboration.module.ts +++ b/src/collaboration/collaboration.module.ts @@ -1,19 +1,22 @@ import { Module } from '@nestjs/common'; import { TypeOrmModule } from '@nestjs/typeorm'; -import { YjsDocumentOrmEntity } from './infrastructure/persistence/orm/yjs-document.orm-entity'; -import { YjsDocumentRepositoryImpl } from './infrastructure/persistence/yjs-document.repository.impl'; -import { YJS_DOCUMENT_REPOSITORY } from './domain/repository/yjs-document.repository'; +import { CardsetContentOrmEntity } from './infrastructure/persistence/orm/cardset-content.orm-entity'; +import { CardsetIncrementalOrmEntity } from './infrastructure/persistence/orm/cardset-incremental.orm-entity'; +import { YjsDocumentService } from './infrastructure/redis/yjs-document.service'; import { CollaborationUseCase } from './application/collaboration.use-case'; import { CollaborationGateway } from './infrastructure/gateway/collaboration.gateway'; import { AuthModule } from '../auth/auth.module'; @Module({ - imports: [TypeOrmModule.forFeature([YjsDocumentOrmEntity]), AuthModule], - providers: [ - { provide: YJS_DOCUMENT_REPOSITORY, useClass: YjsDocumentRepositoryImpl }, - CollaborationUseCase, - CollaborationGateway, + imports: [ + TypeOrmModule.forFeature([ + CardsetContentOrmEntity, + CardsetIncrementalOrmEntity, + ]), + AuthModule, ], + providers: [YjsDocumentService, CollaborationUseCase, CollaborationGateway], + exports: [CollaborationUseCase], }) export class CollaborationModule {} diff --git a/src/collaboration/infrastructure/gateway/collaboration.gateway.ts b/src/collaboration/infrastructure/gateway/collaboration.gateway.ts index fffe992..7668613 100644 --- a/src/collaboration/infrastructure/gateway/collaboration.gateway.ts +++ b/src/collaboration/infrastructure/gateway/collaboration.gateway.ts @@ -9,9 +9,11 @@ import { } from '@nestjs/websockets'; import { Logger, UseGuards } from '@nestjs/common'; import { Server, Socket } from 'socket.io'; +import * as Y from 'yjs'; import { WsAuthGuard } from '../../../auth/infrastructure/guard/ws-auth.guard'; import { WsUser } from '../../../shared/decorator/ws-user.decorator'; import type { UserAuth } from '../../../shared/types/user-auth.type'; +import { YjsDocumentService } from '../redis/yjs-document.service'; import { CollaborationUseCase } from '../../application/collaboration.use-case'; @UseGuards(WsAuthGuard) @@ -27,16 +29,23 @@ export class CollaborationGateway @WebSocketServer() server!: Server; + private static readonly FLUSH_DELAY_MS = 5000; + private readonly logger = new Logger(CollaborationGateway.name); + private flushTimeouts = new Map(); - constructor(private readonly collaborationUseCase: CollaborationUseCase) {} + constructor( + private readonly yjsDocumentService: YjsDocumentService, + private readonly collaborationUseCase: CollaborationUseCase, + ) {} handleConnection(client: Socket) { this.logger.log(`Client connected: ${client.id}`); } - handleDisconnect(client: Socket) { + async handleDisconnect(client: Socket) { this.logger.log(`Client disconnected: ${client.id}`); + await this.removeClientFromAllCardsets(client); } @SubscribeMessage('join-cardset') @@ -45,24 +54,54 @@ export class CollaborationGateway @ConnectedSocket() client: Socket, @MessageBody() data: { cardsetId: string }, ) { - try { - const { cardsetId } = data; - this.logger.log(`User ${user.userId} joining cardset ${cardsetId}`); + const { cardsetId } = data; + this.logger.log(`User ${user.userId} joining cardset ${cardsetId}`); + try { void client.join(`cardset:${cardsetId}`); - const state = await this.collaborationUseCase.getState( - parseInt(cardsetId), - ); - client.emit('sync-response', { - cardsetId, - update: Array.from(state), - }); + await this.yjsDocumentService.registerClient(cardsetId, client.id); + this.clearScheduledFlush(cardsetId); + + let doc = await this.yjsDocumentService.loadDocument(cardsetId); + if (!doc) { + doc = await this.loadDocumentFromDBOrCreate(cardsetId); + } + + if (!doc) { + this.logger.warn( + `Failed to load or create document for cardset ${cardsetId}, creating empty document`, + ); + doc = new Y.Doc(); + } + + const state = Y.encodeStateAsUpdate(doc); + client.emit('sync', { cardsetId, update: Array.from(state) }); this.logger.log(`User ${user.userId} joined cardset ${cardsetId}`); } catch (error) { this.logger.error('Error joining cardset:', error); - client.emit('error', { message: 'Failed to join cardset' }); + this.logger.error('Error details:', { + cardsetId, + userId: user?.userId, + errorMessage: error instanceof Error ? error.message : String(error), + errorStack: error instanceof Error ? error.stack : undefined, + }); + + try { + const emptyDoc = new Y.Doc(); + const state = Y.encodeStateAsUpdate(emptyDoc); + client.emit('sync', { cardsetId, update: Array.from(state) }); + this.logger.warn( + `Sent empty document to client due to error for cardset ${cardsetId}`, + ); + } catch (fallbackError) { + this.logger.error('Failed to send fallback document:', fallbackError); + client.emit('error', { + message: 'Failed to join cardset', + details: error instanceof Error ? error.message : String(error), + }); + } } } @@ -73,19 +112,38 @@ export class CollaborationGateway @MessageBody() data: { cardsetId: string }, ) { try { - await client.leave(`cardset:${data.cardsetId}`); - this.logger.log(`User ${user.userId} left cardset ${data.cardsetId}`); + const { cardsetId } = data; + this.logger.log(`User ${user.userId} leaving cardset ${cardsetId}`); + + void client.leave(`cardset:${cardsetId}`); + await this.yjsDocumentService.unregisterClient(cardsetId, client.id); + const activeCount = + await this.yjsDocumentService.getActiveClientCount(cardsetId); + if (activeCount === 0) { + this.scheduleFlush(cardsetId); + } + this.logger.log(`User ${user.userId} left cardset ${cardsetId}`); } catch (error) { this.logger.error('Error leaving cardset:', error); } } - @SubscribeMessage('sync') - async handleSync( + @SubscribeMessage('awareness') + handleAwareness( + client: Socket, + payload: { cardsetId: string; awareness: number[] }, + ) { + const { cardsetId, awareness } = payload; + client.to(`cardset:${cardsetId}`).emit('awareness', { + data: { cardsetId, awareness: new Uint8Array(awareness) }, + }); + } + + @SubscribeMessage('update') + async handleUpdate( @WsUser() user: UserAuth, @ConnectedSocket() client: Socket, - @MessageBody() - data: { cardsetId: string; update?: number[] }, + @MessageBody() data: { cardsetId: string; update?: number[] }, ) { try { const { cardsetId, update } = data; @@ -93,84 +151,124 @@ export class CollaborationGateway `Sync request from user ${user.userId} for cardset ${cardsetId}`, ); - let state: Uint8Array; - if (update) { - state = await this.collaborationUseCase.applyUpdate( - parseInt(cardsetId), - update, + if (!update) { + client.emit('error', { message: 'Update data is required' }); + return; + } + + let doc = await this.yjsDocumentService.loadDocument(cardsetId); + if (!doc) { + doc = new Y.Doc(); + this.logger.log( + `Created new Yjs document for cardset ${cardsetId} during update`, ); - } else { - state = await this.collaborationUseCase.getState(parseInt(cardsetId)); } - client.emit('sync-response', { + const updateBuffer = new Uint8Array(update); + Y.applyUpdate(doc, updateBuffer); + + await this.yjsDocumentService.saveUpdate(cardsetId, updateBuffer); + + const state = Y.encodeStateAsUpdate(doc); + this.server.to(`cardset:${cardsetId}`).emit('sync', { cardsetId, - update: Array.from(state), + update: state, }); + this.logger.log( + `Sync update from user ${user.userId} broadcasted to all clients in cardset ${cardsetId}`, + ); } catch (error) { this.logger.error('Error during sync:', error); client.emit('error', { message: 'Sync failed' }); } } - @SubscribeMessage('update-card') - async handleUpdateCard( - @WsUser() user: UserAuth, - @ConnectedSocket() client: Socket, - @MessageBody() - data: { - cardsetId: string; - cardId: string; - updates: Partial<{ content: string; order: number }>; - }, - ) { + private async loadDocumentFromDBOrCreate(cardsetId: string): Promise { + const numericCardsetId = Number(cardsetId); + try { - const { cardsetId } = data; - const state = await this.collaborationUseCase.getState( - parseInt(cardsetId), + const doc = + await this.collaborationUseCase.loadCardsetContentFromDB( + numericCardsetId, + ); + if (doc) { + await this.yjsDocumentService + .saveDocument(cardsetId, doc) + .catch((error) => { + this.logger.warn( + `Failed to save document to Redis after DB load: ${error}`, + ); + }); + this.logger.log( + `Loaded Yjs document from DB and saved to Redis for cardset ${cardsetId}`, + ); + return doc; + } + } catch (error) { + this.logger.warn( + `Failed to load from DB for cardset ${cardsetId}, creating new document: ${error}`, ); + } - this.server.to(`cardset:${cardsetId}`).emit('sync-response', { - cardsetId, - update: Array.from(state), + return this.createNewDocument(cardsetId); + } + + private async createNewDocument(cardsetId: string): Promise { + const doc = new Y.Doc(); + this.logger.log(`Created new Yjs document for cardset ${cardsetId}`); + await this.yjsDocumentService + .saveDocument(cardsetId, doc) + .catch((error) => { + this.logger.warn( + `Failed to save new document to Redis: ${error}, continuing anyway`, + ); }); + return doc; + } - this.logger.log( - `User ${user.userId} updated card in cardset ${cardsetId}`, - ); - } catch (error) { - this.logger.error('Error updating card:', error); - client.emit('error', { message: 'Failed to update card' }); + private async removeClientFromAllCardsets(client: Socket) { + const cardsets = await this.yjsDocumentService.getClientCardsets(client.id); + if (cardsets.length === 0) return; + + for (const cardsetId of cardsets) { + void client.leave(`cardset:${cardsetId}`); + await this.yjsDocumentService.unregisterClient(cardsetId, client.id); + const activeCount = + await this.yjsDocumentService.getActiveClientCount(cardsetId); + if (activeCount === 0) { + this.scheduleFlush(cardsetId); + } } } - @SubscribeMessage('reorder-cards') - async handleReorderCards( - @WsUser() user: UserAuth, - @ConnectedSocket() client: Socket, - @MessageBody() - data: { - cardsetId: string; - cardOrders: { cardId: string; order: number }[]; - }, - ) { - try { - const { cardsetId } = data; - const state = await this.collaborationUseCase.getState( - parseInt(cardsetId), - ); + private scheduleFlush(cardsetId: string) { + if (this.flushTimeouts.has(cardsetId)) return; + const timeout = setTimeout(() => { + this.flushTimeouts.delete(cardsetId); + void this.flushCardset(cardsetId); + }, CollaborationGateway.FLUSH_DELAY_MS); + this.flushTimeouts.set(cardsetId, timeout); + this.logger.log(`Scheduled cardset ${cardsetId} flush`); + } - this.server.to(`cardset:${cardsetId}`).emit('sync-response', { - cardsetId, - update: Array.from(state), - }); + private clearScheduledFlush(cardsetId: string) { + const timeout = this.flushTimeouts.get(cardsetId); + if (timeout) { + clearTimeout(timeout); + this.flushTimeouts.delete(cardsetId); + } + } - this.logger.log( - `User ${user.userId} reordered cards in cardset ${cardsetId}`, - ); + private async flushCardset(cardsetId: string) { + const activeCount = + await this.yjsDocumentService.getActiveClientCount(cardsetId); + if (activeCount > 0) return; + + try { + await this.collaborationUseCase.saveCardsetContent(Number(cardsetId)); + this.logger.log(`Flushed cardset ${cardsetId} snapshot to database`); } catch (error) { - this.logger.error('Error reordering cards:', error); - client.emit('error', { message: 'Failed to reorder cards' }); + this.logger.error(`Failed to flush cardset ${cardsetId}:`, error); } } } diff --git a/src/collaboration/infrastructure/persistence/orm/cardset-content.orm-entity.ts b/src/collaboration/infrastructure/persistence/orm/cardset-content.orm-entity.ts new file mode 100644 index 0000000..d20d43e --- /dev/null +++ b/src/collaboration/infrastructure/persistence/orm/cardset-content.orm-entity.ts @@ -0,0 +1,25 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + UpdateDateColumn, +} from 'typeorm'; + +@Entity('cardset_content') +export class CardsetContentOrmEntity { + @PrimaryGeneratedColumn() + id!: number; + + @Column({ name: 'cardset_id', type: 'int', unique: true }) + cardsetId!: number; + + @Column({ type: 'text', nullable: true }) + content!: string; + + @CreateDateColumn({ name: 'created_at' }) + createdAt!: Date; + + @UpdateDateColumn({ name: 'updated_at' }) + updatedAt!: Date; +} diff --git a/src/collaboration/infrastructure/persistence/orm/cardset-incremental.orm-entity.ts b/src/collaboration/infrastructure/persistence/orm/cardset-incremental.orm-entity.ts new file mode 100644 index 0000000..d8b8285 --- /dev/null +++ b/src/collaboration/infrastructure/persistence/orm/cardset-incremental.orm-entity.ts @@ -0,0 +1,24 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, +} from 'typeorm'; + +@Entity('cardset_incrementals') +export class CardsetIncrementalOrmEntity { + @PrimaryGeneratedColumn() + id!: number; + + @Column({ name: 'cardset_id', type: 'int' }) + cardsetId!: number; + + @Column({ name: 'incremental_value', type: 'blob' }) + incrementalValue!: Buffer; + + @Column({ name: 'is_flushed', type: 'boolean', default: false }) + isFlushed!: boolean; + + @CreateDateColumn({ name: 'created_at' }) + createdAt!: Date; +} diff --git a/src/collaboration/infrastructure/redis/yjs-document.service.ts b/src/collaboration/infrastructure/redis/yjs-document.service.ts new file mode 100644 index 0000000..84aa52e --- /dev/null +++ b/src/collaboration/infrastructure/redis/yjs-document.service.ts @@ -0,0 +1,234 @@ +import { + Injectable, + OnModuleInit, + OnModuleDestroy, + Logger, +} from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { InjectRepository } from '@nestjs/typeorm'; +import Redis from 'ioredis'; +import * as Y from 'yjs'; +import { Repository } from 'typeorm'; +import { CardsetIncrementalOrmEntity } from '../persistence/orm/cardset-incremental.orm-entity'; + +@Injectable() +export class YjsDocumentService implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(YjsDocumentService.name); + private redisClient!: Redis; + private readonly mysqlFlushDelayMs: number; + private readonly flushTimeouts = new Map(); + + constructor( + private readonly configService: ConfigService, + @InjectRepository(CardsetIncrementalOrmEntity) + private readonly cardsetIncrementalRepository: Repository, + ) { + this.mysqlFlushDelayMs = + this.configService.get('YJS_MYSQL_FLUSH_DELAY_MS') ?? 60000; + } + + onModuleInit() { + this.redisClient = new Redis({ + host: this.configService.get('REDIS_HOST') ?? 'localhost', + port: this.configService.get('REDIS_PORT') ?? 6379, + password: this.configService.get('REDIS_PASSWORD'), + retryStrategy: (times) => Math.min(times * 50, 2000), + }); + + this.redisClient.on('connect', () => { + this.logger.log('Redis connected successfully'); + }); + + this.redisClient.on('error', (error) => { + this.logger.error('Redis connection error:', error); + }); + } + + onModuleDestroy() { + if (this.redisClient) { + this.redisClient.disconnect(); + this.logger.log('Redis disconnected'); + } + this.flushTimeouts.forEach((timeout) => clearTimeout(timeout)); + this.flushTimeouts.clear(); + } + + async saveDocument(cardsetId: string, doc: Y.Doc): Promise { + try { + if (!this.redisClient) return; + const state = Y.encodeStateAsUpdate(doc); + const key = `yjs:cardset:${cardsetId}`; + await this.redisClient.set(key, Buffer.from(state)); + await this.redisClient.expire(key, 86400 * 7); + } catch (error) { + this.logger.error( + `Failed to save document for cardset ${cardsetId}:`, + error, + ); + } + } + + async loadDocument(cardsetId: string): Promise { + try { + if (!this.redisClient) return null; + const data = await this.redisClient.getBuffer(`yjs:cardset:${cardsetId}`); + if (!data) return null; + const doc = new Y.Doc(); + Y.applyUpdate(doc, data); + return doc; + } catch (error) { + this.logger.error( + `Failed to load document for cardset ${cardsetId}:`, + error, + ); + return null; + } + } + + async saveUpdate(cardsetId: string, update: Uint8Array): Promise { + try { + if (!this.redisClient) return; + const key = `yjs:cardset:${cardsetId}`; + const historyKey = `yjs:cardset:${cardsetId}:updates`; + const updateBuffer = Buffer.from(update); + + const existingData = await this.redisClient.getBuffer(key); + if (existingData) { + const doc = new Y.Doc(); + Y.applyUpdate(doc, existingData); + Y.applyUpdate(doc, update); + const newState = Y.encodeStateAsUpdate(doc); + await this.redisClient.set(key, Buffer.from(newState)); + } else { + const doc = new Y.Doc(); + Y.applyUpdate(doc, update); + const state = Y.encodeStateAsUpdate(doc); + await this.redisClient.set(key, Buffer.from(state)); + } + await this.redisClient.expire(key, 86400 * 7); + await this.redisClient.rpush(historyKey, updateBuffer.toString('base64')); + + this.scheduleMySqlPersistence(cardsetId); + } catch (error) { + this.logger.error( + `Failed to save update for cardset ${cardsetId}:`, + error, + ); + throw error; + } + } + + async flushIncrementalHistory(cardsetId: string): Promise { + await this.persistCardsetIncrementals(cardsetId, true); + } + + async registerClient(cardsetId: string, clientId: string): Promise { + try { + const cardsetKey = `yjs:cardset:${cardsetId}:clients`; + const clientKey = `yjs:client:${clientId}:cardsets`; + await Promise.all([ + this.redisClient.sadd(cardsetKey, clientId), + this.redisClient.sadd(clientKey, cardsetId), + this.redisClient.expire(cardsetKey, 86400), + this.redisClient.expire(clientKey, 86400), + ]); + } catch (error) { + this.logger.error( + `Failed to register client ${clientId} for cardset ${cardsetId}:`, + error, + ); + } + } + + async unregisterClient(cardsetId: string, clientId: string): Promise { + try { + await Promise.all([ + this.redisClient.srem(`yjs:cardset:${cardsetId}:clients`, clientId), + this.redisClient.srem(`yjs:client:${clientId}:cardsets`, cardsetId), + ]); + } catch (error) { + this.logger.error( + `Failed to unregister client ${clientId} from cardset ${cardsetId}:`, + error, + ); + } + } + + async getActiveClientCount(cardsetId: string): Promise { + try { + return await this.redisClient.scard(`yjs:cardset:${cardsetId}:clients`); + } catch { + return 0; + } + } + + async getClientCardsets(clientId: string): Promise { + try { + return await this.redisClient.smembers(`yjs:client:${clientId}:cardsets`); + } catch { + return []; + } + } + + async deleteDocument(cardsetId: string): Promise { + try { + await this.redisClient.del(`yjs:cardset:${cardsetId}`); + } catch (error) { + this.logger.error( + `Failed to delete document for cardset ${cardsetId}:`, + error, + ); + throw error; + } + } + + private scheduleMySqlPersistence(cardsetId: string) { + const existing = this.flushTimeouts.get(cardsetId); + if (existing) clearTimeout(existing); + + const timeout = setTimeout(() => { + void this.persistCardsetIncrementals(cardsetId, false); + }, this.mysqlFlushDelayMs); + this.flushTimeouts.set(cardsetId, timeout); + } + + private async persistCardsetIncrementals( + cardsetId: string, + markAsFlushed: boolean, + ): Promise { + this.flushTimeouts.delete(cardsetId); + + const numericCardsetId = Number(cardsetId); + if (Number.isNaN(numericCardsetId)) { + this.logger.error( + `Cannot persist cardset ${cardsetId}: invalid numeric id`, + ); + return; + } + + const historyKey = `yjs:cardset:${cardsetId}:updates`; + try { + const updates = await this.redisClient.lrange(historyKey, 0, -1); + if (updates.length === 0) return; + + const entities = updates.map((base64Value) => + this.cardsetIncrementalRepository.create({ + cardsetId: numericCardsetId, + incrementalValue: Buffer.from(base64Value, 'base64'), + isFlushed: markAsFlushed, + }), + ); + await this.cardsetIncrementalRepository.save(entities); + await this.redisClient.del(historyKey); + + this.logger.log( + `Persisted ${entities.length} incremental updates for cardset ${cardsetId}`, + ); + } catch (error) { + this.logger.error( + `Failed to persist incremental updates for cardset ${cardsetId}:`, + error, + ); + } + } +} diff --git a/src/reaction/reaction.consumer.ts b/src/reaction/reaction.consumer.ts index 6dd4a9b..29dcd42 100644 --- a/src/reaction/reaction.consumer.ts +++ b/src/reaction/reaction.consumer.ts @@ -5,7 +5,11 @@ import { CARDSET_METADATA_REPOSITORY } from '../cardset/domain/repository/cardse import { Inject } from '@nestjs/common'; interface ReactionMessage { - eventType: 'LIKE_ADDED' | 'LIKE_REMOVED' | 'BOOKMARK_ADDED' | 'BOOKMARK_REMOVED'; + eventType: + | 'LIKE_ADDED' + | 'LIKE_REMOVED' + | 'BOOKMARK_ADDED' + | 'BOOKMARK_REMOVED'; targetType: string; targetId: number; userId: number; @@ -34,7 +38,9 @@ export class ReactionConsumer { if (msg.targetType !== 'CARD_SET') return; const cardSetId = Number(msg.targetId); - this.logger.log(`Reaction event: ${msg.eventType} for cardSetId=${cardSetId}`); + this.logger.log( + `Reaction event: ${msg.eventType} for cardSetId=${cardSetId}`, + ); switch (msg.eventType) { case 'LIKE_ADDED': diff --git a/src/reaction/reaction.module.ts b/src/reaction/reaction.module.ts index 43ec399..9d0d598 100644 --- a/src/reaction/reaction.module.ts +++ b/src/reaction/reaction.module.ts @@ -16,7 +16,10 @@ import { CARDSET_METADATA_REPOSITORY } from '../cardset/domain/repository/cardse TypeOrmModule.forFeature([CardSetMetadataOrmEntity]), ], providers: [ - { provide: CARDSET_METADATA_REPOSITORY, useClass: CardSetMetadataRepositoryImpl }, + { + provide: CARDSET_METADATA_REPOSITORY, + useClass: CardSetMetadataRepositoryImpl, + }, ReactionConsumer, ], exports: [CARDSET_METADATA_REPOSITORY],