diff --git a/README.md b/README.md index 399441a3..9a6545e0 100644 --- a/README.md +++ b/README.md @@ -269,6 +269,7 @@ yarn start:dev | --- | --- | | **서버** | `NODE_ENV`, `PORT`, `BACKEND_BASE_URL`, `FRONTEND_BASE_URL` | | **DB** | `DATABASE_URL` | +| **Redis (선택)** | `REDIS_URL` — GraphQL subscription PubSub. 미설정 시 `redis://localhost:6379`(로컬 docker-compose) | | **JWT / Auth** | `JWT_ACCESS_SECRET`, `JWT_ACCESS_EXPIRES_SECONDS`, `AUTH_REFRESH_EXPIRES_DAYS`, `AUTH_COOKIE_DOMAIN`, `AUTH_COOKIE_SECURE` | | **OIDC (Google)** | `OIDC_GOOGLE_CLIENT_ID`, `OIDC_GOOGLE_CLIENT_SECRET`, `OIDC_GOOGLE_ISSUER_URL` | | **OIDC (Kakao)** | `OIDC_KAKAO_CLIENT_ID`, `OIDC_KAKAO_CLIENT_SECRET`, `OIDC_KAKAO_ISSUER_URL` | diff --git a/docker-compose.yml b/docker-compose.yml index b62f95bf..12910912 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -25,5 +25,17 @@ services: timeout: 5s retries: 5 + redis: + image: redis:7-alpine + container_name: caquick-redis + # GraphQL subscription PubSub 백엔드. 영속화 불필요(휘발성 이벤트 브로커). + ports: + - '6379:6379' + healthcheck: + test: ['CMD', 'redis-cli', 'ping'] + interval: 10s + timeout: 5s + retries: 5 + volumes: caquick-mysql-data: diff --git a/package.json b/package.json index 9a07201d..595237ee 100644 --- a/package.json +++ b/package.json @@ -58,6 +58,10 @@ "cookie-parser": "^1.4.7", "express": "^5.2.1", "graphql": "^16.14.0", + "graphql-redis-subscriptions": "^2.7.0", + "graphql-subscriptions": "^3.0.0", + "graphql-ws": "^6.2.1", + "ioredis": "^5.3.2", "logform": "^2.7.0", "openid-client": "5.7.1", "passport": "^0.7.0", diff --git a/src/app.module.ts b/src/app.module.ts index b17dcddc..617e0121 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -13,13 +13,13 @@ import { ConfigModule, ConfigService } from '@nestjs/config'; import { GraphQLModule } from '@nestjs/graphql'; import { ScheduleModule } from '@nestjs/schedule'; import { ServeStaticModule } from '@nestjs/serve-static'; -import type { Request, Response } from 'express'; import { CommonModule } from '@/common/common.module'; import authConfig from '@/config/auth.config'; import databaseConfig from '@/config/database.config'; import docsConfig from '@/config/docs.config'; import oidcConfig from '@/config/oidc.config'; +import redisConfig from '@/config/redis.config'; import s3Config from '@/config/s3.config'; import { AuthModule } from '@/features/auth/auth.module'; import { ConversationModule } from '@/features/conversation'; @@ -31,9 +31,11 @@ import { StoreModule } from '@/features/store'; import { SystemModule } from '@/features/system/system.module'; import { UserModule } from '@/features/user/user.module'; import { AuthGlobalModule } from '@/global/auth/auth-global.module'; +import { buildGraphqlContext } from '@/global/graphql/graphql-context.helper'; import { GraphqlGlobalModule } from '@/global/graphql/graphql.module'; import { LoggerModule } from '@/global/logger/logger.module'; import { DocsAccessMiddleware } from '@/global/middlewares/docs-access.middleware'; +import { PubSubModule } from '@/global/pubsub'; import { RequestContextMiddleware, RequestContextModule, @@ -46,7 +48,14 @@ import { PrismaModule } from '@/prisma'; ConfigModule.forRoot({ isGlobal: true, cache: true, - load: [authConfig, databaseConfig, docsConfig, oidcConfig, s3Config], + load: [ + authConfig, + databaseConfig, + docsConfig, + oidcConfig, + redisConfig, + s3Config, + ], }), ServeStaticModule.forRoot({ rootPath: join(process.cwd(), 'public'), @@ -58,6 +67,7 @@ import { PrismaModule } from '@/prisma'; LoggerModule, AuthGlobalModule, GraphqlGlobalModule, + PubSubModule, StorageModule, // 인기 검색어 스냅샷 크론(SearchModule) 활성화 ScheduleModule.forRoot(), @@ -73,21 +83,17 @@ import { PrismaModule } from '@/prisma'; : join(process.cwd(), 'src/features/**/*.graphql'), ], playground: false, + // 실시간 subscription(graphql-ws). 인증은 connectionParams → + // buildGraphqlContext가 HTTP 헤더로 이식해 기존 JWT 가드를 재사용한다. + subscriptions: { + 'graphql-ws': true, + }, plugins: [ isProd ? ApolloServerPluginLandingPageDisabled() : ApolloServerPluginLandingPageLocalDefault({ embed: true }), ], - context: ({ - req, - res, - }: { - req: Request; - res?: Response; - }): { - req: Request; - res?: Response; - } => ({ req, res }), + context: buildGraphqlContext, }; }, }), diff --git a/src/config/redis.config.ts b/src/config/redis.config.ts new file mode 100644 index 00000000..47ede9df --- /dev/null +++ b/src/config/redis.config.ts @@ -0,0 +1,20 @@ +import { registerAs } from '@nestjs/config'; + +/** + * Redis 설정 타입 (GraphQL subscription PubSub용) + */ +export interface RedisConfig { + url: string; +} + +/** + * Redis 설정. + * DATABASE_URL과 달리 미설정 시 로컬 docker-compose 기본값으로 폴백한다 — + * 현재 배포 인프라가 꺼져 있고 FE도 로컬 백엔드로 테스트하는 개발 단계라, + * 필수 강제보다 로컬 DX(compose up 후 바로 동작)를 우선한다. + */ +export default registerAs('redis', (): RedisConfig => { + return { + url: process.env.REDIS_URL ?? 'redis://localhost:6379', + }; +}); diff --git a/src/features/conversation/conversation-subscription.graphql b/src/features/conversation/conversation-subscription.graphql new file mode 100644 index 00000000..ebd22cd3 --- /dev/null +++ b/src/features/conversation/conversation-subscription.graphql @@ -0,0 +1,40 @@ +extend type Subscription { + """ + 대화방 신규 메시지 실시간 구독. 구매자(대화 소유자)와 판매자(해당 매장 + 소유자)만 구독할 수 있다. 인증은 graphql-ws connectionParams.authorization. + """ + conversationMessageAdded(conversationId: ID!): ConversationMessage! + """구매자 대화 목록/배지 갱신 이벤트(새 메시지 도착 시).""" + myConversationUpdated: ConversationListUpdate! + """판매자 대화 목록 갱신 이벤트(고객 메시지 도착 시).""" + sellerConversationUpdated: SellerConversationListUpdate! +} + +""" +구매자 대화 목록 갱신 이벤트. +이벤트 간 도착 순서는 보장되지 않는다 — 구독자는 (lastMessageAt, lastReadAt)을 +사전식으로 비교해 이미 표시 중인 상태보다 오래된 이벤트를 폐기해야 한다. +읽음 처리는 lastReadAt만 전진시키므로, 읽음 이후 도착한 지연 이벤트도 이 비교로 +걸러진다(메시지 스트림은 id 기준). +""" +type ConversationListUpdate { + conversationId: ID! + storeId: ID! + storeName: String! + """마지막 메시지 미리보기(HTML은 태그 제거)""" + lastMessagePreview: String + lastMessageAt: DateTime! + """이벤트 스냅샷 시점의 구매자 마지막 읽음 시각(폐기 규칙 비교용)""" + lastReadAt: DateTime + """이벤트 시점의 안읽은 수신 메시지 수""" + unreadCount: Int! +} + +"""판매자 대화 목록 갱신 이벤트. 도착 순서 비보장 — lastMessageAt 기준 폐기 규칙 동일.""" +type SellerConversationListUpdate { + conversationId: ID! + accountId: ID! + """마지막 메시지 미리보기(HTML은 태그 제거)""" + lastMessagePreview: String + lastMessageAt: DateTime! +} diff --git a/src/features/conversation/conversation.module.ts b/src/features/conversation/conversation.module.ts index a07dcf76..493424e9 100644 --- a/src/features/conversation/conversation.module.ts +++ b/src/features/conversation/conversation.module.ts @@ -4,18 +4,24 @@ import { ConversationRepository } from '@/features/conversation/repositories/con import { ConversationCenterQueryResolver } from '@/features/conversation/resolvers/conversation-center-query.resolver'; import { ConversationInquiryMutationResolver } from '@/features/conversation/resolvers/conversation-inquiry-mutation.resolver'; import { ConversationInquiryQueryResolver } from '@/features/conversation/resolvers/conversation-inquiry-query.resolver'; +import { ConversationSubscriptionResolver } from '@/features/conversation/resolvers/conversation-subscription.resolver'; import { ConversationCenterService } from '@/features/conversation/services/conversation-center.service'; +import { ConversationEventsService } from '@/features/conversation/services/conversation-events.service'; import { ConversationInquiryService } from '@/features/conversation/services/conversation-inquiry.service'; +import { ConversationSubscriptionService } from '@/features/conversation/services/conversation-subscription.service'; @Module({ providers: [ ConversationRepository, + ConversationEventsService, ConversationInquiryService, ConversationCenterService, + ConversationSubscriptionService, ConversationInquiryQueryResolver, ConversationInquiryMutationResolver, ConversationCenterQueryResolver, + ConversationSubscriptionResolver, ], - exports: [ConversationRepository], + exports: [ConversationRepository, ConversationEventsService], }) export class ConversationModule {} diff --git a/src/features/conversation/index.ts b/src/features/conversation/index.ts index 85b8af14..4db65102 100644 --- a/src/features/conversation/index.ts +++ b/src/features/conversation/index.ts @@ -1,3 +1,7 @@ // cross-feature 공개 API. 단일 구현 repo라 토큰/인터페이스 없이 구체 클래스로 주입(의도적). export { ConversationModule } from '@/features/conversation/conversation.module'; export { ConversationRepository } from '@/features/conversation/repositories/conversation.repository'; +// subscription 이벤트 발행/구독 어댑터 — 판매자 답장(seller feature)도 같은 토픽을 쓴다. +export { ConversationEventsService } from '@/features/conversation/services/conversation-events.service'; +export { toEventPreview } from '@/features/conversation/services/conversation-events-mappers.helper'; +export { toLastMessagePreview } from '@/features/conversation/services/conversation-center-mappers.helper'; diff --git a/src/features/conversation/repositories/conversation.repository.ts b/src/features/conversation/repositories/conversation.repository.ts index 71a1c9ae..93c58a45 100644 --- a/src/features/conversation/repositories/conversation.repository.ts +++ b/src/features/conversation/repositories/conversation.repository.ts @@ -295,6 +295,69 @@ export class ConversationRepository { }); } + /** + * 목록 갱신 이벤트용 대화 스냅샷 — 한 트랜잭션(단일 REPEATABLE READ + * 스냅샷)에서 대화·매장명·최신 메시지·안읽음 수를 함께 읽는다. 독립 + * 조회로 쪼개면 경쟁 커밋이 끼어들어 "B의 미리보기 + A의 시각" 같은 + * 혼합 상태가 이벤트로 나갈 수 있다(리뷰 반영). + */ + async getConversationEventSnapshot(conversationId: bigint) { + return this.prisma.$transaction(async (tx) => { + const conversation = await tx.storeConversation.findFirst({ + where: { id: conversationId, deleted_at: undefined }, + select: { + id: true, + account_id: true, + store_id: true, + last_message_at: true, + last_read_at: true, + store: { select: { store_name: true } }, + }, + }); + if (!conversation) return null; + + const [lastMessage, unreadCount] = await Promise.all([ + tx.storeConversationMessage.findFirst({ + where: { conversation_id: conversationId }, + orderBy: { id: 'desc' }, + select: { body_format: true, body_text: true, body_html: true }, + }), + tx.storeConversationMessage.count({ + where: { + conversation_id: conversationId, + sender_type: { not: ConversationSenderType.USER }, + ...(conversation.last_read_at + ? { created_at: { gt: conversation.last_read_at } } + : {}), + }, + }), + ]); + + return { conversation, lastMessage, unreadCount }; + }); + } + + /** subscription 구독 권한 판정용 — 대화 소유 구매자/해당 매장 판매자 확인. */ + async findConversationAccess(conversationId: bigint) { + return this.prisma.storeConversation.findFirst({ + where: { id: conversationId }, + select: { + id: true, + account_id: true, + store_id: true, + store: { select: { seller_account_id: true } }, + }, + }); + } + + /** 판매자 구독 대상 매장(활성) 조회. */ + async findStoreBySellerAccount(sellerAccountId: bigint) { + return this.prisma.store.findFirst({ + where: { seller_account_id: sellerAccountId, ...activeWhere }, + select: { id: true }, + }); + } + /** * 구매자 메시지 저장. 대화가 없으면 같은 트랜잭션에서 생성하고, 인사말은 * "대화의 첫 메시지"일 때만(메시지 0건) 유저 메시지보다 앞서 저장한다. diff --git a/src/features/conversation/resolvers/conversation-center.resolver.spec.ts b/src/features/conversation/resolvers/conversation-center.resolver.spec.ts index 825dd52c..765d571f 100644 --- a/src/features/conversation/resolvers/conversation-center.resolver.spec.ts +++ b/src/features/conversation/resolvers/conversation-center.resolver.spec.ts @@ -1,11 +1,14 @@ // 전체 경로(리졸버→서비스→레포→DB) 통합 검증만 담당. 분기/집계 세부 검증은 service.spec.ts에서 담당 import type { PrismaClient } from '@prisma/client'; +import { PubSub } from 'graphql-subscriptions'; import { ConversationRepository } from '@/features/conversation/repositories/conversation.repository'; import { ConversationCenterQueryResolver } from '@/features/conversation/resolvers/conversation-center-query.resolver'; import { ConversationInquiryMutationResolver } from '@/features/conversation/resolvers/conversation-inquiry-mutation.resolver'; import { ConversationCenterService } from '@/features/conversation/services/conversation-center.service'; +import { ConversationEventsService } from '@/features/conversation/services/conversation-events.service'; import { ConversationInquiryService } from '@/features/conversation/services/conversation-inquiry.service'; +import { PUB_SUB } from '@/global/pubsub'; import { disconnectTestPrismaClient } from '@/test/db/prisma-test-client'; import { closeTruncateConnection, truncateAll } from '@/test/db/truncate'; import { @@ -28,6 +31,8 @@ describe('Conversation Center Resolvers (real DB)', () => { ConversationCenterService, ConversationInquiryService, ConversationRepository, + ConversationEventsService, + { provide: PUB_SUB, useValue: new PubSub() }, ], }); centerResolver = module.get(ConversationCenterQueryResolver); diff --git a/src/features/conversation/resolvers/conversation-inquiry.resolver.spec.ts b/src/features/conversation/resolvers/conversation-inquiry.resolver.spec.ts index 5663ba64..a992999d 100644 --- a/src/features/conversation/resolvers/conversation-inquiry.resolver.spec.ts +++ b/src/features/conversation/resolvers/conversation-inquiry.resolver.spec.ts @@ -1,10 +1,13 @@ // 전체 경로(리졸버→서비스→레포→DB) 통합 검증만 담당. 분기·예외 세부는 service.spec.ts에서 담당 import type { PrismaClient } from '@prisma/client'; +import { PubSub } from 'graphql-subscriptions'; import { ConversationRepository } from '@/features/conversation/repositories/conversation.repository'; import { ConversationInquiryMutationResolver } from '@/features/conversation/resolvers/conversation-inquiry-mutation.resolver'; import { ConversationInquiryQueryResolver } from '@/features/conversation/resolvers/conversation-inquiry-query.resolver'; +import { ConversationEventsService } from '@/features/conversation/services/conversation-events.service'; import { ConversationInquiryService } from '@/features/conversation/services/conversation-inquiry.service'; +import { PUB_SUB } from '@/global/pubsub'; import { disconnectTestPrismaClient } from '@/test/db/prisma-test-client'; import { closeTruncateConnection, truncateAll } from '@/test/db/truncate'; import { @@ -26,6 +29,8 @@ describe('Conversation Inquiry Resolvers (real DB)', () => { ConversationInquiryMutationResolver, ConversationInquiryService, ConversationRepository, + ConversationEventsService, + { provide: PUB_SUB, useValue: new PubSub() }, ], }); queryResolver = module.get(ConversationInquiryQueryResolver); diff --git a/src/features/conversation/resolvers/conversation-subscription.resolver.ts b/src/features/conversation/resolvers/conversation-subscription.resolver.ts new file mode 100644 index 00000000..d8a49a81 --- /dev/null +++ b/src/features/conversation/resolvers/conversation-subscription.resolver.ts @@ -0,0 +1,54 @@ +import { UseGuards } from '@nestjs/common'; +import { Args, Resolver, Subscription } from '@nestjs/graphql'; + +import { ConversationSubscriptionService } from '@/features/conversation/services/conversation-subscription.service'; +import { + CurrentUser, + JwtAuthGuard, + parseAccountId, + type JwtUser, +} from '@/global/auth'; + +/** + * 이벤트 payload는 발행 시 이미 GraphQL 출력 형태로 조립돼 있어 + * resolve는 payload를 그대로 통과시킨다. + */ +const passthrough = { resolve: (payload: unknown): unknown => payload }; + +@Resolver('Subscription') +@UseGuards(JwtAuthGuard) +export class ConversationSubscriptionResolver { + constructor( + private readonly subscriptionService: ConversationSubscriptionService, + ) {} + + @Subscription('conversationMessageAdded', passthrough) + conversationMessageAdded( + @CurrentUser() user: JwtUser, + @Args('conversationId') conversationId: string, + ): Promise> { + const accountId = parseAccountId(user); + return this.subscriptionService.subscribeConversationMessages( + accountId, + conversationId, + ); + } + + @Subscription('myConversationUpdated', passthrough) + myConversationUpdated( + @CurrentUser() user: JwtUser, + ): Promise> { + const accountId = parseAccountId(user); + return this.subscriptionService.subscribeMyConversationUpdates(accountId); + } + + @Subscription('sellerConversationUpdated', passthrough) + sellerConversationUpdated( + @CurrentUser() user: JwtUser, + ): Promise> { + const accountId = parseAccountId(user); + return this.subscriptionService.subscribeSellerConversationUpdates( + accountId, + ); + } +} diff --git a/src/features/conversation/services/conversation-events-mappers.helper.spec.ts b/src/features/conversation/services/conversation-events-mappers.helper.spec.ts new file mode 100644 index 00000000..ef9fec7d --- /dev/null +++ b/src/features/conversation/services/conversation-events-mappers.helper.spec.ts @@ -0,0 +1,47 @@ +import { + toConversationMessageEvent, + toEventPreview, +} from '@/features/conversation/services/conversation-events-mappers.helper'; +import type { ConversationMessageOutput } from '@/features/conversation/types/conversation-output.type'; + +function message( + overrides: Partial = {}, +): ConversationMessageOutput { + return { + id: '1', + conversationId: '2', + senderType: 'USER', + bodyFormat: 'TEXT', + bodyText: '안녕하세요', + bodyHtml: null, + createdAt: new Date('2026-08-01T12:00:00Z'), + ...overrides, + }; +} + +describe('conversation-events-mappers.helper', () => { + it('메시지 출력의 날짜를 ISO 문자열로 바꿔 이벤트 payload를 만든다', () => { + expect(toConversationMessageEvent(message())).toEqual({ + id: '1', + conversationId: '2', + senderType: 'USER', + bodyFormat: 'TEXT', + bodyText: '안녕하세요', + bodyHtml: null, + createdAt: '2026-08-01T12:00:00.000Z', + }); + }); + + it('미리보기는 TEXT 원문 / HTML 태그 제거 텍스트를 쓴다', () => { + expect(toEventPreview(message())).toBe('안녕하세요'); + expect( + toEventPreview( + message({ + bodyFormat: 'HTML', + bodyText: null, + bodyHtml: '

