Refactor CollaborationSession and Document schemas to improve code clarity and functionality. Enhance error handling in collaboration session retrieval and update logic. Introduce new request sync events in Document schema and update Minio service for document page management. Adjust cron job to use current date for future schedule date checks. Update user schema for better error handling and improve service descriptions in order schema. Remove global decorators from Workshop modules for better module encapsulation.

src/CollaborationSession/collaborationsession.schema.ts

@@ -1,10 +1,5 @@
-import { Inject, Injectable } from '@nestjs/common'
+import { Inject, Injectable, Logger } from '@nestjs/common'
-import {
+import { Pothos, PothosRef, PothosSchema, SchemaBuilderToken } from '@smatch-corp/nestjs-pothos'
-  Pothos,
-  PothosRef,
-  PothosSchema,
-  SchemaBuilderToken,
-} from '@smatch-corp/nestjs-pothos'
 import { Builder } from '../Graphql/graphql.builder'
 import { PrismaService } from '../Prisma/prisma.service'
 // import { LiveKitRoomService } from 'src/LiveKit/livekit.room.service'
@@ -71,8 +66,7 @@ export class CollaborationSessionSchema extends PothosSchema {
             required: true,
           }),
         },
-        description:
+        description: 'Retrieve a single collaboration session by its unique identifier.',
-          'Retrieve a single collaboration session by its unique identifier.',
         resolve: async (_query, _root, args, ctx, _info) => {
           if (ctx.isSubscription) throw new Error('Not allowed')
           if (!ctx.http.me) throw new Error('Cannot get your info')
@@ -83,8 +77,7 @@ export class CollaborationSessionSchema extends PothosSchema {
           })
           if (!scheduleDate) throw new Error('Schedule date not found')
           let collaborationSession: CollaborationSession | null = null
