chore: refactor pub/sub iterator usage across schema files

- Updated pub/sub iterator methods from `asyncIterableIterator` to `asyncIterator` in multiple schema files for improved compatibility with the latest GraphQL subscriptions.
- Refactored subscription logic in CollaborationSession, Document, Message, User, and other schema files to enhance readability and maintainability.
- Adjusted imports in GraphQL builder to utilize RedisPubSub for better performance in subscription handling.

src/CollaborationSession/collaborationsession.schema.ts

@@ -324,7 +324,9 @@ export class CollaborationSessionSchema extends PothosSchema {
           if (!collaborationSession) {
             throw new Error('Collaboration session not found')
           }
-          return ctx.websocket.pubSub.asyncIterableIterator(`collaborationSessionUpdated:${collaborationSession.id}`)
+          return ctx.websocket.pubSub.asyncIterator([
+            `collaborationSessionUpdated:${collaborationSession.id}`,
+          ]) as unknown as AsyncIterable<CollaborationSession>
         },
         resolve: async (payload: CollaborationSession) => payload,
       }),

src/Document/document.schema.ts

@@ -447,9 +447,6 @@ export class DocumentSchema extends PothosSchema {
           if (!ctx.isSubscription) {
             throw new Error('Not allowed')
           }
-          const {
-            websocket: { pubSub },
-          } = ctx
           const documentId = args.documentId
           // check user permission
           const document = await this.prisma.document.findUnique({
@@ -469,7 +466,7 @@ export class DocumentSchema extends PothosSchema {
               throw new Error('User is not owner or collaborator of document')
             }
           }
-          return pubSub.asyncIterableIterator([
+          return ctx.websocket.pubSub.asyncIterator([
             `${DocumentEvent.CHANGED}.${documentId}`,
             `${DocumentEvent.DELETED}.${documentId}`,
             `${DocumentEvent.SAVED}.${documentId}`,

src/Graphql/graphql.builder.ts

@@ -12,8 +12,8 @@ import { User } from '@prisma/client'
 import { JsonValue } from '@prisma/client/runtime/library'
 import { Request, Response } from 'express'
 import { Kind, ValueNode } from 'graphql'
+import { RedisPubSub } from 'graphql-redis-subscriptions'
 import { JSONObjectResolver } from 'graphql-scalars'
-import { PubSub } from 'graphql-subscriptions'
 // @ts-expect-error
 import GraphQLUpload from 'graphql-upload/GraphQLUpload.mjs'
 // @ts-expect-error
@@ -30,7 +30,7 @@ export type SchemaContext =
       isSubscription: true
       websocket: {
         req: Request
-        pubSub: PubSub
+        pubSub: RedisPubSub
         sessionId: string
         me: User
         generator: PrismaCrudGenerator<BuilderTypes>
@@ -42,7 +42,7 @@ export type SchemaContext =
         req: Request
         res: Response
         me: User | null
-        pubSub: PubSub
+        pubSub: RedisPubSub
         invalidateCache: () => Promise<void>
         generator: PrismaCrudGenerator<BuilderTypes>
       }
@@ -101,8 +101,8 @@ export class Builder extends SchemaBuilder<SchemaBuilderOption> {
         debounceDelay: 1000,
         ...subscribeOptionsFromIterator((name, context) => {
           return context.isSubscription
-            ? context.websocket.pubSub.asyncIterableIterator(name)
-            : context.http.pubSub.asyncIterableIterator(name)
+            ? context.websocket.pubSub.asyncIterator(name)
+            : context.http.pubSub.asyncIterator(name)
         }),
       },
       zod: {

src/Graphql/graphql.module.ts

@@ -119,6 +119,8 @@ import { GraphqlService } from './graphql.service'
               }
               // @ts-expect-error: Request is not typed
               ctx.extra.request.headers['x-session-id'] = ctx.connectionParams['x-session-id']
+              // @ts-expect-error: Request is not typed
+              Logger.log(ctx.extra.request.headers['x-session-id'], 'Session ID')
             },
           },
         },

src/Message/message.schema.ts

@@ -197,10 +197,9 @@ export class MessageSchema extends PothosSchema {
           if (!ctx.isSubscription) {
             throw new Error('Not allowed')
           }
-          const {
-            websocket: { pubSub },
-          } = ctx
-          return pubSub.asyncIterableIterator([`${PubSubEvent.MESSAGE_SENT}.${args.chatRoomId}`])
+          return ctx.websocket.pubSub.asyncIterator([
+            `${PubSubEvent.MESSAGE_SENT}.${args.chatRoomId}`,
+          ]) as unknown as AsyncIterable<Message>
         },
         resolve: (payload: Message) => payload,
       }),

src/User/user.schema.ts

@@ -1,6 +1,6 @@
 import { clerkClient } from '@clerk/express'
 import { Inject, Injectable, Logger } from '@nestjs/common'
-import { ChatRoom, Message, MessageContextType, MessageType, Role } from '@prisma/client'
+import { Message, Role } from '@prisma/client'
 import { Pothos, PothosRef, PothosSchema, SchemaBuilderToken } from '@smatch-corp/nestjs-pothos'
 import { ChatroomSchema } from '../ChatRoom/chatroom.schema'
 import { Builder, SchemaContext } from '../Graphql/graphql.builder'
@@ -547,17 +547,15 @@ export class UserSchema extends PothosSchema {
     this.builder.subscriptionFields((t) => ({
       userScopedMessage: t.field({
         type: this.messageSchema.message(),
-        subscribe: async (_, _args, ctx: SchemaContext) => {
+        subscribe: async (_, _args, ctx) => {
           if (!ctx.isSubscription) {
             throw new Error('Not allowed')
           }
-          const {
-            websocket: { pubSub },
-          } = ctx
-          return pubSub.asyncIterableIterator([
+
+          return ctx.websocket.pubSub.asyncIterator([
             `${PubSubEvent.NEW_MESSAGE}.${ctx.websocket.me?.id}`,
             `${PubSubEvent.NOTIFICATION}.${ctx.websocket.me?.id}`,
-          ])
+          ]) as unknown as AsyncIterable<Message>
         },
         resolve: async (payload: Message) => payload,
       }),