diff --git a/src/CollaborationSession/collaborationsession.schema.ts b/src/CollaborationSession/collaborationsession.schema.ts index 1b95bae..361012e 100644 --- a/src/CollaborationSession/collaborationsession.schema.ts +++ b/src/CollaborationSession/collaborationsession.schema.ts @@ -1,10 +1,5 @@ -import { Inject, Injectable } from '@nestjs/common' -import { - Pothos, - PothosRef, - PothosSchema, - SchemaBuilderToken, -} from '@smatch-corp/nestjs-pothos' +import { Inject, Injectable, Logger } from '@nestjs/common' +import { 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: - 'Retrieve a single collaboration session by its unique identifier.', + description: '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,18 +77,15 @@ export class CollaborationSessionSchema extends PothosSchema { }) if (!scheduleDate) throw new Error('Schedule date not found') let collaborationSession: CollaborationSession | null = null - collaborationSession = - await this.prisma.collaborationSession.findUnique({ - where: { - scheduleDateId: scheduleDate.id, - }, - }) + collaborationSession = await this.prisma.collaborationSession.findUnique({ + where: { + scheduleDateId: scheduleDate.id, + }, + }) /* ---------- 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 ( - !collaborationSession.collaboratorsIds.includes(ctx.http.me?.id) - ) { + if (!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 ( - ctx.http.me.role !== Role.CENTER_MENTOR && - ctx.http.me.role !== Role.CENTER_OWNER - ) { - if (!collaborationSession) - throw new Error( - 'Mentor does not created collaboration session yet', - ) + if (ctx.http.me.role !== Role.CENTER_MENTOR && 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)) - throw new Error('User not allowed') + if (!scheduleDate.participantIds.includes(ctx.http.me.id)) 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,15 +133,14 @@ export class CollaborationSessionSchema extends PothosSchema { }) if (!chatRoom) throw new Error('Chat room not found') // create new one - const newCollaborationSession = - await this.prisma.collaborationSession.create({ - data: { - scheduleDateId: scheduleDate.id, - // assign chat room - chatRoomId: chatRoom.id, - collaboratorsIds: [ctx.http.me.id], - }, - }) + const newCollaborationSession = await this.prisma.collaborationSession.create({ + data: { + scheduleDateId: scheduleDate.id, + // assign chat room + chatRoomId: chatRoom.id, + collaboratorsIds: [ctx.http.me.id], + }, + }) // case after start time and before end time, mark as late if (now > startTime && now < endTime) { // mark as late @@ -179,8 +162,7 @@ export class CollaborationSessionSchema extends PothosSchema { collaborationSessions: t.prismaField({ type: [this.collaborationSession()], args: this.builder.generator.findManyArgs('CollaborationSession'), - description: - 'Retrieve a list of collaboration sessions with optional filtering, ordering, and pagination.', + description: '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,30 +189,23 @@ export class CollaborationSessionSchema extends PothosSchema { required: true, }), }, - description: - 'Update the active document ID for a collaboration session.', + description: '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 = - await this.prisma.collaborationSession.findUnique({ - where: { - id: args.collaborationSessionId, - }, - include: { - scheduleDate: true, - }, - }) - if (!collaborationSession) - throw new Error('Collaboration session not found') - if ( - !collaborationSession.scheduleDate.participantIds.includes( - ctx.http.me.id, - ) - ) + const collaborationSession = await this.prisma.collaborationSession.findUnique({ + where: { + id: args.collaborationSessionId, + }, + include: { + scheduleDate: true, + }, + }) + if (!collaborationSession) throw new Error('Collaboration session not found') + 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 + }, + resolve: async (payload: CollaborationSession) => payload, + }), + })) } } diff --git a/src/Cron/cron.service.ts b/src/Cron/cron.service.ts index 1471003..0bd19f1 100644 --- a/src/Cron/cron.service.ts +++ b/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 }, + }, }) } } diff --git a/src/Document/document.event.ts b/src/Document/document.event.ts index 22de7ab..06e25f7 100644 --- a/src/Document/document.event.ts +++ b/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', } diff --git a/src/Document/document.schema.ts b/src/Document/document.schema.ts index 96d6b73..f24e489 100644 --- a/src/Document/document.schema.ts +++ b/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({ - type: this.documentDeltaInput(), - required: true, - }), + data: t.arg({ 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, - eventType: DocumentEvent.REQUEST_SYNC, + totalPage, + 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 }, }), diff --git a/src/Document/document.type.ts b/src/Document/document.type.ts index 9638b67..0c5b4f7 100644 --- a/src/Document/document.type.ts +++ b/src/Document/document.type.ts @@ -4,4 +4,6 @@ export type DocumentDelta = Delta & { pageIndex: number documentId: string senderId?: string + requestSync?: boolean + totalPage?: number } diff --git a/src/Minio/minio.service.ts b/src/Minio/minio.service.ts index 28a4a59..19fab96 100644 --- a/src/Minio/minio.service.ts +++ b/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 { + return new Promise((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 diff --git a/src/OpenAI/openai.service.ts b/src/OpenAI/openai.service.ts index 5f9cc3e..0d2007e 100644 --- a/src/OpenAI/openai.service.ts +++ b/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( - mail: string, - username: string, - url: string, - ) { + async generateInvitationMailContent(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 { + 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 + } } diff --git a/src/Order/order.schema.ts b/src/Order/order.schema.ts index 6ecf142..0f41d01 100644 --- a/src/Order/order.schema.ts +++ b/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('', service.id), diff --git a/src/User/user.schema.ts b/src/User/user.schema.ts index b7dc5ec..699a4c8 100644 --- a/src/User/user.schema.ts +++ b/src/User/user.schema.ts @@ -1,22 +1,11 @@ import { Inject, Injectable, Logger } from '@nestjs/common' -import { - Pothos, - PothosRef, - PothosSchema, - SchemaBuilderToken, -} from '@smatch-corp/nestjs-pothos' +import { 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 { - ChatRoom, - Message, - MessageContextType, - MessageType, - Role, -} from '@prisma/client' +import { 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: - 'Retrieve a list of users with optional filtering, ordering, and pagination.', + description: '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,10 +368,9 @@ export class UserSchema extends PothosSchema { } const buffer = Buffer.concat(chunks) - const { id: userId, imageUrl } = - await clerkClient.users.updateUserProfileImage(id, { - file: new Blob([buffer]), - }) + const { id: userId, imageUrl } = await clerkClient.users.updateUserProfileImage(id, { + file: new Blob([buffer]), + }) await this.prisma.user.update({ where: { id: userId }, data: { @@ -502,10 +491,7 @@ export class UserSchema extends PothosSchema { }, }) // publish message - await ctx.http.pubSub.publish( - `${PubSubEvent.NEW_MESSAGE}.${message.recipientId}`, - message, - ) + await ctx.http.pubSub.publish(`${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 ( - ctx.http.me?.role !== Role.ADMIN && - ctx.http.me?.role !== Role.MODERATOR - ) { + if (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 ( - banningUser.role === Role.MODERATOR || - banningUser.role === Role.ADMIN - ) { + if (banningUser.role === Role.MODERATOR || banningUser.role === Role.ADMIN) { throw new Error(`Cannot ban moderator or admin`) } // ban user from clerk diff --git a/src/Workshop/workshop.module.ts b/src/Workshop/workshop.module.ts index 305ffcd..77c29df 100644 --- a/src/Workshop/workshop.module.ts +++ b/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], diff --git a/src/WorkshopMeetingRoom/workshopmeetingroom.module.ts b/src/WorkshopMeetingRoom/workshopmeetingroom.module.ts index 77e72ef..ae93ea7 100644 --- a/src/WorkshopMeetingRoom/workshopmeetingroom.module.ts +++ b/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],