-          collaborationSession =
+          collaborationSession = await this.prisma.collaborationSession.findUnique({
-            await this.prisma.collaborationSession.findUnique({
             where: {
               scheduleDateId: scheduleDate.id,
             },
@@ -92,9 +85,7 @@ export class CollaborationSessionSchema extends PothosSchema {
           /* ---------- use case 1 : customer get collaboration session by id --------- */
           if (ctx.http.me?.role === Role.CUSTOMER && collaborationSession) {
             // if collaboratorsIds not include current user id, add it
-            if (
+            if (!collaborationSession.collaboratorsIds.includes(ctx.http.me?.id)) {
-              !collaborationSession.collaboratorsIds.includes(ctx.http.me?.id)
-            ) {
               collaborationSession.collaboratorsIds.push(ctx.http.me?.id)
               await this.prisma.collaborationSession.update({
                 where: {
@@ -108,19 +99,12 @@ export class CollaborationSessionSchema extends PothosSchema {
             return collaborationSession
           }
           /* ---------- use case 2 : center mentor get collaboration session by schedule date id --------- */
-          if (
+          if (ctx.http.me.role !== Role.CENTER_MENTOR && ctx.http.me.role !== Role.CENTER_OWNER) {
-            ctx.http.me.role !== Role.CENTER_MENTOR &&
+            if (!collaborationSession) throw new Error('Mentor does not created collaboration session yet')
-            ctx.http.me.role !== Role.CENTER_OWNER
-          ) {
-            if (!collaborationSession)
-              throw new Error(
-                'Mentor does not created collaboration session yet',
-              )
             throw new Error('User not allowed')
           }
           // check if user is participant
-          if (!scheduleDate.participantIds.includes(ctx.http.me.id))
+          if (!scheduleDate.participantIds.includes(ctx.http.me.id)) throw new Error('User not allowed')
-            throw new Error('User not allowed')
           // check if order is exist in schedule date
           if (!scheduleDate.orderId) throw new Error('Order not found')
           const order = await this.prisma.order.findUnique({
@@ -149,8 +133,7 @@ export class CollaborationSessionSchema extends PothosSchema {
             })
             if (!chatRoom) throw new Error('Chat room not found')
             // create new one
-            const newCollaborationSession =
+            const newCollaborationSession = await this.prisma.collaborationSession.create({
-              await this.prisma.collaborationSession.create({
               data: {
                 scheduleDateId: scheduleDate.id,
                 // assign chat room
@@ -179,8 +162,7 @@ export class CollaborationSessionSchema extends PothosSchema {
       collaborationSessions: t.prismaField({
         type: [this.collaborationSession()],
         args: this.builder.generator.findManyArgs('CollaborationSession'),
-        description:
+        description: 'Retrieve a list of collaboration sessions with optional filtering, ordering, and pagination.',
-          'Retrieve a list of collaboration sessions with optional filtering, ordering, and pagination.',
         resolve: async (query, _root, args, _ctx, _info) => {
           return await this.prisma.collaborationSession.findMany({
             ...query,
@@ -207,14 +189,12 @@ export class CollaborationSessionSchema extends PothosSchema {
             required: true,
           }),
         },
-        description:
+        description: 'Update the active document ID for a collaboration session.',
-          'Update the active document ID for a collaboration session.',
         resolve: async (_query, _root, args, ctx, _info) => {
           if (ctx.isSubscription) throw new Error('Not allowed')
           if (!ctx.http.me) throw new Error('Cannot get your info')
           // check permission
-          const collaborationSession =
+          const collaborationSession = await this.prisma.collaborationSession.findUnique({
-            await this.prisma.collaborationSession.findUnique({
             where: {
               id: args.collaborationSessionId,
             },
@@ -222,15 +202,10 @@ export class CollaborationSessionSchema extends PothosSchema {
               scheduleDate: true,
             },
           })
-          if (!collaborationSession)
+          if (!collaborationSession) throw new Error('Collaboration session not found')
-            throw new Error('Collaboration session not found')
+          if (!collaborationSession.scheduleDate.participantIds.includes(ctx.http.me.id))
-          if (
-            !collaborationSession.scheduleDate.participantIds.includes(
-              ctx.http.me.id,
-            )
-          )
             throw new Error('User not allowed')
-          return await this.prisma.collaborationSession.update({
+          const updatedCollaborationSession = await this.prisma.collaborationSession.update({
             where: {
               id: args.collaborationSessionId,
             },
@@ -238,8 +213,38 @@ export class CollaborationSessionSchema extends PothosSchema {
               activeDocumentId: args.activeDocumentId,
             },
           })
+          ctx.http.pubSub.publish(`collaborationSessionUpdated:${collaborationSession.id}`, updatedCollaborationSession)
+          Logger.log(`Collaboration session updated: ${updatedCollaborationSession.id}`, 'updateActiveDocumentId')
+          return updatedCollaborationSession
         },
       }),
     }))
+
+    this.builder.subscriptionFields((t) => ({
+      collaborationSessionUpdated: t.field({
+        type: this.collaborationSession(),
+        description: 'Subscribe to collaboration session updates.',
+        args: {
+          collaborationSessionId: t.arg.string({
+            description: 'The ID of the collaboration session.',
+            required: true,
+          }),
+        },
+        subscribe: async (_parent, args, ctx) => {
+          if (!ctx.isSubscription) throw new Error('Not allowed')
+          if (!ctx.websocket.me) throw new Error('Cannot get your info')
+          const collaborationSession = await this.prisma.collaborationSession.findUnique({
+            where: {
+              id: args.collaborationSessionId,
+            },
+          })
+          if (!collaborationSession) throw new Error('Collaboration session not found')
+          return ctx.websocket.pubSub.asyncIterator(
+            `collaborationSessionUpdated:${collaborationSession.id}`,
+          ) as unknown as AsyncIterable<CollaborationSession>
+        },
+        resolve: async (payload: CollaborationSession) => payload,
+      }),
+    }))
   }
 }

src/Cron/cron.service.ts

@@ -98,6 +98,7 @@ export class CronService {
   @Cron(CronExpression.EVERY_MINUTE)
   async handleRefundTicket() {
     Logger.log('Handling refund ticket', 'handleRefundTicket')
+    const now = new Date()
     // get all orders where status is REFUNDED and has schedule.dates in future
     const orders = await this.prisma.order.findMany({
       where: {
@@ -106,7 +107,7 @@ export class CronService {
           dates: {
             some: {
               end: {
-                gt: new Date(),
+                gt: now,
               },
             },
           },
@@ -130,7 +131,10 @@ export class CronService {
     // remove schedule date in future
     for (const order of orders) {
       await this.prisma.scheduleDate.deleteMany({
-        where: { id: { in: order.schedule.dates.map((d) => d.id) } },
+        where: {
+          id: { in: order.schedule.dates.map((d) => d.id) },
+          start: { gt: now },
+        },
       })
     }
   }

src/Document/document.event.ts

@@ -5,5 +5,6 @@ export enum DocumentEvent {
   PAGE_CREATED = 'document_page_created',
   PAGE_DELETED = 'document_page_deleted',
   ACTIVE_DOCUMENT_ID_CHANGED = 'document_active_document_id_changed',
-  REQUEST_SYNC = 'document_request_sync',
+  CLIENT_REQUEST_SYNC = 'document_client_request_sync',
+  SERVER_REQUEST_SYNC = 'document_server_request_sync',
 }

src/Document/document.schema.ts

@@ -66,6 +66,12 @@ export class DocumentSchema extends PothosSchema {
         senderId: t.string({
           nullable: true,
         }),
+        requestSync: t.boolean({
+          nullable: true,
+        }),
+        totalPage: t.int({
+          nullable: true,
+        }),
       }),
     })
   }
@@ -143,6 +149,28 @@ export class DocumentSchema extends PothosSchema {
           return document
         },
       }),
+      eventDocumentClientRequestSync: t.field({
+        type: this.documentDelta(),
+        args: {
+          documentId: t.arg({ type: 'String', required: true }),
+          pageIndex: t.arg({ type: 'Int', required: true }),
+        },
+        resolve: async (_, args, ctx: SchemaContext) => {
+          if (ctx.isSubscription) throw new Error('Not allowed')
+          if (!ctx.http?.me?.id) throw new Error('User not found')
+          if (!args.documentId) throw new Error('Document id not found')
+          if (!args.pageIndex) throw new Error('Page index not found')
+          const delta = await this.minio.getDocumentPage(args.documentId, args.pageIndex)
+          if (!delta) throw new Error('Delta not found')
+          return {
+            documentId: args.documentId,
+            pageIndex: args.pageIndex,
+            delta,
+            senderId: 'server',
+            eventType: DocumentEvent.CLIENT_REQUEST_SYNC,
+          }
+        },
+      }),
     }))
 
     this.builder.mutationFields((t) => ({
@@ -226,24 +254,32 @@ export class DocumentSchema extends PothosSchema {
           return args.data
         },
       }),
-      eventDocumentRequestSync: t.field({
+
+      eventDocumentServerRequestSync: t.field({
         type: this.documentDelta(),
         args: {
-          data: t.arg({
+          data: t.arg({ type: this.documentDeltaInput(), required: true }),
-            type: this.documentDeltaInput(),
-            required: true,
-          }),
         },
         resolve: async (_, args, ctx: SchemaContext) => {
           if (ctx.isSubscription) throw new Error('Not allowed')
+          const senderId = ctx.http?.me?.id
+          if (!args.data.documentId) throw new Error('Document id not found')
+          if (!senderId) throw new Error('User not found')
+          if (!args.data.pageIndex) throw new Error('Page index not found')
+
+          // save delta to minio
+          const delta = args.data.delta
+          if (!delta) throw new Error('Delta not found')
+          await this.minio.upsertDocumentPage(args.data.documentId, args.data.pageIndex, delta)
+          const totalPage = await this.minio.countDocumentPages(args.data.documentId)
           return {
             ...args.data,
-            senderId: ctx.http?.me?.id,
+            totalPage,
-            eventType: DocumentEvent.REQUEST_SYNC,
+            senderId,
+            eventType: DocumentEvent.SERVER_REQUEST_SYNC,
           }
         },
       }),
-
       updateDocument: t.prismaField({
         type: this.document(),
         args: {
@@ -356,7 +392,16 @@ export class DocumentSchema extends PothosSchema {
         },
         resolve: async (payload: DocumentDelta, _args, ctx: SchemaContext) => {
           if (!ctx.isSubscription) throw new Error('Not allowed')
-          if (payload.senderId === ctx.websocket?.me?.id) return
+          if (!payload.requestSync) {
+            // using randomize sync mechanism to avoid performance issue
+            const random = Math.random()
+            // 0.5% chance to request sync
+            if (random <= 0.005) {
+              // set requestSync to true
+              payload.requestSync = true
+              return payload
+            }
+          }
           return payload
         },
       }),

src/Document/document.type.ts

@@ -4,4 +4,6 @@ export type DocumentDelta = Delta & {
   pageIndex: number
   documentId: string
   senderId?: string
+  requestSync?: boolean
+  totalPage?: number
 }

src/Minio/minio.service.ts

@@ -1,10 +1,11 @@
 import { Inject, Injectable, Logger } from '@nestjs/common'
 import { ConfigService } from '@nestjs/config'
 import { FileUpload } from 'graphql-upload/processRequest.js'
-import { Client } from 'minio'
+import { Client, BucketItem } from 'minio'
 import { MINIO_CONNECTION } from 'nestjs-minio'
 import { DateTimeUtils } from 'src/common/utils/datetime.utils'
 import { v4 as uuidv4 } from 'uuid'
+import Delta from 'quill-delta'
 @Injectable()
 export class MinioService {
   constructor(
@@ -85,13 +86,30 @@ export class MinioService {
     return await this.minioClient.putObject(this.configService.get('BUCKET_NAME') ?? 'epess', `documents/${id}`, '')
   }
 
-  async upsertDocumentFolder(id: string, page: string) {
+  async upsertDocumentPage(id: string, page: number, delta: Delta) {
     return await this.minioClient.putObject(
       this.configService.get('BUCKET_NAME') ?? 'epess',
       `documents/${id}/${page}`,
-      '',
+      JSON.stringify(delta),
     )
   }
+  async getDocumentPage(id: string, page: number) {
+    const delta = (await this.minioClient.getObject(
+      this.configService.get('BUCKET_NAME') ?? 'epess',
+      `documents/${id}/${page}`,
+    )) as unknown as Delta
+    return delta
+  }
+  async countDocumentPages(id: string): Promise<number> {
+    return new Promise<number>((resolve, reject) => {
+      const stream = this.minioClient.listObjects(this.configService.get('BUCKET_NAME') ?? 'epess', `documents/${id}`)
+      const items: BucketItem[] = []
+
+      stream.on('data', (item) => items.push(item))
+      stream.on('end', () => resolve(items.length))
+      stream.on('error', (err) => reject(err))
+    })
+  }
   // export document to docx format by get all pages and convert to docx
   async exportDocument(id: string) {
     // get all pages

src/OpenAI/openai.service.ts

@@ -1,15 +1,12 @@
 import { Injectable } from '@nestjs/common'
 import { OpenAI } from 'openai'
+import { DocumentDelta } from 'src/Document/document.type'
 
 @Injectable()
 export class OpenaiService {
   constructor(private openai: OpenAI) {}
 
-  async generateInvitationMailContent(
+  async generateInvitationMailContent(mail: string, username: string, url: string) {
-    mail: string,
-    username: string,
-    url: string,
-  ) {
     const prompt = `
     give me mail content for invitation to a workshop to EPESS and replace {{ mail }} with ${mail}, {{ username }} with ${username} and {{ url }} with ${url}
     `
@@ -21,4 +18,17 @@ export class OpenaiService {
 
     return response.choices[0].message.content
   }
+
+  async documentSuggestEditDelta(documentDelta: DocumentDelta): Promise<DocumentDelta> {
+    const prompt = `
+    give me suggestion for edit document delta to EPESS and replace {{ documentDelta }} with ${documentDelta}
+    `
+
+    const response = await this.openai.chat.completions.create({
+      model: 'gpt-4o',
+      messages: [{ role: 'user', content: prompt }],
+    })
+
+    return response.choices[0].message.content as unknown as DocumentDelta
+  }
 }

src/Order/order.schema.ts

@@ -204,7 +204,7 @@ export class OrderSchema extends PothosSchema {
           const paymentData = await this.payosService.createPayment({
             orderCode: paymentCode,
             amount: service.price,
-            description: service.name,
+            description: service.name.split(' ').slice(0, 20).join(' '),
             buyerName: ctx.http.me?.name ?? '',
             buyerEmail: ctx.http.me?.email ?? '',
             returnUrl: `${process.env.PAYOS_RETURN_URL}`.replace('<serviceId>', service.id),

src/User/user.schema.ts

@@ -1,22 +1,11 @@
 import { Inject, Injectable, Logger } from '@nestjs/common'
-import {
+import { Pothos, PothosRef, PothosSchema, SchemaBuilderToken } from '@smatch-corp/nestjs-pothos'
-  Pothos,
-  PothosRef,
-  PothosSchema,
-  SchemaBuilderToken,
-} from '@smatch-corp/nestjs-pothos'
 import { Builder, SchemaContext } from '../Graphql/graphql.builder'
 import { PrismaService } from '../Prisma/prisma.service'
 import { clerkClient } from '@clerk/express'
 import { MailService } from '../Mail/mail.service'
 import { MessageSchema } from 'src/Message/message.schema'
-import {
+import { ChatRoom, Message, MessageContextType, MessageType, Role } from '@prisma/client'
-  ChatRoom,
-  Message,
-  MessageContextType,
-  MessageType,
-  Role,
-} from '@prisma/client'
 import { PubSubEvent } from 'src/common/pubsub/pubsub-event'
 import { DateTimeUtils } from 'src/common/utils/datetime.utils'
 import { ChatroomSchema } from '../ChatRoom/chatroom.schema'
@@ -245,8 +234,7 @@ export class UserSchema extends PothosSchema {
       }),
 
       users: t.prismaField({
-        description:
+        description: 'Retrieve a list of users with optional filtering, ordering, and pagination.',
-          'Retrieve a list of users with optional filtering, ordering, and pagination.',
         type: [this.user()],
         args: this.builder.generator.findManyArgs('User'),
         resolve: async (query, _root, args) => {
@@ -265,10 +253,12 @@ export class UserSchema extends PothosSchema {
         type: this.user(),
         args: this.builder.generator.findUniqueArgs('User'),
         resolve: async (query, _root, args) => {
-          return await this.prisma.user.findUniqueOrThrow({
+          const user = await this.prisma.user.findUnique({
             ...query,
             where: args.where,
           })
+          if (!user) throw new Error('User not found')
+          return user
         },
       }),
       userBySession: t.prismaField({
@@ -378,8 +368,7 @@ export class UserSchema extends PothosSchema {
               }
 
               const buffer = Buffer.concat(chunks)
-              const { id: userId, imageUrl } =
+              const { id: userId, imageUrl } = await clerkClient.users.updateUserProfileImage(id, {
-                await clerkClient.users.updateUserProfileImage(id, {
                 file: new Blob([buffer]),
               })
               await this.prisma.user.update({
@@ -502,10 +491,7 @@ export class UserSchema extends PothosSchema {
             },
           })
           // publish message
-          await ctx.http.pubSub.publish(
+          await ctx.http.pubSub.publish(`${PubSubEvent.NEW_MESSAGE}.${message.recipientId}`, message)
-            `${PubSubEvent.NEW_MESSAGE}.${message.recipientId}`,
-            message,
-          )
           return message
         },
       }),
@@ -518,10 +504,7 @@ export class UserSchema extends PothosSchema {
           if (ctx.isSubscription) {
             throw new Error('Not allowed')
           }
-          if (
+          if (ctx.http.me?.role !== Role.ADMIN && ctx.http.me?.role !== Role.MODERATOR) {
-            ctx.http.me?.role !== Role.ADMIN &&
-            ctx.http.me?.role !== Role.MODERATOR
-          ) {
             throw new Error(`Only admin or moderator can ban user`)
           }
           if (args.userId === ctx.http.me?.id) {
@@ -535,10 +518,7 @@ export class UserSchema extends PothosSchema {
             throw new Error(`User ${args.userId} not found`)
           }
           // if banning user is moderator or admin, throw error
-          if (
+          if (banningUser.role === Role.MODERATOR || banningUser.role === Role.ADMIN) {
-            banningUser.role === Role.MODERATOR ||
-            banningUser.role === Role.ADMIN
-          ) {
             throw new Error(`Cannot ban moderator or admin`)
           }
           // ban user from clerk

src/Workshop/workshop.module.ts

@@ -1,7 +1,6 @@
 import { Global, Module } from '@nestjs/common'
 import { WorkshopSchema } from './workshop.schema'
 
-@Global()
 @Module({
   providers: [WorkshopSchema],
   exports: [WorkshopSchema],

src/WorkshopMeetingRoom/workshopmeetingroom.module.ts

@@ -1,7 +1,6 @@
 import { Module, Global } from '@nestjs/common'
 import { WorkshopMeetingRoomSchema } from './workshopmeetingroom.schema'
 
-@Global()
 @Module({
   providers: [WorkshopMeetingRoomSchema],
   exports: [WorkshopMeetingRoomSchema],