fix: handle new message for group list

README.md

@@ -111,6 +111,8 @@ pnpm --filter web test
 - [x] infinite scroll cursor pagination (messages/groups/members)
 - [x] tanstack react-query integration 
 - [x] add members
+- [x] realtime unread count 
+- [ ] leave group, transfer ownership, delete group
 - [ ] alert component
 - [ ] confirm dialog
 - [ ] delete group

server/src/modules/groups/groups.controller.ts

@@ -167,9 +167,6 @@ export const listUserGroups: RequestHandler = async (req, res, next) => {
         lastActivity: groupsWithLastMessage.lastActivity,
         id: groupsWithLastMessage.id,
         name: groupsWithLastMessage.name,
-        ownerId: groupsWithLastMessage.ownerId,
-        createdAt: groupsWithLastMessage.createdAt,
-        updatedAt: groupsWithLastMessage.updatedAt,
         unreadCount: sql<number>`COALESCE (${unreadCounts.unreadCount}, 0)`
           .mapWith(Number)
           .as('unread_count'),
@@ -222,9 +219,9 @@ export const addGroupMembers: RequestHandler = async (req, res, next) => {
 
     const io = req.app.get('io') as TypedIOServer
 
-    io.to(getGroupRoomId(req.params.groupId)).emit('newMembers', newMembers)
+    // let existing members know new member is joined
 
-    // let existing members know new member is joined in member list
+    io.to(getGroupRoomId(req.params.groupId)).emit('newMembers', newMembers)
 
     res.json(newMembers)
   } catch (error) {

server/src/modules/members/members.service.ts

@@ -60,13 +60,7 @@ export const addMembers = async (
 
   setMemberRolesForAGroup(group.id, memberRoles)
 
-  const userSockets = await getMultipleUserSockets(memberIds)
-
-  userSockets.forEach(socketId => {
-    const socket = io.sockets.sockets.get(socketId)
-    console.log('member socket', socket)
-    socket?.join(group.id.toString())
-  })
+  const userSockets = await joinMultiSocketRooms(io, memberIds, [group.id])
 
   io.to(userSockets).emit('newGroup', group)
 
@@ -80,7 +74,10 @@ export const joinMultiSocketRooms = async (
 ) => {
   const userSockets = await getMultipleUserSockets(userIds)
 
-  userSockets.forEach(socketId => {
-    io.sockets.sockets.get(socketId)?.join(groupIds.map(String))
-  })
+  for (const socketId of userSockets) {
+    const socket = io.sockets.sockets.get(socketId)
+    socket?.join(groupIds.map(String))
+  }
+
+  return userSockets
 }

server/src/modules/messages/messages.service.ts

@@ -1,4 +1,7 @@
 import { db } from '@/database'
+import { getUserSockets } from '@/redis/handlers'
+import { and, eq, isNull } from 'drizzle-orm'
+import { groups } from '../groups/groups.schema'
 import { checkPermission } from '../members/members.service'
 import { messageRecipients, messages } from './messages.schema'
 
@@ -11,6 +14,10 @@ export const insertMessage = async (
   if (!isAllowed) {
     throw new Error('createMessage: Not authorized')
   }
+  const [group] = await db
+    .select({ groupName: groups.name })
+    .from(groups)
+    .where(eq(groups.id, groupId))
   const [message] = await db
     .insert(messages)
     .values({
@@ -19,15 +26,76 @@ export const insertMessage = async (
       senderId,
     })
     .returning()
-  return message
+  return { ...message, groupName: group.groupName }
 }
 
 export const markMessageAsRead = async (
   messageId: number,
   recipientId: number,
 ) => {
+  const [message] = await db
+    .select({ senderId: messages.senderId, groupId: messages.groupId })
+    .from(messages)
+    .where(eq(messages.id, messageId))
+    .limit(1)
+  if (!message?.groupId) {
+    throw new Error('markMessageAsRead: message does not belongs to a group')
+  }
+
+  const { isAllowed } = await checkPermission(
+    message.groupId,
+    recipientId,
+    'member',
+  )
+
+  if (!isAllowed) {
+    throw new Error(
+      "markMessageAsRead: you don't have permission to mark the message as read",
+    )
+  }
+
   await db.insert(messageRecipients).values({
     messageId,
     recipientId,
   })
+
+  return getUserSockets(message.senderId)
+}
+
+export const markGroupMessagesAsRead = async (
+  groupId: number,
+  recipientId: number,
+) => {
+  const { isAllowed } = await checkPermission(groupId, recipientId, 'member')
+
+  if (!isAllowed) {
+    throw new Error(
+      "markGroupMessagesAsRead: you don't have permission to mark the message as read",
+    )
+  }
+
+  const unreadMessages = await db
+    .select({ messageId: messages.id, senderId: messages.senderId })
+    .from(messages)
+    .leftJoin(
+      messageRecipients,
+      and(
+        eq(messageRecipients.messageId, messages.id),
+        eq(messageRecipients.recipientId, recipientId),
+      ),
+    )
+    .where(
+      and(eq(messages.groupId, groupId), isNull(messageRecipients.messageId)),
+    )
+
+  if (unreadMessages.length) {
+    await db.insert(messageRecipients).values(
+      unreadMessages.map(message => ({
+        messageId: message.messageId,
+        recipientId,
+      })),
+    )
+
+    return getUserSockets(recipientId)
+  }
 }

server/src/socket/events.ts

@@ -1,24 +1,21 @@
 import { db } from '@/database'
 import { groups } from '@/modules/groups/groups.schema'
 import { members } from '@/modules/members/members.schema'
-import { checkPermission } from '@/modules/members/members.service'
-import { messageRecipients, messages } from '@/modules/messages/messages.schema'
 import {
   insertMessage,
+  markGroupMessagesAsRead,
   markMessageAsRead,
 } from '@/modules/messages/messages.service'
 import {
   addUserSocket,
-  getMultipleUserSockets,
   getTypingUsers,
-  getUserSockets,
   markUserOffline,
   markUserOnline,
   removeTypingUser,
   removeUserSocket,
   setTypingUser,
 } from '@/redis/handlers'
-import { and, eq, isNull } from 'drizzle-orm'
+import { eq } from 'drizzle-orm'
 import { config } from '../config'
 import { getGroupRoomId } from './helpers'
 import { TypedIOServer, TypedSocket } from './socket.interface'
@@ -82,75 +79,24 @@ export const registerSocketEvents = (io: TypedIOServer) => {
     })
 
     socket.on('markMessageAsRead', async messageId => {
-      const [message] = await db
-        .select({ senderId: messages.senderId, groupId: messages.groupId })
-        .from(messages)
-        .where(eq(messages.id, messageId))
-        .limit(1)
-      if (!message?.groupId) {
-        throw new Error(
-          'markMessageAsRead: message does not belongs to a group',
-        )
-      }
-
-      const { isAllowed } = await checkPermission(
-        message.groupId,
+      const messageSenderSocketIds = await markMessageAsRead(
+        messageId,
         socket.data.user.id,
-        'member',
       )
-
-      if (!isAllowed) {
-        throw new Error(
-          "markMessageAsRead: you don't have permission to mark the message as read",
-        )
-      }
-
-      await markMessageAsRead(messageId, socket.data.user.id)
-      const senderSocketIds = await getMultipleUserSockets([message.senderId])
-      io.to(senderSocketIds).emit('messageRead', messageId)
+      // let message sender know that his message is read by the current socket user
+      io.to(messageSenderSocketIds).emit('messageRead', messageId)
     })
 
     socket.on('markGroupMessagesAsRead', async groupId => {
-      const { isAllowed } = await checkPermission(
+      const socketIds = await markGroupMessagesAsRead(
         groupId,
         socket.data.user.id,
-        'member',
       )
 
-      if (!isAllowed) {
-        throw new Error(
-          "markGroupMessagesAsRead: you don't have permission to mark the message as read",
-        )
-      }
-
-      const unreadMessages = await db
-        .select({ messageId: messages.id, senderId: messages.senderId })
-        .from(messages)
-        .leftJoin(
-          messageRecipients,
-          and(
-            eq(messageRecipients.messageId, messages.id),
-            eq(messageRecipients.recipientId, socket.data.user.id),
-          ),
-        )
-        .where(
-          and(
-            eq(messages.groupId, groupId),
-            isNull(messageRecipients.messageId),
-          ),
-        )
-
-      if (unreadMessages.length) {
-        await db.insert(messageRecipients).values(
-          unreadMessages.map(message => ({
-            messageId: message.messageId,
-            recipientId: socket.data.user.id,
-          })),
-        )
-
-        const socketIds = await getUserSockets(socket.data.user.id)
-
+      if (socketIds?.length) {
+        // let the current user know that the unread messages of the group is marked as read
         io.to(socketIds).emit('groupMarkedAsRead', groupId)
+        // TODO: let the message senders know their message is read
       }
     })
 

server/src/socket/socket.interface.ts

@@ -6,7 +6,9 @@ import { Server, Socket } from 'socket.io'
 export interface ServerToClientEvents {
   userOnline: (userId: number) => void
   userOffline: (userId: number) => void
-  newMessage: (message: Message & { username: string }) => void
+  newMessage: (
+    message: Message & { username: string; groupName: string },
+  ) => void
   newMember: (member: Member & { username: string }) => void
   newMembers: (member: Member[]) => void
   newGroup: (group: Group) => void

web/src/features/group/components/UserGroupList.tsx

@@ -62,7 +62,7 @@ export const UserGroupList = () => {
       )
     }
 
-    function handleNewMessage(message: IMessage) {
+    function handleNewMessage(message: IMessage & { groupName: string }) {
       queryClient.setQueryData<IPaginatedInfiniteGroups>(
         ['userGroups', auth],
         data => {
@@ -80,20 +80,24 @@ export const UserGroupList = () => {
               }
             })
 
-            if (messageGroup) {
-              messageGroup.lastMessage = {
+            const group: IGroupWithLastMessage = {
+              id: messageGroup?.id || message.groupId,
+              name: messageGroup?.name || message.groupName,
+              lastActivity: message.createdAt,
+              unreadCount: messageGroup?.unreadCount || 0,
+              lastMessage: {
                 id: message.id,
                 content: message.content,
                 senderId: message.senderId,
-              }
-              messageGroup.lastActivity = message.createdAt
-              if (params.groupId === message.groupId.toString()) {
-                socket.emit('markMessageAsRead', message.id)
-              } else {
-                messageGroup.unreadCount++
-              }
-              draft.pages[0].data.unshift(messageGroup)
+              },
+            }
+
+            if (params.groupId === message.groupId.toString()) {
+              socket.emit('markMessageAsRead', message.id)
+            } else {
+              group.unreadCount++
             }
+            draft.pages[0].data.unshift(group)
           })
           return updatedData
         },

web/src/features/group/group.interface.ts

@@ -11,7 +11,8 @@ export interface IGroup {
   createdAt: string
 }
 
-export interface IGroupWithLastMessage extends IGroup {
+export interface IGroupWithLastMessage
+  extends Omit<IGroup, 'createdAt' | 'ownerId'> {
   lastMessage?: {
     id: number
     content: string

web/src/interfaces/socket.interface.ts

@@ -6,7 +6,7 @@ import { IMessage } from '../features/message/message.interface'
 export interface ServerToClientEvents {
   userOnline: (userId: number) => void
   userOffline: (userId: number) => void
-  newMessage: (message: IMessage) => void
+  newMessage: (message: IMessage & { groupName: string }) => void
   newMember: (member: IMember) => void
   newMembers: (member: IMember[]) => void
   newGroup: (group: IGroup) => void