From ba32ee8dc1f11325100f71af6a3725886fee2456 Mon Sep 17 00:00:00 2001 From: Ly Tuan Kiet Date: Sun, 8 Dec 2024 21:54:23 +0700 Subject: [PATCH] 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. --- .../collaborationsession.schema.ts | 4 +++- src/Document/document.schema.ts | 5 +---- src/Graphql/graphql.builder.ts | 10 +++++----- src/Graphql/graphql.module.ts | 2 ++ src/Message/message.schema.ts | 7 +++---- src/User/user.schema.ts | 12 +++++------- 6 files changed, 19 insertions(+), 21 deletions(-) diff --git a/src/CollaborationSession/collaborationsession.schema.ts b/src/CollaborationSession/collaborationsession.schema.ts index 4e3c0f0..e5a674b 100644 --- a/src/CollaborationSession/collaborationsession.schema.ts +++ b/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 }, resolve: async (payload: CollaborationSession) => payload, }), diff --git a/src/Document/document.schema.ts b/src/Document/document.schema.ts index 55fb8b8..7d64d03 100644 --- a/src/Document/document.schema.ts +++ b/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}`, diff --git a/src/Graphql/graphql.builder.ts b/src/Graphql/graphql.builder.ts index d80a216..311a600 100644 --- a/src/Graphql/graphql.builder.ts +++ b/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 @@ -42,7 +42,7 @@ export type SchemaContext = req: Request res: Response me: User | null - pubSub: PubSub + pubSub: RedisPubSub invalidateCache: () => Promise generator: PrismaCrudGenerator } @@ -101,8 +101,8 @@ export class Builder extends SchemaBuilder { 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: { diff --git a/src/Graphql/graphql.module.ts b/src/Graphql/graphql.module.ts index 4b1e3a5..a6eafa4 100644 --- a/src/Graphql/graphql.module.ts +++ b/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') }, }, }, diff --git a/src/Message/message.schema.ts b/src/Message/message.schema.ts index 8b66fe4..3aaad03 100644 --- a/src/Message/message.schema.ts +++ b/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 }, resolve: (payload: Message) => payload, }), diff --git a/src/User/user.schema.ts b/src/User/user.schema.ts index 0465926..1b3f0db 100644 --- a/src/User/user.schema.ts +++ b/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 }, resolve: async (payload: Message) => payload, }),