started live chat, almost done

backend/index.ts

@@ -5,6 +5,8 @@ import { verify } from 'jsonwebtoken';
 import Message from './models/message';
 import LiveChatSession from './models/live-session-model';
 import { jwtVerify } from 'jose';
+import './models/user';
+import './models/message-vote';
 
 enum SocketEvent {
   NewMessage = 'new message',
@@ -30,7 +32,7 @@ await mongoose
 const io = new Server();
 
 const authenticateSocket = async (socket: Socket, next: any) => {
-  const token = socket.handshake.headers['authorization'];
+  const token = socket.handshake.query.token as string;
 
   if (token) {
     const { payload } = await jwtVerify(
@@ -70,8 +72,6 @@ const joinSession = async (socket: Socket, sessionId: string) => {
 };
 
 io.use(authenticateSocket).on('connection', (socket: Socket) => {
-  console.log(`User connected: ${socket.id}`);
-
   socket.on(SocketEvent.JoinSession, (sessionId: string) =>
     joinSession(socket, sessionId)
   );
@@ -86,44 +86,70 @@ enum ChangeType {
 
 Message.watch().on('change', async (change) => {
   try {
-    if (change.operationType === ChangeType.Insert) {
-      const newMessage = change.fullDocument;
+    let message: any;
+    if (['insert', 'update', 'replace'].includes(change.operationType)) {
+      const messageId =
+        change.operationType === 'insert'
+          ? change.fullDocument._id
+          : change.documentKey._id;
+
+      message = await Message.findById(messageId).populate('authorId').lean();
+    }
+    if (!message) {
+      return;
+    }
 
-      const sessionId = newMessage.liveChatSession.toString();
-      io.to(sessionId).emit(SocketEvent.NewMessage, newMessage);
-    } else if (change.operationType === ChangeType.Update) {
-      const updatedFields = change.updateDescription?.updatedFields;
-      const messageId = change.documentKey._id.toHexString(); // Convert ObjectId to string
+    message.author = message.authorId;
 
-      const updatedMessage = await Message.findById(messageId);
-      if (updatedMessage) {
-        const sessionId = updatedMessage.liveChatSession.toString();
+    switch (change.operationType) {
+      case ChangeType.Insert:
+        io.to(message.sessionId.toString()).emit(
+          SocketEvent.NewMessage,
+          message
+        );
+        break;
 
+      case ChangeType.Update:
+        const updatedFields = change.updateDescription?.updatedFields;
         if (
           updatedFields &&
           ('upVotes' in updatedFields || 'downVotes' in updatedFields)
         ) {
-          io.to(sessionId).emit(SocketEvent.MessageVoteUpdate, {
-            messageId: updatedMessage._id,
-            upVotes: updatedMessage.upVotes,
-            downVotes: updatedMessage.downVotes,
-          });
+          io.to(message.sessionId.toString()).emit(
+            SocketEvent.MessageVoteUpdate,
+            {
+              messageId: message._id,
+              upVotes: message.upVotes,
+              downVotes: message.downVotes,
+            }
+          );
         }
-      }
-    } else if (change.operationType === ChangeType.Delete) {
-      const messageId = change.documentKey._id.toHexString();
-      io.emit(SocketEvent.MessageDeleted, { messageId });
-    } else if (change.operationType === ChangeType.Replace) {
-      const newMessage = change.fullDocument;
-      if (newMessage) {
-        const sessionId = newMessage.liveChatSession.toString();
-        io.to(sessionId).emit(SocketEvent.MessageReplaced, newMessage);
-      }
+        break;
+
+      case ChangeType.Delete:
+        const messageId = change.documentKey._id.toHexString();
+        io.emit(SocketEvent.MessageDeleted, { messageId });
+        break;
+
+      case ChangeType.Replace:
+        io.to(message.sessionId.toString()).emit(
+          SocketEvent.MessageReplaced,
+          message
+        );
+        break;
     }
   } catch (error) {
     console.error('Error processing change stream event:', error);
   }
 });
-
 const PORT = process.env.PORT ? parseInt(process.env.PORT, 10) : 3000;
-io.listen(PORT);
+io.listen(PORT, {
+  cors: {
+    origin: process.env.CORS_ORIGINS?.split(',') || '*', // Allow all origins
+    methods: ['GET', 'POST'],
+    allowedHeaders: '*',
+    credentials: true,
+  },
+  path: '/socket.io',
+  transports: ['websocket'],
+});

backend/models/message-vote.ts

@@ -5,8 +5,8 @@ import { IMessage } from './message';
 export interface IMessageVote extends Document {
   _id: string;
   value: number;
-  user: IUser['_id'];
-  message: IMessage['_id'];
+  user: IUser['_id'] 
+  messageId: IMessage['_id'];
   createdAt: Date;
   updatedAt: Date;
 }
@@ -15,7 +15,7 @@ const messageVoteSchema: Schema = new Schema(
   {
     value: { type: Number, required: true },
     user: { type: Schema.Types.ObjectId, ref: 'User' },
-    message: { type: Schema.Types.ObjectId, ref: 'Message' },
+    messageId: { type: Schema.Types.ObjectId, ref: 'Message' },
     createdAt: { type: Date, default: Date.now },
     updatedAt: { type: Date, default: Date.now },
   },

backend/models/message.ts

@@ -6,9 +6,11 @@ import { IMessageVote } from './message-vote';
 export interface IMessage extends Document {
   _id: string;
   content: string;
-  author: IUser['_id'];
-  liveChatSession: ILiveChatSession['_id'];
+  authorId: IUser['_id']
+  author: IUser
+  sessionId: ILiveChatSession['_id'];
   upVotes: number;
+  totalVotes: number;
   downVotes: number;
   createdAt: Date;
   updatedAt: Date;
@@ -18,10 +20,12 @@ export interface IMessage extends Document {
 const messageSchema: Schema = new Schema(
   {
     content: { type: String, required: true },
-    author: { type: Schema.Types.ObjectId, ref: 'User' },
-    liveChatSession: { type: Schema.Types.ObjectId, ref: 'LiveChatSession' },
+
+    authorId: { type: Schema.Types.ObjectId, ref: 'User' },
+    sessionId: { type: Schema.Types.ObjectId, ref: 'LiveChatSession' },
     createdAt: { type: Date, default: Date.now },
     upVotes: { type: Number, default: 0 },
+    totalVotes: { type: Number, default: 0 },
     downVotes: { type: Number, default: 0 },
     updatedAt: { type: Date, default: Date.now },
     votes: [{ type: Schema.Types.ObjectId, ref: 'MessageVote' }],

frontend/actions/message/index.ts

@@ -18,6 +18,32 @@ import { createSafeAction } from '@/lib/create-safe-action';
 import { Session } from 'next-auth/types';
 import { auth } from '@/auth';
 import { Roles } from '@/types';
+
+export const fetchMessagesFromDatabase = async (
+  sessionId: string,
+  page: number,
+  pageSize: number
+) => {
+  const messages = await prisma.message.findMany({
+    where: {
+      sessionId: sessionId,
+    },
+    orderBy: {
+      createdAt: 'asc',
+    },
+    skip: page * pageSize,
+    take: pageSize + 1, // one extra record to check if there are more records
+    include: {
+      author: true,
+    },
+  });
+
+  const hasMore = messages.length > pageSize;
+  const result = hasMore ? messages.slice(0, -1) : messages; // remove extra record
+
+  return { result, nextPage: hasMore ? page + 1 : null };
+};
+
 const createMessageHandler = async (
   data: InputTypeCreateMessage
 ): Promise<ReturnTypeCreateMessage> => {

frontend/actions/message/schema.ts

@@ -2,7 +2,6 @@ import { z } from 'zod';
 
 export const MessageInsertSchema = z.object({
   content: z.string().min(1, 'Message content is required'),
-  authorId: z.string(),
   sessionId: z.string(),
 });
 

frontend/actions/message/types.ts

@@ -7,6 +7,7 @@ import {
   MessageUpdateSchema,
 } from './schema';
 import { Delete } from '@/types';
+import { Author } from '../question/types';
 
 export type InputTypeCreateMessage = z.infer<typeof MessageInsertSchema>;
 export type ReturnTypeCreateMessage = ActionState<
@@ -25,3 +26,7 @@ export type ReturnTypeDeleteMessage = ActionState<
   InputTypeDeleteMessage,
   Delete
 >;
+
+export interface ExtentedMessage extends Message {
+  author: Author;
+}

frontend/actions/messageVote/index.ts

@@ -0,0 +1,109 @@
+'use server';
+
+import { createSafeAction } from '@/lib/create-safe-action';
+import { MessageVoteSchema } from './schema';
+import prisma from '@/PrismaClientSingleton';
+import { auth } from '@/auth';
+import { MessageVoteUpdateType, ReturnTypeMessageVoteUpdate } from './types';
+
+const handleMessageVote = async (
+  data: MessageVoteUpdateType
+): Promise<ReturnTypeMessageVoteUpdate> => {
+  const session = await auth();
+
+  if (!session || !session.user.id) {
+    return {
+      error: 'Unauthorized',
+    };
+  }
+
+  const { value, messageId } = data;
+
+  if (![1, -1].includes(value)) {
+    return {
+      error: 'Invalid vote value',
+    };
+  }
+
+  try {
+    await prisma.$transaction(async prisma => {
+      const existingVote = await prisma.messageVote.findFirst({
+        where: {
+          userId: session.user.id,
+          messageId: messageId,
+        },
+      });
+
+      const incrementField = value === 1 ? 'upVotes' : 'downVotes';
+      const decrementField = value === 1 ? 'downVotes' : 'upVotes';
+
+      if (existingVote) {
+        if (existingVote.value === value) {
+          return {
+            error: `You have already ${
+              value === 1 ? 'upvoted' : 'downvoted'
+            } this message.`,
+          };
+        }
+
+        await prisma.messageVote.update({
+          where: {
+            id: existingVote.id,
+          },
+          data: {
+            value,
+          },
+        });
+
+        // Update message vote counts
+        const updateData = {
+          [decrementField]: { decrement: 1 },
+          [incrementField]: { increment: 1 },
+          totalVotes: { increment: value * 2 },
+        };
+
+        await prisma.message.update({
+          where: { id: messageId },
+          data: updateData,
+        });
+      } else {
+        await prisma.messageVote.create({
+          data: {
+            value,
+            userId: session.user.id!,
+            messageId: messageId,
+          },
+        });
+
+        // Update message vote counts
+        const updateData = {
+          [incrementField]: { increment: 1 },
+          totalVotes: { increment: value },
+        };
+
+        await prisma.message.update({
+          where: { id: messageId },
+          data: updateData,
+        });
+      }
+    });
+
+    const updatedMessage = await prisma.message.findUnique({
+      where: { id: messageId },
+      select: { upVotes: true, downVotes: true, totalVotes: true },
+    });
+
+    // Optionally revalidate paths as needed
+    return { data: updatedMessage };
+  } catch (error) {
+    console.error(error);
+    return {
+      error: 'Failed to process the vote.',
+    };
+  }
+};
+
+export const updateMessageVote = createSafeAction(
+  MessageVoteSchema,
+  handleMessageVote
+);

frontend/actions/messageVote/schema.ts

@@ -0,0 +1,6 @@
+import { z } from 'zod';
+
+export const MessageVoteSchema = z.object({
+  value: z.union([z.literal(-1), z.literal(1)]),
+  messageId: z.string(),
+});

frontend/actions/messageVote/types.ts

@@ -0,0 +1,9 @@
+import { z } from 'zod';
+import { ActionState } from '@/lib/create-safe-action';
+import { MessageVoteSchema } from './schema';
+
+export type MessageVoteUpdateType = z.infer<typeof MessageVoteSchema>;
+export type ReturnTypeMessageVoteUpdate = ActionState<
+  MessageVoteUpdateType,
+  { upVotes: number; downVotes: number; totalVotes: number } | null
+>;