feat(conversation): 실시간 subscription — graphql-ws + Redis PubSub

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` |

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:

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",

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,
         };
       },
     }),

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',
+  };
+});

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!
+}

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 {}

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';

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건) 유저 메시지보다 앞서 저장한다.

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);

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);

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<AsyncIterator<unknown>> {
+    const accountId = parseAccountId(user);
+    return this.subscriptionService.subscribeConversationMessages(
+      accountId,
+      conversationId,
+    );
+  }
+
+  @Subscription('myConversationUpdated', passthrough)
+  myConversationUpdated(
+    @CurrentUser() user: JwtUser,
+  ): Promise<AsyncIterator<unknown>> {
+    const accountId = parseAccountId(user);
+    return this.subscriptionService.subscribeMyConversationUpdates(accountId);
+  }
+
+  @Subscription('sellerConversationUpdated', passthrough)
+  sellerConversationUpdated(
+    @CurrentUser() user: JwtUser,
+  ): Promise<AsyncIterator<unknown>> {
+    const accountId = parseAccountId(user);
+    return this.subscriptionService.subscribeSellerConversationUpdates(
+      accountId,
+    );
+  }
+}