자동 응답

', + }), + ), + ).toBe('자동 응답'); + }); +}); diff --git a/src/features/conversation/services/conversation-events-mappers.helper.ts b/src/features/conversation/services/conversation-events-mappers.helper.ts new file mode 100644 index 00000000..ad16bbc3 --- /dev/null +++ b/src/features/conversation/services/conversation-events-mappers.helper.ts @@ -0,0 +1,33 @@ +import { toLastMessagePreview } from '@/features/conversation/services/conversation-center-mappers.helper'; +import type { + ConversationMessageEvent, + ConversationMessageOutput, +} from '@/features/conversation/types/conversation-output.type'; + +/** DI-free 순수 함수만 둔다 — subscription 이벤트 payload 변환. */ + +/** 메시지 출력 → 이벤트 payload(날짜는 ISO 문자열). */ +export function toConversationMessageEvent( + message: ConversationMessageOutput, +): ConversationMessageEvent { + return { + id: message.id, + conversationId: message.conversationId, + senderType: message.senderType, + bodyFormat: message.bodyFormat, + bodyText: message.bodyText, + bodyHtml: message.bodyHtml, + createdAt: message.createdAt.toISOString(), + }; +} + +/** 메시지 출력 → 목록 이벤트용 미리보기 텍스트. */ +export function toEventPreview( + message: ConversationMessageOutput, +): string | null { + return toLastMessagePreview({ + body_format: message.bodyFormat, + body_text: message.bodyText, + body_html: message.bodyHtml, + }); +} diff --git a/src/features/conversation/services/conversation-events.service.spec.ts b/src/features/conversation/services/conversation-events.service.spec.ts new file mode 100644 index 00000000..9a6cec35 --- /dev/null +++ b/src/features/conversation/services/conversation-events.service.spec.ts @@ -0,0 +1,150 @@ +/** + * 실 Redis(testcontainers) 기반 발행/구독 왕복 검증 — JSON 직렬화를 거친 + * payload가 구독자에게 그대로 도착하는지까지 확인한다(DB 불필요). + */ +import { RedisPubSub } from 'graphql-redis-subscriptions'; +import type { PubSubEngine } from 'graphql-subscriptions'; +import Redis from 'ioredis'; +import { GenericContainer, type StartedTestContainer } from 'testcontainers'; + +import { ConversationEventsService } from '@/features/conversation/services/conversation-events.service'; +import type { ConversationMessageOutput } from '@/features/conversation/types/conversation-output.type'; + +jest.setTimeout(180_000); + +describe('ConversationEventsService (real Redis)', () => { + let container: StartedTestContainer; + let pubSub: RedisPubSub; + let service: ConversationEventsService; + + beforeAll(async () => { + container = await new GenericContainer('redis:7-alpine') + .withExposedPorts(6379) + .start(); + const url = `redis://${container.getHost()}:${container.getMappedPort(6379)}`; + pubSub = new RedisPubSub({ + publisher: new Redis(url), + subscriber: new Redis(url), + }); + service = new ConversationEventsService(pubSub); + }); + + afterAll(async () => { + await pubSub.close(); + await container.stop(); + }); + + async function nextEvent(iterator: AsyncIterator) { + return (await iterator.next()).value; + } + + /** + * asyncIterableIterator는 첫 next() 호출 시점에 Redis SUBSCRIBE를 보낸다 — + * 발행 전에 next()를 먼저 걸고 SUBSCRIBE 완료를 잠깐 기다려야 이벤트를 + * 놓치지 않는다(구독 등록 전 발행분은 유실되는 게 Pub/Sub 의미론). + */ + async function startListening(iterator: AsyncIterator) { + const pending = nextEvent(iterator); + await new Promise((resolve) => setTimeout(resolve, 300)); + // Promise를 그대로 반환하면 호출부의 await가 이벤트 도착까지 평탄화해 + // 기다려 버린다 — 객체로 감싸 구독 시작만 보장하고 대기는 호출부 몫으로. + return { pending }; + } + + it('대화방 메시지 발행이 해당 대화 구독자에게만 도착한다', async () => { + const target = service.messageAddedIterator('10'); + const other = service.messageAddedIterator('99'); + const { pending: pendingTarget } = await startListening(target); + const otherReceived = jest.fn(); + const { pending: pendingOther } = await startListening(other); + void pendingOther.then(otherReceived); + + const message: ConversationMessageOutput = { + id: '1', + conversationId: '10', + senderType: 'USER', + bodyFormat: 'TEXT', + bodyText: '픽업 문의', + bodyHtml: null, + createdAt: new Date('2026-08-01T12:00:00Z'), + }; + await service.publishMessagesAdded([message]); + + // Redis JSON 왕복 후에도 이벤트 payload가 보존된다(날짜는 ISO 문자열) + await expect(pendingTarget).resolves.toEqual({ + id: '1', + conversationId: '10', + senderType: 'USER', + bodyFormat: 'TEXT', + bodyText: '픽업 문의', + bodyHtml: null, + createdAt: '2026-08-01T12:00:00.000Z', + }); + expect(otherReceived).not.toHaveBeenCalled(); + await other.return?.(); + }); + + it('구매자/판매자 목록 갱신 이벤트가 각 토픽으로 도착한다', async () => { + const buyer = service.buyerListIterator('7'); + const seller = service.sellerListIterator('3'); + const { pending: pendingBuyer } = await startListening(buyer); + const { pending: pendingSeller } = await startListening(seller); + + await service.publishBuyerListUpdate('7', { + conversationId: '10', + storeId: '3', + storeName: '해즈 케이크', + lastMessagePreview: '답변 드리겠습니다', + lastMessageAt: '2026-08-01T12:00:00.000Z', + lastReadAt: null, + unreadCount: 2, + }); + await service.publishSellerListUpdate('3', { + conversationId: '10', + accountId: '7', + lastMessagePreview: '픽업 문의', + lastMessageAt: '2026-08-01T12:00:00.000Z', + }); + + await expect(pendingBuyer).resolves.toMatchObject({ + storeName: '해즈 케이크', + unreadCount: 2, + }); + await expect(pendingSeller).resolves.toMatchObject({ + accountId: '7', + lastMessagePreview: '픽업 문의', + }); + }); + + it('발행 실패는 삼킨다 — 커밋된 전송을 Redis 장애가 실패로 만들지 않는다', async () => { + const failing = { + publish: jest.fn().mockRejectedValue(new Error('redis down')), + } as unknown as PubSubEngine; + const failingService = new ConversationEventsService(failing); + + await expect( + failingService.publishBuyerListUpdate('1', { + conversationId: '1', + storeId: '2', + storeName: '매장', + lastMessagePreview: null, + lastMessageAt: '2026-08-01T12:00:00.000Z', + lastReadAt: null, + unreadCount: 0, + }), + ).resolves.toBeUndefined(); + await expect( + failingService.publishMessagesAdded([ + { + id: '1', + conversationId: '1', + senderType: 'USER', + bodyFormat: 'TEXT', + bodyText: '문의', + bodyHtml: null, + createdAt: new Date(), + }, + ]), + ).resolves.toBeUndefined(); + }); +}); diff --git a/src/features/conversation/services/conversation-events.service.ts b/src/features/conversation/services/conversation-events.service.ts new file mode 100644 index 00000000..5b760acd --- /dev/null +++ b/src/features/conversation/services/conversation-events.service.ts @@ -0,0 +1,89 @@ +import { Inject, Injectable, Logger } from '@nestjs/common'; +import type { PubSubEngine } from 'graphql-subscriptions'; + +import { toConversationMessageEvent } from '@/features/conversation/services/conversation-events-mappers.helper'; +import type { + ConversationListUpdateEvent, + ConversationMessageOutput, + SellerConversationListUpdateEvent, +} from '@/features/conversation/types/conversation-output.type'; +import { PUB_SUB } from '@/global/pubsub'; + +/** + * 대화 subscription 이벤트 발행/구독 어댑터. + * 토픽 문자열은 여기서만 조립한다 — 발행자(구매자 전송·FAQ 자동응답·판매자 + * 답장)와 구독 리졸버가 같은 토픽을 보게 하는 단일 소스. + */ +@Injectable() +export class ConversationEventsService { + private readonly logger = new Logger(ConversationEventsService.name); + + constructor(@Inject(PUB_SUB) private readonly pubSub: PubSubEngine) {} + + /** + * 발행은 DB 커밋 이후의 부수효과 — Redis 장애가 이미 성공한 전송을 + * 실패로 둔갑시키면 클라이언트 재시도로 중복 전송이 난다(리뷰 반영). + * 실패는 경고 로그만 남기고 삼킨다(구독자는 재조회 폴백). + */ + private async safePublish(topic: string, payload: unknown): Promise { + try { + await this.pubSub.publish(topic, payload); + } catch (e) { + this.logger.warn( + `subscription publish 실패 (topic=${topic}): ${ + e instanceof Error ? e.message : String(e) + }`, + ); + } + } + + private messageTopic(conversationId: string): string { + return `conversation.message.${conversationId}`; + } + + private buyerTopic(accountId: string): string { + return `conversation.buyer.${accountId}`; + } + + private sellerTopic(storeId: string): string { + return `conversation.seller.${storeId}`; + } + + /** 저장된 메시지들을 대화방 토픽에 순서대로 발행한다. */ + async publishMessagesAdded( + messages: ConversationMessageOutput[], + ): Promise { + for (const message of messages) { + await this.safePublish( + this.messageTopic(message.conversationId), + toConversationMessageEvent(message), + ); + } + } + + async publishBuyerListUpdate( + accountId: string, + event: ConversationListUpdateEvent, + ): Promise { + await this.safePublish(this.buyerTopic(accountId), event); + } + + async publishSellerListUpdate( + storeId: string, + event: SellerConversationListUpdateEvent, + ): Promise { + await this.safePublish(this.sellerTopic(storeId), event); + } + + messageAddedIterator(conversationId: string): AsyncIterator { + return this.pubSub.asyncIterableIterator(this.messageTopic(conversationId)); + } + + buyerListIterator(accountId: string): AsyncIterator { + return this.pubSub.asyncIterableIterator(this.buyerTopic(accountId)); + } + + sellerListIterator(storeId: string): AsyncIterator { + return this.pubSub.asyncIterableIterator(this.sellerTopic(storeId)); + } +} diff --git a/src/features/conversation/services/conversation-inquiry.service.spec.ts b/src/features/conversation/services/conversation-inquiry.service.spec.ts index 75d0c94e..3db802fd 100644 --- a/src/features/conversation/services/conversation-inquiry.service.spec.ts +++ b/src/features/conversation/services/conversation-inquiry.service.spec.ts @@ -5,9 +5,12 @@ import { UnauthorizedException, } from '@nestjs/common'; import type { PrismaClient } from '@prisma/client'; +import { PubSub } from 'graphql-subscriptions'; import { ConversationRepository } from '@/features/conversation/repositories/conversation.repository'; +import { ConversationEventsService } from '@/features/conversation/services/conversation-events.service'; import { ConversationInquiryService } from '@/features/conversation/services/conversation-inquiry.service'; +import { PUB_SUB } from '@/global/pubsub'; import { disconnectTestPrismaClient } from '@/test/db/prisma-test-client'; import { closeTruncateConnection, truncateAll } from '@/test/db/truncate'; import { @@ -23,7 +26,13 @@ describe('ConversationInquiryService (real DB)', () => { beforeAll(async () => { const { module, prisma: p } = await createTestingModuleWithRealDb({ - providers: [ConversationInquiryService, ConversationRepository], + providers: [ + ConversationInquiryService, + ConversationRepository, + ConversationEventsService, + // 발행 경로 실검증은 events service spec(실 Redis) 담당 — 여기선 in-memory + { provide: PUB_SUB, useValue: new PubSub() }, + ], }); service = module.get(ConversationInquiryService); prisma = p; diff --git a/src/features/conversation/services/conversation-inquiry.service.ts b/src/features/conversation/services/conversation-inquiry.service.ts index a64a9f0d..b6ebbded 100644 --- a/src/features/conversation/services/conversation-inquiry.service.ts +++ b/src/features/conversation/services/conversation-inquiry.service.ts @@ -1,4 +1,4 @@ -import { Injectable, NotFoundException } from '@nestjs/common'; +import { Injectable, Logger, NotFoundException } from '@nestjs/common'; import { ConversationBodyFormat, ConversationSenderType } from '@prisma/client'; import { parseId } from '@/common/utils/id-parser'; @@ -12,6 +12,9 @@ import { type ConversationMessageEntry, } from '@/features/conversation/repositories/conversation.repository'; import { ConversationBaseService } from '@/features/conversation/services/conversation-base.service'; +import { toLastMessagePreview } from '@/features/conversation/services/conversation-center-mappers.helper'; +import { toEventPreview } from '@/features/conversation/services/conversation-events-mappers.helper'; +import { ConversationEventsService } from '@/features/conversation/services/conversation-events.service'; import { renderGreeting, toConversationMessageOutput, @@ -24,7 +27,12 @@ import type { @Injectable() export class ConversationInquiryService extends ConversationBaseService { - constructor(repo: ConversationRepository) { + private readonly logger = new Logger(ConversationInquiryService.name); + + constructor( + repo: ConversationRepository, + private readonly events: ConversationEventsService, + ) { super(repo); } @@ -151,12 +159,96 @@ export class ConversationInquiryService extends ConversationBaseService { entries: args.entries, }); + const messages = result.messages.map(toConversationMessageOutput); + await this.publishBuyerSendEvents({ + accountId: args.accountId, + storeId: args.storeId, + storeName: args.storeName, + conversationId: result.conversationId, + messages, + }); + return { conversationId: result.conversationId.toString(), - messages: result.messages.map(toConversationMessageOutput), + messages, }; } + /** + * 실시간 이벤트 발행 — 대화방 메시지 + 양측 목록/배지 갱신. + * 저장 트랜잭션 밖의 부수효과라 실패해도 전송 자체는 성공으로 남는다 + * (구독자는 폴백 재조회 가능). + */ + private async publishBuyerSendEvents(args: { + accountId: bigint; + storeId: bigint; + storeName: string; + conversationId: bigint; + messages: ConversationMessagesPayload['messages']; + }): Promise { + // 커밋 이후의 부수효과 전체(스냅샷 조회 포함)를 격리한다 — 여기서 나는 + // 예외가 mutation을 실패로 둔갑시키면 클라이언트 재시도로 중복 전송이 + // 난다(리뷰 반영). 실패는 경고 로그만 남긴다. + try { + await this.doPublishBuyerSendEvents(args); + } catch (e) { + this.logger.warn( + `대화 이벤트 발행 실패 (conversationId=${args.conversationId}): ${ + e instanceof Error ? e.message : String(e) + }`, + ); + } + } + + private async doPublishBuyerSendEvents(args: { + accountId: bigint; + storeId: bigint; + storeName: string; + conversationId: bigint; + messages: ConversationMessagesPayload['messages']; + }): Promise { + const lastMessage = args.messages[args.messages.length - 1]; + if (!lastMessage) return; + + // 목록 이벤트는 "발행 시점의 최신 커밋 상태"를 단일 트랜잭션 스냅샷 + // 으로 다시 읽어 조립한다 — 독립 조회로 쪼개면 경쟁 커밋이 끼어들어 + // 혼합 상태(남의 미리보기 + 내 시각)가 나갈 수 있다(리뷰 반영). + // 메시지 스트림 이벤트는 id를 실어 구독자가 정렬한다. + const snapshot = await this.repo.getConversationEventSnapshot( + args.conversationId, + ); + + const preview = snapshot + ? toLastMessagePreview(snapshot.lastMessage) + : toEventPreview(lastMessage); + const lastMessageAtIso = ( + snapshot?.conversation.last_message_at ?? lastMessage.createdAt + ).toISOString(); + const lastReadAtIso = + snapshot?.conversation.last_read_at?.toISOString() ?? null; + const unreadCount = snapshot?.unreadCount ?? 0; + // 매장명도 스냅샷 값을 우선한다 — 최초 조회 후 개명되면 최신 메시지 + // 상태에 옛 이름이 실려 나갈 수 있다(리뷰 반영) + const storeName = snapshot?.conversation.store.store_name ?? args.storeName; + + await this.events.publishMessagesAdded(args.messages); + await this.events.publishBuyerListUpdate(args.accountId.toString(), { + conversationId: args.conversationId.toString(), + storeId: args.storeId.toString(), + storeName, + lastMessagePreview: preview, + lastMessageAt: lastMessageAtIso, + lastReadAt: lastReadAtIso, + unreadCount, + }); + await this.events.publishSellerListUpdate(args.storeId.toString(), { + conversationId: args.conversationId.toString(), + accountId: args.accountId.toString(), + lastMessagePreview: preview, + lastMessageAt: lastMessageAtIso, + }); + } + private async requireInquiryStore(storeId: bigint) { const store = await this.repo.findInquiryStore(storeId); if (!store) { diff --git a/src/features/conversation/services/conversation-subscription.service.spec.ts b/src/features/conversation/services/conversation-subscription.service.spec.ts new file mode 100644 index 00000000..8d9c7c65 --- /dev/null +++ b/src/features/conversation/services/conversation-subscription.service.spec.ts @@ -0,0 +1,180 @@ +import { + ForbiddenException, + NotFoundException, + UnauthorizedException, +} from '@nestjs/common'; +import type { PrismaClient } from '@prisma/client'; +import { PubSub } from 'graphql-subscriptions'; + +import { ConversationRepository } from '@/features/conversation/repositories/conversation.repository'; +import { ConversationEventsService } from '@/features/conversation/services/conversation-events.service'; +import { ConversationInquiryService } from '@/features/conversation/services/conversation-inquiry.service'; +import { ConversationSubscriptionService } from '@/features/conversation/services/conversation-subscription.service'; +import { PUB_SUB } from '@/global/pubsub'; +import { disconnectTestPrismaClient } from '@/test/db/prisma-test-client'; +import { closeTruncateConnection, truncateAll } from '@/test/db/truncate'; +import { + createAccount, + createStore, + createUserProfile, +} from '@/test/factories'; +import { createTestingModuleWithRealDb } from '@/test/modules/testing-module.builder'; + +describe('ConversationSubscriptionService (real DB)', () => { + let service: ConversationSubscriptionService; + let inquiryService: ConversationInquiryService; + let prisma: PrismaClient; + + beforeAll(async () => { + const { module, prisma: p } = await createTestingModuleWithRealDb({ + providers: [ + ConversationSubscriptionService, + ConversationInquiryService, + ConversationEventsService, + ConversationRepository, + // 발행-구독 왕복은 실 Redis spec(events service) 담당 — 여기선 in-memory + { provide: PUB_SUB, useValue: new PubSub() }, + ], + }); + service = module.get(ConversationSubscriptionService); + inquiryService = module.get(ConversationInquiryService); + prisma = p; + }); + + afterAll(async () => { + await closeTruncateConnection(); + await disconnectTestPrismaClient(); + }); + + beforeEach(async () => { + await truncateAll(); + }); + + async function setupBuyer() { + const account = await createAccount(prisma, { account_type: 'USER' }); + await createUserProfile(prisma, { account_id: account.id }); + return account; + } + + describe('subscribeConversationMessages', () => { + it('대화 소유 구매자와 해당 매장 판매자는 구독할 수 있고, 제3자는 NotFound', async () => { + const buyer = await setupBuyer(); + const stranger = await setupBuyer(); + const seller = await createAccount(prisma, { account_type: 'SELLER' }); + const store = await createStore(prisma, { + seller_account_id: seller.id, + }); + const conversation = await prisma.storeConversation.create({ + data: { account_id: buyer.id, store_id: store.id }, + }); + const id = conversation.id.toString(); + + await expect( + service.subscribeConversationMessages(buyer.id, id), + ).resolves.toBeDefined(); + await expect( + service.subscribeConversationMessages(seller.id, id), + ).resolves.toBeDefined(); + await expect( + service.subscribeConversationMessages(stranger.id, id), + ).rejects.toThrow(NotFoundException); + await expect( + service.subscribeConversationMessages(buyer.id, '999999'), + ).rejects.toThrow(NotFoundException); + }); + + it('구독 중이면 구매자 전송 이벤트를 실제로 수신한다(발행 경로 통합)', async () => { + const buyer = await setupBuyer(); + const store = await createStore(prisma, { store_name: '해즈 케이크' }); + + // 첫 전송으로 대화 생성 → 그 대화를 구독 → 두 번째 전송 수신 확인 + const first = await inquiryService.sendConversationMessage(buyer.id, { + storeId: store.id.toString(), + bodyText: '첫 문의', + }); + const iterator = await service.subscribeConversationMessages( + buyer.id, + first.conversationId, + ); + const pending = iterator.next(); + + await inquiryService.sendConversationMessage(buyer.id, { + storeId: store.id.toString(), + bodyText: '추가 문의', + }); + + const { value } = await pending; + expect(value).toMatchObject({ + conversationId: first.conversationId, + senderType: 'USER', + bodyText: '추가 문의', + }); + await iterator.return?.(); + }); + }); + + describe('subscribeMyConversationUpdates', () => { + it('활성 USER만 구독 가능하고, 전송 시 목록 갱신 이벤트를 수신한다', async () => { + const buyer = await setupBuyer(); + const seller = await createAccount(prisma, { account_type: 'SELLER' }); + const store = await createStore(prisma, { store_name: '달콤 케이크' }); + + await expect( + service.subscribeMyConversationUpdates(BigInt(999999)), + ).rejects.toThrow(UnauthorizedException); + await expect( + service.subscribeMyConversationUpdates(seller.id), + ).rejects.toThrow(ForbiddenException); + + const iterator = await service.subscribeMyConversationUpdates(buyer.id); + const pending = iterator.next(); + + await inquiryService.sendConversationMessage(buyer.id, { + storeId: store.id.toString(), + bodyText: '문의합니다', + }); + + const { value } = await pending; + expect(value).toMatchObject({ + storeId: store.id.toString(), + storeName: '달콤 케이크', + lastMessagePreview: '문의합니다', + // 인사말은 mutation 응답으로 즉시 표시돼 읽음 처리된다 + unreadCount: 0, + }); + await iterator.return?.(); + }); + }); + + describe('subscribeSellerConversationUpdates', () => { + it('매장 보유 판매자만 구독 가능하고, 고객 전송 이벤트를 수신한다', async () => { + const buyer = await setupBuyer(); + const seller = await createAccount(prisma, { account_type: 'SELLER' }); + const store = await createStore(prisma, { + seller_account_id: seller.id, + }); + + // 매장 없는 계정은 구독 불가 + await expect( + service.subscribeSellerConversationUpdates(buyer.id), + ).rejects.toThrow(NotFoundException); + + const iterator = await service.subscribeSellerConversationUpdates( + seller.id, + ); + const pending = iterator.next(); + + await inquiryService.sendConversationMessage(buyer.id, { + storeId: store.id.toString(), + bodyText: '픽업 시간 문의', + }); + + const { value } = await pending; + expect(value).toMatchObject({ + accountId: buyer.id.toString(), + lastMessagePreview: '픽업 시간 문의', + }); + await iterator.return?.(); + }); + }); +}); diff --git a/src/features/conversation/services/conversation-subscription.service.ts b/src/features/conversation/services/conversation-subscription.service.ts new file mode 100644 index 00000000..cfa61147 --- /dev/null +++ b/src/features/conversation/services/conversation-subscription.service.ts @@ -0,0 +1,60 @@ +import { Injectable, NotFoundException } from '@nestjs/common'; + +import { parseId } from '@/common/utils/id-parser'; +import { CONVERSATION_ERRORS } from '@/features/conversation/constants/conversation-error-messages'; +import { ConversationRepository } from '@/features/conversation/repositories/conversation.repository'; +import { ConversationBaseService } from '@/features/conversation/services/conversation-base.service'; +import { ConversationEventsService } from '@/features/conversation/services/conversation-events.service'; + +/** + * subscription 구독 진입점 — 구독 권한 검증 후 토픽 iterator를 돌려준다. + * 이벤트 발행은 각 전송 서비스(구매자 전송·판매자 답장)가 담당한다. + */ +@Injectable() +export class ConversationSubscriptionService extends ConversationBaseService { + constructor( + repo: ConversationRepository, + private readonly events: ConversationEventsService, + ) { + super(repo); + } + + /** 대화방 메시지 구독 — 대화 소유 구매자 또는 해당 매장 판매자만. */ + async subscribeConversationMessages( + accountId: bigint, + conversationIdRaw: string, + ): Promise> { + const conversationId = parseId(conversationIdRaw); + const conversation = await this.repo.findConversationAccess(conversationId); + + // 존재하지 않는 대화와 권한 없는 대화를 구분하지 않는다(존재 여부 노출 방지) + const allowed = + conversation && + (conversation.account_id === accountId || + conversation.store.seller_account_id === accountId); + if (!allowed) { + throw new NotFoundException(CONVERSATION_ERRORS.CONVERSATION_NOT_FOUND); + } + + return this.events.messageAddedIterator(conversationId.toString()); + } + + /** 구매자 대화 목록/배지 갱신 구독. */ + async subscribeMyConversationUpdates( + accountId: bigint, + ): Promise> { + await this.requireActiveUser(accountId); + return this.events.buyerListIterator(accountId.toString()); + } + + /** 판매자 대화 목록 갱신 구독. */ + async subscribeSellerConversationUpdates( + accountId: bigint, + ): Promise> { + const store = await this.repo.findStoreBySellerAccount(accountId); + if (!store) { + throw new NotFoundException(CONVERSATION_ERRORS.STORE_NOT_FOUND); + } + return this.events.sellerListIterator(store.id.toString()); + } +} diff --git a/src/features/conversation/types/conversation-output.type.ts b/src/features/conversation/types/conversation-output.type.ts index aa04943e..b1965fed 100644 --- a/src/features/conversation/types/conversation-output.type.ts +++ b/src/features/conversation/types/conversation-output.type.ts @@ -63,3 +63,34 @@ export interface ConversationMessageConnection { hasMore: boolean; nextCursor: string | null; } + +/** + * subscription 이벤트 payload — Redis JSON 직렬화를 거치므로 날짜는 ISO + * 문자열로 나른다(DateTime 스칼라가 문자열도 직렬화 가능). + */ +export interface ConversationMessageEvent { + id: string; + conversationId: string; + senderType: ConversationSenderType; + bodyFormat: ConversationBodyFormat; + bodyText: string | null; + bodyHtml: string | null; + createdAt: string; +} + +export interface ConversationListUpdateEvent { + conversationId: string; + storeId: string; + storeName: string; + lastMessagePreview: string | null; + lastMessageAt: string; + lastReadAt: string | null; + unreadCount: number; +} + +export interface SellerConversationListUpdateEvent { + conversationId: string; + accountId: string; + lastMessagePreview: string | null; + lastMessageAt: string; +} diff --git a/src/features/core/root.graphql b/src/features/core/root.graphql index c74beee4..f956249f 100644 --- a/src/features/core/root.graphql +++ b/src/features/core/root.graphql @@ -12,3 +12,8 @@ type Mutation { """스키마 루트 유지를 위한 no-op 필드""" _noop: Boolean } + +type Subscription { + """스키마 루트 유지를 위한 no-op 필드""" + _subscriptionNoop: Boolean +} diff --git a/src/features/seller/resolvers/seller-conversation.resolver.spec.ts b/src/features/seller/resolvers/seller-conversation.resolver.spec.ts index 4eac2235..559631d2 100644 --- a/src/features/seller/resolvers/seller-conversation.resolver.spec.ts +++ b/src/features/seller/resolvers/seller-conversation.resolver.spec.ts @@ -1,13 +1,16 @@ import { NotFoundException } from '@nestjs/common'; import type { PrismaClient } from '@prisma/client'; +import { PubSub } from 'graphql-subscriptions'; import { AUDIT_LOG_REPOSITORY } from '@/features/audit-log'; import { AuditLogRepository } from '@/features/audit-log/repositories/audit-log.repository'; import { ConversationRepository } from '@/features/conversation'; +import { ConversationEventsService } from '@/features/conversation'; import { SellerRepository } from '@/features/seller/repositories/seller.repository'; import { SellerConversationMutationResolver } from '@/features/seller/resolvers/seller-conversation-mutation.resolver'; import { SellerConversationQueryResolver } from '@/features/seller/resolvers/seller-conversation-query.resolver'; import { SellerConversationService } from '@/features/seller/services/seller-conversation.service'; +import { PUB_SUB } from '@/global/pubsub'; import { disconnectTestPrismaClient } from '@/test/db/prisma-test-client'; import { closeTruncateConnection, truncateAll } from '@/test/db/truncate'; import { createAccount, setupSellerWithStore } from '@/test/factories'; @@ -26,6 +29,8 @@ describe('Seller Conversation Resolvers (real DB)', () => { SellerConversationService, SellerRepository, ConversationRepository, + ConversationEventsService, + { provide: PUB_SUB, useValue: new PubSub() }, { provide: AUDIT_LOG_REPOSITORY, useClass: AuditLogRepository, diff --git a/src/features/seller/services/seller-conversation.service.spec.ts b/src/features/seller/services/seller-conversation.service.spec.ts index 8ae22cdd..98ea43ba 100644 --- a/src/features/seller/services/seller-conversation.service.spec.ts +++ b/src/features/seller/services/seller-conversation.service.spec.ts @@ -1,11 +1,14 @@ import { BadRequestException, NotFoundException } from '@nestjs/common'; import type { PrismaClient } from '@prisma/client'; +import { PubSub } from 'graphql-subscriptions'; import { AUDIT_LOG_REPOSITORY } from '@/features/audit-log'; import { AuditLogRepository } from '@/features/audit-log/repositories/audit-log.repository'; import { ConversationRepository } from '@/features/conversation'; +import { ConversationEventsService } from '@/features/conversation'; import { SellerRepository } from '@/features/seller/repositories/seller.repository'; import { SellerConversationService } from '@/features/seller/services/seller-conversation.service'; +import { PUB_SUB } from '@/global/pubsub'; import { disconnectTestPrismaClient } from '@/test/db/prisma-test-client'; import { closeTruncateConnection, truncateAll } from '@/test/db/truncate'; import { createAccount, setupSellerWithStore } from '@/test/factories'; @@ -21,6 +24,8 @@ describe('SellerConversationService (real DB)', () => { SellerConversationService, SellerRepository, ConversationRepository, + ConversationEventsService, + { provide: PUB_SUB, useValue: new PubSub() }, { provide: AUDIT_LOG_REPOSITORY, useClass: AuditLogRepository, diff --git a/src/features/seller/services/seller-conversation.service.ts b/src/features/seller/services/seller-conversation.service.ts index e0401dec..4d46a9ed 100644 --- a/src/features/seller/services/seller-conversation.service.ts +++ b/src/features/seller/services/seller-conversation.service.ts @@ -2,6 +2,7 @@ import { BadRequestException, Inject, Injectable, + Logger, NotFoundException, } from '@nestjs/common'; import { @@ -16,7 +17,12 @@ import { AUDIT_LOG_REPOSITORY, type IAuditLogRepository, } from '@/features/audit-log'; -import { ConversationRepository } from '@/features/conversation'; +import { + ConversationEventsService, + ConversationRepository, + toEventPreview, + toLastMessagePreview, +} from '@/features/conversation'; import { BODY_HTML_REQUIRED, BODY_TEXT_REQUIRED, @@ -43,11 +49,14 @@ import type { @Injectable() export class SellerConversationService extends SellerBaseService { + private readonly logger = new Logger(SellerConversationService.name); + constructor( repo: SellerRepository, @Inject(AUDIT_LOG_REPOSITORY) auditLogs: IAuditLogRepository, private readonly conversationRepository: ConversationRepository, + private readonly conversationEvents: ConversationEventsService, ) { super(repo, auditLogs); } @@ -156,7 +165,99 @@ export class SellerConversationService extends SellerBaseService { }, }); - return this.toConversationMessageOutput(row); + const output = this.toConversationMessageOutput(row); + await this.publishSellerReplyEvents({ + conversation, + storeId: ctx.storeId, + message: output, + }); + + return output; + } + + /** + * 실시간 이벤트 발행 — 대화방 메시지 + 구매자·판매자 목록 갱신. + * 저장 트랜잭션 밖의 부수효과라 실패해도 답장 자체는 성공으로 남는다. + */ + private async publishSellerReplyEvents(args: { + conversation: { + id: bigint; + account_id: bigint; + last_read_at: Date | null; + }; + storeId: bigint; + message: SellerConversationMessageOutput; + }): Promise { + // 커밋 이후의 부수효과 전체(스냅샷 조회 포함)를 격리한다 — 예외가 + // 이미 저장된 답장을 실패로 둔갑시키면 재시도 중복이 난다(리뷰 반영). + try { + await this.doPublishSellerReplyEvents(args); + } catch (e) { + this.logger.warn( + `대화 이벤트 발행 실패 (conversationId=${args.conversation.id}): ${ + e instanceof Error ? e.message : String(e) + }`, + ); + } + } + + private async doPublishSellerReplyEvents(args: { + conversation: { + id: bigint; + account_id: bigint; + last_read_at: Date | null; + }; + storeId: bigint; + message: SellerConversationMessageOutput; + }): Promise { + const message = { + id: args.message.id, + conversationId: args.message.conversationId, + senderType: args.message.senderType, + bodyFormat: args.message.bodyFormat, + bodyText: args.message.bodyText, + bodyHtml: args.message.bodyHtml, + createdAt: args.message.createdAt, + }; + // 목록 이벤트는 발행 시점의 최신 커밋 상태를 단일 트랜잭션 스냅샷으로 + // 조립한다 — 독립 조회로 쪼개면 경쟁 커밋이 끼어들어 혼합 상태가 나갈 + // 수 있다(리뷰 반영). 메시지 스트림은 id 정렬. + const snapshot = + await this.conversationRepository.getConversationEventSnapshot( + args.conversation.id, + ); + const preview = snapshot + ? toLastMessagePreview(snapshot.lastMessage) + : toEventPreview(message); + const lastMessageAtIso = ( + snapshot?.conversation.last_message_at ?? args.message.createdAt + ).toISOString(); + const storeName = snapshot?.conversation.store.store_name ?? ''; + const lastReadAtIso = + snapshot?.conversation.last_read_at?.toISOString() ?? null; + + await this.conversationEvents.publishMessagesAdded([message]); + await this.conversationEvents.publishBuyerListUpdate( + args.conversation.account_id.toString(), + { + conversationId: args.conversation.id.toString(), + storeId: args.storeId.toString(), + storeName, + lastMessagePreview: preview, + lastMessageAt: lastMessageAtIso, + lastReadAt: lastReadAtIso, + unreadCount: snapshot?.unreadCount ?? 0, + }, + ); + await this.conversationEvents.publishSellerListUpdate( + args.storeId.toString(), + { + conversationId: args.conversation.id.toString(), + accountId: args.conversation.account_id.toString(), + lastMessagePreview: preview, + lastMessageAt: lastMessageAtIso, + }, + ); } private toConversationBodyFormat(raw: string): ConversationBodyFormat { diff --git a/src/global/graphql/graphql-context.helper.spec.ts b/src/global/graphql/graphql-context.helper.spec.ts new file mode 100644 index 00000000..8a69e3af --- /dev/null +++ b/src/global/graphql/graphql-context.helper.spec.ts @@ -0,0 +1,39 @@ +import type { IncomingMessage } from 'node:http'; + +import type { Request, Response } from 'express'; + +import { buildGraphqlContext } from '@/global/graphql/graphql-context.helper'; + +describe('buildGraphqlContext', () => { + it('HTTP 요청은 req/res를 그대로 전달한다', () => { + const req = { headers: {} } as Request; + const res = {} as Response; + + expect(buildGraphqlContext({ req, res })).toEqual({ req, res }); + }); + + it('ws 연결은 connectionParams.authorization을 헤더로 이식한다', () => { + const request = { headers: { host: 'localhost' } } as IncomingMessage; + + const ctx = buildGraphqlContext({ + extra: { request }, + connectionParams: { authorization: 'Bearer token-123' }, + }); + + expect(ctx.req).toBe(request); + expect(request.headers.authorization).toBe('Bearer token-123'); + expect(request.headers.host).toBe('localhost'); + }); + + it('authorization이 없거나 문자열이 아니면 헤더를 건드리지 않는다', () => { + const request = { headers: {} } as IncomingMessage; + + buildGraphqlContext({ + extra: { request }, + connectionParams: { authorization: 123 }, + }); + buildGraphqlContext({ extra: { request } }); + + expect(request.headers.authorization).toBeUndefined(); + }); +}); diff --git a/src/global/graphql/graphql-context.helper.ts b/src/global/graphql/graphql-context.helper.ts new file mode 100644 index 00000000..bb39112e --- /dev/null +++ b/src/global/graphql/graphql-context.helper.ts @@ -0,0 +1,34 @@ +import type { IncomingMessage } from 'node:http'; + +import type { Request, Response } from 'express'; + +/** + * HTTP·WebSocket(graphql-ws) 공용 GraphQL context 조립. + * + * ws 연결의 인증 토큰은 connectionParams.authorization으로 들어오므로, + * upgrade 요청 객체에 HTTP 헤더 형태로 이식해 기존 JwtAuthGuard/passport + * 경로(req.headers.authorization)를 그대로 태운다 — 가드 이원화 방지. + */ +export interface GraphqlContextArgs { + req?: Request; + res?: Response; + extra?: { request?: IncomingMessage }; + connectionParams?: Record; +} + +export interface GraphqlContext { + req: Request | IncomingMessage | undefined; + res?: Response; +} + +export function buildGraphqlContext(ctx: GraphqlContextArgs): GraphqlContext { + const wsRequest = ctx?.extra?.request; + if (wsRequest) { + const authorization = ctx.connectionParams?.authorization; + if (typeof authorization === 'string' && authorization.length > 0) { + wsRequest.headers = { ...wsRequest.headers, authorization }; + } + return { req: wsRequest }; + } + return { req: ctx?.req, res: ctx?.res }; +} diff --git a/src/global/pubsub/index.ts b/src/global/pubsub/index.ts new file mode 100644 index 00000000..1f881398 --- /dev/null +++ b/src/global/pubsub/index.ts @@ -0,0 +1,2 @@ +export { PubSubModule } from '@/global/pubsub/pubsub.module'; +export { PUB_SUB } from '@/global/pubsub/pubsub.constants'; diff --git a/src/global/pubsub/pubsub.constants.ts b/src/global/pubsub/pubsub.constants.ts new file mode 100644 index 00000000..4630172f --- /dev/null +++ b/src/global/pubsub/pubsub.constants.ts @@ -0,0 +1,5 @@ +/** + * PubSub 주입 토큰. cross-cutting 포트(외부 부수효과 어댑터)라 토큰 주입을 + * 사용한다 — 프로덕션은 RedisPubSub, spec은 in-memory PubSub로 대체된다. + */ +export const PUB_SUB = Symbol('PUB_SUB'); diff --git a/src/global/pubsub/pubsub.module.ts b/src/global/pubsub/pubsub.module.ts new file mode 100644 index 00000000..bf27a4ef --- /dev/null +++ b/src/global/pubsub/pubsub.module.ts @@ -0,0 +1,47 @@ +import { Global, Module, type OnModuleDestroy, Inject } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import { RedisPubSub } from 'graphql-redis-subscriptions'; +import Redis, { type RedisOptions } from 'ioredis'; + +import type { RedisConfig } from '@/config/redis.config'; +import { PUB_SUB } from '@/global/pubsub/pubsub.constants'; + +/** + * GraphQL subscription용 Redis PubSub 전역 모듈. + * 수평 확장(다중 인스턴스) 시에도 이벤트가 모든 인스턴스에 전파되도록 + * 처음부터 Redis 백엔드를 쓴다(사용자 확정). publisher/subscriber는 + * Redis 프로토콜 제약(SUBSCRIBE 중 일반 명령 불가) 때문에 커넥션을 분리한다. + */ +@Global() +@Module({ + providers: [ + { + provide: PUB_SUB, + inject: [ConfigService], + useFactory: (config: ConfigService): RedisPubSub => { + const redisConfig = config.getOrThrow('redis'); + const options: RedisOptions = { + // Redis 미기동 시 부팅을 막지 않고 재시도만 한다(로컬 DX) + retryStrategy: (times: number) => Math.min(times * 500, 5000), + // 발행은 DB 커밋 이후의 부수효과 — 무한 재시도(null)로 두면 Redis + // 장애 시 mutation 응답이 매달린다(리뷰 반영). 짧게 실패시키고 + // 실패 처리는 발행부(try/catch)가 담당한다. + maxRetriesPerRequest: 2, + enableOfflineQueue: false, + }; + return new RedisPubSub({ + publisher: new Redis(redisConfig.url, options), + subscriber: new Redis(redisConfig.url, options), + }); + }, + }, + ], + exports: [PUB_SUB], +}) +export class PubSubModule implements OnModuleDestroy { + constructor(@Inject(PUB_SUB) private readonly pubSub: RedisPubSub) {} + + async onModuleDestroy(): Promise { + await this.pubSub.close(); + } +} diff --git a/yarn.lock b/yarn.lock index c6094299..abb88dd2 100644 --- a/yarn.lock +++ b/yarn.lock @@ -3398,6 +3398,13 @@ __metadata: languageName: node linkType: hard +"@ioredis/commands@npm:1.10.0": + version: 1.10.0 + resolution: "@ioredis/commands@npm:1.10.0" + checksum: 10c0/baf91e62d0e64ef2b5f7ca4413dc2456fe250e87483beac4a1c8ef1fe5ad0d2fcdeb9b89d4556d8ef6c7455c64a964359d729601fdb06b2f4c76c35dd59afa99 + languageName: node + linkType: hard + "@isaacs/balanced-match@npm:^4.0.1": version: 4.0.1 resolution: "@isaacs/balanced-match@npm:4.0.1" @@ -7394,7 +7401,11 @@ __metadata: express: "npm:^5.2.1" globals: "npm:^17.11.0" graphql: "npm:^16.14.0" + graphql-redis-subscriptions: "npm:^2.7.0" + graphql-subscriptions: "npm:^3.0.0" + graphql-ws: "npm:^6.2.1" husky: "npm:^9.1.7" + ioredis: "npm:^5.3.2" jest: "npm:^30" knip: "npm:^6.32.2" lint-staged: "npm:^17.3.0" @@ -7722,6 +7733,13 @@ __metadata: languageName: node linkType: hard +"cluster-key-slot@npm:1.1.1": + version: 1.1.1 + resolution: "cluster-key-slot@npm:1.1.1" + checksum: 10c0/079b1ae86b20e2d53308a877b08de5e830722a45c07810569d0dab4955bed569da33ac9f79998289d014adf02cca7223a0647cb0ee6548a12ab3c4f9beac1377 + languageName: node + linkType: hard + "co@npm:^4.6.0": version: 4.6.0 resolution: "co@npm:4.6.0" @@ -8440,6 +8458,13 @@ __metadata: languageName: node linkType: hard +"denque@npm:2.1.0": + version: 2.1.0 + resolution: "denque@npm:2.1.0" + checksum: 10c0/f9ef81aa0af9c6c614a727cb3bd13c5d7db2af1abf9e6352045b86e85873e629690f6222f4edd49d10e4ccf8f078bbeec0794fafaf61b659c0589d0c511ec363 + languageName: node + linkType: hard + "depd@npm:2.0.0, depd@npm:^2.0.0, depd@npm:~2.0.0": version: 2.0.0 resolution: "depd@npm:2.0.0" @@ -10377,6 +10402,20 @@ __metadata: languageName: node linkType: hard +"graphql-redis-subscriptions@npm:^2.7.0": + version: 2.7.0 + resolution: "graphql-redis-subscriptions@npm:2.7.0" + dependencies: + ioredis: "npm:^5.3.2" + peerDependencies: + graphql-subscriptions: ^1.0.0 || ^2.0.0 || ^3.0.0 + dependenciesMeta: + ioredis: + optional: true + checksum: 10c0/f98e9a16aa60d5470f6916f5a85b0b91898e3ec341a70ae3ddac878aa5b415dae9081ba872afdab5873cef3933fde1c3f1ee690ffa6d5b6d164a9165aed5cad1 + languageName: node + linkType: hard + "graphql-scalars@npm:^1.15.0": version: 1.25.0 resolution: "graphql-scalars@npm:1.25.0" @@ -10388,6 +10427,15 @@ __metadata: languageName: node linkType: hard +"graphql-subscriptions@npm:^3.0.0": + version: 3.0.0 + resolution: "graphql-subscriptions@npm:3.0.0" + peerDependencies: + graphql: ^15.7.2 || ^16.0.0 + checksum: 10c0/10445c0d376773d15c887237ac460226b03e0b9c76789846f9133e3cfba6275b529f946634cd5a4e1f1f6c728b960714c8efc6c504af20ab05a3a1219ac9cc09 + languageName: node + linkType: hard + "graphql-tag@npm:2.12.6, graphql-tag@npm:^2.11.0": version: 2.12.6 resolution: "graphql-tag@npm:2.12.6" @@ -10399,7 +10447,7 @@ __metadata: languageName: node linkType: hard -"graphql-ws@npm:6.2.1": +"graphql-ws@npm:6.2.1, graphql-ws@npm:^6.2.1": version: 6.2.1 resolution: "graphql-ws@npm:6.2.1" peerDependencies: @@ -11096,6 +11144,21 @@ __metadata: languageName: node linkType: hard +"ioredis@npm:^5.3.2": + version: 5.11.1 + resolution: "ioredis@npm:5.11.1" + dependencies: + "@ioredis/commands": "npm:1.10.0" + cluster-key-slot: "npm:1.1.1" + debug: "npm:4.4.3" + denque: "npm:2.1.0" + redis-errors: "npm:1.2.0" + redis-parser: "npm:3.0.0" + standard-as-callback: "npm:2.1.0" + checksum: 10c0/a8b27043cf2c045dfc93f40a32ce24cf9f8b57799a37f4234c4b925c365ccf131629590f94a512f546fda2ba8ed034009c94c4933ecd44c50bc166636d929fd6 + languageName: node + linkType: hard + "ip-address@npm:^10.0.1": version: 10.1.0 resolution: "ip-address@npm:10.1.0" @@ -14619,6 +14682,22 @@ __metadata: languageName: node linkType: hard +"redis-errors@npm:1.2.0, redis-errors@npm:^1.0.0": + version: 1.2.0 + resolution: "redis-errors@npm:1.2.0" + checksum: 10c0/5b316736e9f532d91a35bff631335137a4f974927bb2fb42bf8c2f18879173a211787db8ac4c3fde8f75ed6233eb0888e55d52510b5620e30d69d7d719c8b8a7 + languageName: node + linkType: hard + +"redis-parser@npm:3.0.0": + version: 3.0.0 + resolution: "redis-parser@npm:3.0.0" + dependencies: + redis-errors: "npm:^1.0.0" + checksum: 10c0/ee16ac4c7b2a60b1f42a2cdaee22b005bd4453eb2d0588b8a4939718997ae269da717434da5d570fe0b05030466eeb3f902a58cf2e8e1ca058bf6c9c596f632f + languageName: node + linkType: hard + "reflect-metadata@npm:^0.2.2": version: 0.2.2 resolution: "reflect-metadata@npm:0.2.2" @@ -15524,6 +15603,13 @@ __metadata: languageName: node linkType: hard +"standard-as-callback@npm:2.1.0": + version: 2.1.0 + resolution: "standard-as-callback@npm:2.1.0" + checksum: 10c0/012677236e3d3fdc5689d29e64ea8a599331c4babe86956bf92fc5e127d53f85411c5536ee0079c52c43beb0026b5ce7aa1d834dd35dd026e82a15d1bcaead1f + languageName: node + linkType: hard + "statuses@npm:>= 1.5.0 < 2, statuses@npm:~1.5.0": version: 1.5.0 resolution: "statuses@npm:1.5.0"