Chatroom

Frontend

Syncing Data

• To update chat history on local DB on background

useSyncMessages.ts

import { useEffect, useRef } from "react";

import { AppState } from "react-native";

import { syncMessages } from "@/lib/features/chat/conversation-init";
import { useAuth } from "@/lib/store/authStore";
import { logger } from "@/lib/utils/logger";

import type { AppStateStatus } from "react-native";

const SYNC_INTERVAL_MS = 60_000; // 1 minute

export const useSyncMessages = () => {
  const { user } = useAuth();
  const appStateRef = useRef<AppStateStatus>(AppState.currentState);

  useEffect(() => {
    if (!user?.id) return;

    logger.debug("syncMessages");
    // 1. Sync on initial load
    syncMessages();

    // 2. Sync on app foreground
    const subscription = AppState.addEventListener("change", (nextAppState) => {
      if (
        appStateRef.current.match(/inactive|background/) &&
        nextAppState === "active"
      ) {
        syncMessages();
      }
      appStateRef.current = nextAppState;
    });

    // 3. Periodic sync every 1 minute
    const intervalId = setInterval(syncMessages, SYNC_INTERVAL_MS);

    return () => {
      subscription.remove();
      clearInterval(intervalId);
    };
  }, [user?.id]);
};

conversation-init.ts

import { chatControllerGetUserConversations } from "@/lib/api/chat/chat";
import {
  getConfig,
  getFirstMessage,
  getIsConversationFirstMessageReached,
  getOverallLastMessage,
  setConfig,
  setConversationMetadata,
} from "@/lib/database/v2/queries";
import { logger } from "@/lib/utils/logger";

import {
  getAndStoreRemoteMessagesAfter,
  getAndStoreRemoteMessagesBefore,
} from "./remote-messages";

export const fetchMorePastRemoteMessages = async (conversationId: string) => {
  // `isFirstMessageReached = true` Happens when already fetched all messages before
  const isFirstMessageReached =
    await getIsConversationFirstMessageReached(conversationId);
  if (isFirstMessageReached) return;
  // Get the first message
  const firstMessage = await getFirstMessage(conversationId);

  // If there is no first message, there is no message at all
  if (!firstMessage) {
    // ensure `isFirstMessageReached = true`
    await setConversationMetadata({
      id: conversationId,
      firstMessageReached: true,
    });
    return;
  }
  await getAndStoreRemoteMessagesBefore({
    conversationId,
    beforeDate: new Date(firstMessage.createdAt!),
    limit: 100,
  });
};

export const initConversations = async () => {
  const config = await getConfig();
  const now = new Date();
  if (config?.isInitialized) return;
  // Get all conversations
  const conversations = await chatControllerGetUserConversations();

  // Get 100 messages for each conversation
  for (const conversation of conversations) {
    await getAndStoreRemoteMessagesBefore({
      conversationId: conversation.id,
      beforeDate: new Date(),
      limit: 100,
    });
  }
  await setConfig({
    isInitialized: true,
    lastSynced: now,
  });
};

export const syncMessages = async () => {
  logger.debug("[DEBUG] syncMessages");
  const lastRemoteMessage = await getOverallLastMessage();
  let lastMessageId = lastRemoteMessage?.id;
  logger.debug("[DEBUG] initial lastMessageId", lastMessageId);
  const now = new Date();
  let count = 0;
  while (lastMessageId && count < 3) {
    count++;
    if (!lastMessageId) break;
    logger.debug("[DEBUG] lastMessageId", lastMessageId);
    const { data: messages } = await getAndStoreRemoteMessagesAfter({
      limit: 100,
      afterMessageId: lastMessageId,
    });
    logger.debug("[DEBUG] messages", messages);
    const lastItem = messages[messages.length - 1];
    lastMessageId = lastItem ? parseInt(lastItem.id) : undefined;
    logger.debug("[DEBUG] new lastMessageId", lastMessageId);
  }
  await setConfig({
    lastSynced: now,
  });
};

Fetching Chat History

• Keep fetching chat history from local DB / remote database

• Display the chat history on UI

import { useEffect, useMemo } from "react";

import { FlashList } from "@shopify/flash-list";
import { useQueryClient } from "@tanstack/react-query";
import { View } from "react-native";
import Animated, {
  KeyboardState,
  useAnimatedKeyboard,
  useAnimatedStyle,
} from "react-native-reanimated";
import { useSafeAreaInsets } from "react-native-safe-area-context";
import { createStyleSheet, useStyles } from "react-native-unistyles";

import type { Conversation } from "@/lib/api/famchest.schemas";
import { Typography } from "@/lib/components/Typography";
import { useMessages, useMessageWithSeparators } from "@/lib/hooks/useMessages";
import type { SendMessageParams } from "@/lib/mutations/useSendMessage";
import { useConversationStore } from "@/lib/store/conversationStore";

import { ConversationHeader } from "./ConversationHeader";
import { MessageInput } from "./MessageInput";
import { MessageItem } from "./MessageItem";

export function ConversationScreen({
  profileId,
  conversation,
  onSendMessage,
}: {
  profileId?: string;
  conversation?: Conversation | null;
  onSendMessage: (
    params: Omit<SendMessageParams, "conversationId">,
  ) => Promise<void>;
}) {
  // fetch chat messages api
  const { messages, fetchNextPage, hasNextPage, isLoading } = useMessages(
    conversation?.id || "",
  );

  const withSeparators = useMessageWithSeparators(messages);
  
  // fetch more messages for scrolling
  const onLoadMore = () => {
    if (hasNextPage && !isLoading) {
      fetchNextPage();
    }
  };

  const isGroup = conversation?.type === "group";
  const { enterConversation, leaveConversation } = useConversationStore();
  const queryClient = useQueryClient();

  // Register and unregister the active conversation when component mounts/unmounts
  useEffect(() => {
    if (conversation?.id) {
      enterConversation(conversation.id, queryClient);

      // Clean up when component unmounts or conversation changes
      return () => {
        leaveConversation();
      };
    }
  }, [conversation?.id, enterConversation, leaveConversation, queryClient]);

  const { styles, theme } = useStyles(styleSheet);
  const handleLoadMore = () => {
    if (hasNextPage && !isLoading) {
      onLoadMore();
    }
  };

  const { height, state } = useAnimatedKeyboard();
  const insets = useSafeAreaInsets();
  const wrapperStyle = useAnimatedStyle(() => {
    return {
      backgroundColor: theme.colors.black5,
      paddingBottom:
        state.value === KeyboardState.OPEN ||
          state.value === KeyboardState.OPENING
          ? height.value
          : height.value + insets.bottom,
    };
  });

  const stickyHeaderIndices = useMemo(() => {
    const indices: number[] = [];
    for (let i = 0; i < withSeparators.length; i++) {
      if (withSeparators[i]?.__type === "separator") {
        indices.push(i);
      }
    }
    return indices;
  }, [withSeparators]);

  return (
    <Animated.View style={[wrapperStyle, { flex: 1 }]}>
      {/* TODO: show profile detail */}
      <ConversationHeader
        profileId={profileId}
        conversation={conversation}
        isGroup={isGroup}
      />

      <View style={styles.chatBackground}>
        <FlashList
          maintainVisibleContentPosition={{
            startRenderingFromBottom: true,
            autoscrollToBottomThreshold: 0.5,
            animateAutoScrollToBottom: true,
          }}
          stickyHeaderConfig={{
            hideRelatedCell: true,
          }}
          stickyHeaderIndices={stickyHeaderIndices}
          data={withSeparators}
          keyExtractor={(item) =>
            item.__type === "message" ? item.uniqueId : item.text
          }
          renderItem={({ item }) => {
            switch (item.__type) {
              case "message":
                return <MessageItem item={item} isGroup={isGroup} />;
              case "separator":
                return (
                  <View style={styles.separatorContainer}>
                    <View style={styles.pill}>
                      <Typography
                        fontStyle="caption3"
                        fontWeight="500"
                        color="black5"
                      >
                        {item.text}
                      </Typography>
                    </View>
                  </View>
                );
              default:
                return null;
            }
          }}
          onEndReached={handleLoadMore}
          onEndReachedThreshold={0.5}
        />
      </View>
      <MessageInput onSendMessage={onSendMessage} />
    </Animated.View>
  );
}

const styleSheet = createStyleSheet((theme) => ({
  chatBackground: {
    backgroundColor: theme.colors.yellow3,
    flex: 1,
  },
  separatorContainer: {
    padding: theme.spacing.sm,
    alignItems: "center",
  },
  pill: {
    backgroundColor: theme.colors.black3,
    paddingHorizontal: theme.spacing.sm,
    paddingVertical: theme.spacing.xs,
    borderRadius: 999,
  },
}));

useMessages.ts

import { useMemo } from "react";

import { useInfiniteQuery } from "@tanstack/react-query";
import { and, desc, eq, gte } from "drizzle-orm";
import { useLiveQuery } from "drizzle-orm/expo-sqlite";

import type { LocalMessageV2 } from "@/db/schema";
import { messagesTable } from "@/db/schema";
import { db } from "@/lib/database/v2/db";
import { getMessages } from "@/lib/features/chat";
import { useConversationStore } from "@/lib/store/conversationStore";
import { formatMessageDate } from "@/lib/utils/date-time";

export type Separator = {
  __type: "separator";
  text: string;
};

export type Message = LocalMessageV2 & {
  __type: "message";
};

export type UseMessages = {
  messages: LocalMessageV2[];
  fetchNextPage: () => void;
  hasNextPage: boolean;
  isLoading: boolean;
};
export const useMessages = (conversationId: string): UseMessages => {
  const { enteredAt } = useConversationStore();
  const {
    data: oldMessagesPages,
    fetchNextPage,
    hasNextPage,
    isLoading,
  } = useInfiniteQuery({
    queryKey: ["messages", conversationId, enteredAt?.toISOString()],
    queryFn: async ({ pageParam: beforeDate }) => {
      if (!conversationId || !beforeDate) {
        return [];
      }

      const oldMessages = await getMessages(conversationId, beforeDate);
      return oldMessages;
    },
    getNextPageParam: (lastPage) => {
      // use the last message of the last page
      return lastPage[lastPage.length - 1]?.createdAt;
    },
    initialPageParam: enteredAt,
    enabled: !!enteredAt,
  });

  const { data: newMessages } = useLiveQuery(
    db
      .select()
      .from(messagesTable)
      .where(
        and(
          eq(messagesTable.conversationId, conversationId!),
          enteredAt ? gte(messagesTable.createdAt, enteredAt) : undefined,
        ),
      )
      .orderBy(desc(messagesTable.createdAt)),
    [enteredAt],
  );

  const oldMessages = oldMessagesPages?.pages.flat() || [];
  const messages = [...newMessages, ...oldMessages];
  return { messages, fetchNextPage, hasNextPage, isLoading };
};

export const useLastMessage = (conversationId: string) => {
  return useLiveQuery(
    db.query.messages.findFirst({
      where: eq(messagesTable.conversationId, conversationId),
      orderBy: desc(messagesTable.createdAt),
    }),
    [conversationId],
  );
};

export const useMessageWithSeparators = (messages: LocalMessageV2[]) => {
  const withSeparators = useMemo(() => {
    const results = [] as (Message | Separator)[];
    for (let i = 0; i < messages.length; i++) {
      // messages are in descending time order, so we can check the previous message
      const currentMessage = messages[i]!;

      if (!currentMessage.createdAt) {
        continue;
      }
      const currentMessageDate = formatMessageDate(currentMessage.createdAt!);

      results.push({ __type: "message", ...currentMessage });

      if (i === messages.length - 1) {
        results.push({ __type: "separator", text: currentMessageDate });
        continue;
      }

      const previousMessage = messages[i + 1];

      if (!previousMessage?.createdAt) {
        continue;
      }

      const previousMessageDate = formatMessageDate(previousMessage.createdAt!);

      if (currentMessageDate !== previousMessageDate) {
        results.push({ __type: "separator", text: currentMessageDate });
      }
    }
    return results.reverse();
  }, [messages]);

  return withSeparators;
};

Web socket

• To listen to the message event for frontend handling

MessageSocketProvider.tsx

import { createContext, useContext, useEffect, useState } from "react";

import { useQueryClient } from "@tanstack/react-query";
import { router } from "expo-router";
import { io } from "socket.io-client";
import { Toast } from "toastify-react-native";

import {
  getChatControllerGetLastReadQueryKey,
  getChatControllerGetUserConversationsQueryKey,
} from "@/lib/api/chat/chat";
import {
  type ConversationRead,
  type Message,
  MessageType,
} from "@/lib/api/famchest.schemas";
import { handleApiNewMessages } from "@/lib/features/chat";
import { queryClient } from "@/lib/providers/AppQueryClientProvider";
import { useAuth } from "@/lib/store/authStore";
import { useConversationStore } from "@/lib/store/conversationStore";
import { logger } from "@/lib/utils/logger";

import type { Socket } from "socket.io-client";

type MessageSocketContextType = {
  socket: Socket | null;
};

const MessageSocketContext = createContext<MessageSocketContextType>({
  socket: null,
});
/**
 * Show toast notification for incoming message
 * Only shows if:
 * - Message is not from the current user
 * - User is not in the conversation the message belongs to
 */
function showMessageToast(message: Message) {
  const { user } = useAuth.getState();
  const { activeConversationId } = useConversationStore.getState();
  const currentUserId = user?.id;

  // Don't show toast if message is from current user
  if (message.senderProfileId === currentUserId) {
    return;
  }

  // Don't show toast if user is already in the conversation
  if (message.conversationId === activeConversationId) {
    return;
  }

  // Show the message toast
  Toast.show({
    type: "message" as any, // Custom toast type defined in toast config
    position: "top",
    visibilityTime: 3000,
    props: { message },
    onPress: () => {
      if (message.conversationId) {
        router.push(`/home/chat/conversation/${message.conversationId}`);
      }
    },
  });
}

async function handleMessage(message: Message) {
  // if message type is create-group, refetch the conversations
  if (message.type === MessageType["create-group"]) {
    await queryClient.refetchQueries({
      queryKey: getChatControllerGetUserConversationsQueryKey(),
    });
  }
  handleApiNewMessages([message]);

  // Show toast notification for incoming messages
  showMessageToast(message);
  if (
    useConversationStore.getState().activeConversationId ===
    message.conversationId
  ) {
    useConversationStore
      .getState()
      .markConversationAsRead(message.conversationId, queryClient);
  } else {
    // add the unread count to the query data
    queryClient.setQueryData(
      getChatControllerGetLastReadQueryKey(message.conversationId as string),
      (old: ConversationRead) => ({
        ...old,
        unreadCount: Number(old?.unreadCount ?? 0) + 1,
      }),
    );
  }
}

export const MessageSocketProvider = ({
  children,
}: {
  children: React.ReactNode;
}) => {
  const queryClient = useQueryClient();
  const { session } = useAuth();
  const [socket, setSocket] = useState<Socket | null>(null);

  useEffect(() => {
    if (session?.access_token) {
      const socket = io(`${process.env.EXPO_PUBLIC_SOCKET_URL}/message`, {
        transports: ["websocket"],
        auth: {
          token: session?.access_token,
        },
      });
      function init() {
        socket.emit("joinConversations");
      }

      socket.on("connect_error", (error) => {
        logger.error("Error connecting to socket", error);
      });

      setSocket(socket);

      socket.on("connect", init);
      socket.on("message:created", handleMessage);
      return () => {
        socket?.off("connect", init);
        socket?.off("message:created", handleMessage);
        socket.disconnect();
      };
    }
  }, [queryClient, session?.access_token]);

  return (
    <MessageSocketContext.Provider value={{ socket }}>
      {children}
    </MessageSocketContext.Provider>
  );
};

export const useMessageSocket = () => {
  return useContext(MessageSocketContext);
};

Send Message

• To send the message through backend

• Handle different type of message, e.g: image, video, text

• Store the sent message to local db

import { useMutation } from "@tanstack/react-query";
import * as Crypto from "expo-crypto";
import { File } from "expo-file-system";
import * as VideoThumbnails from "expo-video-thumbnails";

import type { LocalMessageV2 } from "@/db/schema";
import { chatControllerSendMessage } from "@/lib/api/chat/chat";
import type { AttachmentMeta } from "@/lib/api/famchest.schemas";
import { handleApiNewMessages, handleSentMessage } from "@/lib/features/chat";
import { useMyProfile } from "@/lib/queries/useMyProfile";
import { uploadFile } from "@/lib/supabase/calls";
import { getFileNameFromUri } from "@/lib/utils/getFilenameFromUri";
import { logger } from "@/lib/utils/logger";

import type { DocumentPickerAsset } from "expo-document-picker";
import type { ImagePickerAsset } from "expo-image-picker";

export type AudioAsset = {
  uri: string;
  mimeType?: string;
  size?: number;
  duration?: number; // Optional: if we can get duration
};

export type SendMessageParams = {
  conversationId: string;
  message: string;
} & (
  | {
      type: "text";
      file?: undefined;
    }
  | {
      type: "image" | "video";
      file: ImagePickerAsset;
    }
  | {
      type: "file";
      file: DocumentPickerAsset;
    }
  | {
      type: "recording";
      file: AudioAsset;
    }
);

export const useSendMessage = () => {
  const { data: profile } = useMyProfile();

  return useMutation({
    mutationKey: ["send_message"],
    mutationFn: async (params: SendMessageParams) => {
      const uniqueId = Crypto.randomUUID();

      const { type, file } = params;

      let attachmentPath: string | undefined;
      let attachmentMeta: AttachmentMeta = {};
      if (type !== "text" && file) {
        const originalFilename =
          ("name" in file && file?.name) || getFileNameFromUri(file?.uri);
        const fileInfo = new File(file.uri).info();
        if (!fileInfo.exists || !fileInfo.uri) {
          throw new Error("File not found");
        }
        if (type === "video") {
          const { uri: thumbnailUrl } = await VideoThumbnails.getThumbnailAsync(
            file.uri,
            { time: 0 },
          );

          const { path: thumbnailPath } = await uploadFile({
            uri: thumbnailUrl,
            uploadOptions: {
              contentType: file.mimeType,
              metadata: {
                originalFilename,
              },
            },
          });
          attachmentMeta.thumbnailPath = thumbnailPath;
        }
        const uploadResult = await uploadFile({
          uri: file.uri,
          uploadOptions: {
            contentType: file.mimeType,
            metadata: {
              originalFilename,
            },
          },
        });
        attachmentPath = uploadResult.path;
        attachmentMeta = {
          ...attachmentMeta,
          mimeType: file.mimeType,
          size: fileInfo.exists ? fileInfo.size : 0,
          originalFilename,
          width: "width" in file ? file.width : undefined,
          height: "height" in file ? file.height : undefined,
          audioLength:
            type === "recording" && params.file && "duration" in params.file
              ? params.file.duration
              : undefined,
        };
      }

      // Create a local message with pending status
      const localMessage: LocalMessageV2 = {
        uniqueId,
        conversationId: params.conversationId,
        type,
        content: params.message,
        attachmentPath: attachmentPath || null,
        senderProfileId: profile?.id || "",
        createdAt: new Date(),
        updatedAt: new Date(),
        isSynced: false,
        isFailed: false,
        id: null,
        attachmentMeta,
      };

      // Store the local message in SQLite

      try {
        handleSentMessage(localMessage);
        // Send message to server

        const serverMessage = await chatControllerSendMessage(
          params.conversationId,
          {
            content: params.message,
            uniqueId,
            type,
            attachmentPath,
            attachmentMeta,
          },
        );
        handleApiNewMessages([serverMessage]);
        return serverMessage;
      } catch (error) {
        // Handle error but keep the message in local state
        // You may want to mark it as failed to be retried later
        logger.error("Failed to send message:", error);

        throw error;
      }
    },
    onError: (error) => {
      logger.error("Failed to send message:", error);
    },
  });
};

Backend

Send Message

• Store the message to database

• Emit the message for web socket

• Handle join room/conversation by listening event

chat.service.ts

import {
  EntityManager,
  FilterQuery,
  OrderDefinition,
} from "@mikro-orm/postgresql";
import {
  BadRequestException,
  ForbiddenException,
  Injectable,
  NotFoundException,
} from "@nestjs/common";
import { EventEmitter2 } from "@nestjs/event-emitter";

import { CreateMessageDto } from "@/chat/dto/create-message.dto";
import { InitGroupConversationDto } from "@/chat/dto/init-group-conversation.dto";
import { InitPrivateConversationDto } from "@/chat/dto/init-private-conversation.dto";
import { MessagesPageQueryDto } from "@/chat/dto/messages-page-query.dto";
import { UpdateConversationDto } from "@/chat/dto/update-conversation.dto";
import {
  Conversation,
  ConversationType,
} from "@/chat/entities/conversation.entity";
import { Message, MessageType } from "@/chat/entities/message.entity";
import { ConversationProfileRepository } from "@/chat/repositories/conversation-profile.repository";
import { ConversationReadRepository } from "@/chat/repositories/conversation-read.repository";
import { ConversationRepository } from "@/chat/repositories/conversation.repository";
import { MessageRepository } from "@/chat/repositories/message.repository";
import { MessageSentEvent } from "@/common/events/message-sent.event";
import { PageDto } from "@/common/page.dto";
import { Profile } from "@/profile/entities/profile.entity";
import { ProfileRepository } from "@/profile/repositories/profile.repository";

@Injectable()
export class ChatService {
  constructor(
    private em: EntityManager,
    private conversationRepository: ConversationRepository,
    private conversationProfileRepository: ConversationProfileRepository,
    private conversationReadRepository: ConversationReadRepository,
    private messageRepository: MessageRepository,
    private profileRepository: ProfileRepository,
    private eventEmitter: EventEmitter2,
  ) {}
  async createMessage(
    profile: Profile,
    conversationId: string,
    createMessageDto: CreateMessageDto,
  ): Promise<Message> {
    // Get the conversation
    const conversation = await this.conversationRepository.findOne({
      id: conversationId,
    });

    if (!conversation) {
      throw new NotFoundException("Conversation not found");
    }

    // Check if user is a member of this conversation
    const conversationProfile =
      await this.conversationProfileRepository.findOne({
        conversation: { id: conversationId },
        profile: { id: profile.id },
      });

    if (!conversationProfile) {
      throw new ForbiddenException("User is not a member of this conversation");
    }

    // Validate attachment path and metadata for non-text messages
    if (createMessageDto.type !== MessageType.TEXT) {
      if (!createMessageDto.attachmentPath) {
        throw new BadRequestException(
          "Attachment path is required for non-text messages",
        );
      }
      if (!createMessageDto.attachmentMeta) {
        throw new BadRequestException(
          "Attachment metadata is required for non-text messages",
        );
      }
    }

    // Create and persist the message
    const message = this.messageRepository.create({
      conversation: conversationId,
      senderProfile: profile.id,
      content: createMessageDto.content,
      type: createMessageDto.type,
      uniqueId: createMessageDto.uniqueId,
      attachmentPath: createMessageDto.attachmentPath,
      attachmentMeta: createMessageDto.attachmentMeta,
    });

    await this.em.persistAndFlush(message);

    // Update the user's read timestamp for this conversation
    await this.markAsRead(profile, conversationId);

    this.eventEmitter.emit(
      MessageSentEvent.EVENT_NAME,
      new MessageSentEvent(message),
    );
    return message;
  }
}

message.gateway.ts

import { UnauthorizedException } from "@nestjs/common";
import { JwtService } from "@nestjs/jwt";
import {
  ConnectedSocket,
  GatewayMetadata,
  OnGatewayConnection,
  OnGatewayDisconnect,
  OnGatewayInit,
  SubscribeMessage,
  WebSocketGateway,
  WebSocketServer,
} from "@nestjs/websockets";
import { Server, Socket } from "socket.io";

import { MessageService } from "@/message/message.service";

export type SocketWithAuth = Socket & {
  profileId: string;
  name: string;
};

@WebSocketGateway({
  cors: {
    origin: "*",
  },
  namespace: "message",
} satisfies GatewayMetadata)
export class MessageGateway
  implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect
{
  constructor(
    private readonly messageService: MessageService,
    private readonly jwtService: JwtService,
  ) {}

  @WebSocketServer()
  server: Server;

  afterInit(server: Server) {
    this.messageService.server = server;
    server.use(async (socket: SocketWithAuth, next) => {
      const jwt = socket.handshake.auth.token;
      if (!jwt) {
        console.log(
          "failed to obtain token when initializing web socket connection",
        );
        socket.disconnect();
        return;
      }
      try {
        const decoded = this.jwtService.verify(jwt);
        const profile = await this.messageService.getProfile(decoded.sub);
        if (!profile) {
          console.log(
            "profile not found when initializing web socket connection",
          );
          return next(new UnauthorizedException());
        }
        socket.profileId = profile.id;
        socket.name = profile.name || "";
      } catch (error) {
        console.log("error initializing web socket connection", error);
        socket.disconnect();
        return;
      }
      return next();
    });
  }

  async handleConnection(@ConnectedSocket() client: SocketWithAuth) {
    console.log(`User ${client.name} connected`);
    client.join(client.profileId);
  }

  // Event to let client join conversations
  @SubscribeMessage("joinConversations")
  async joinConversations(@ConnectedSocket() client: SocketWithAuth) {
    console.log("joinConversations", client.profileId);
    await this.messageService.syncConversationRooms(client.profileId, client);
    return true;
  }

  handleDisconnect(client: SocketWithAuth) {
    console.log(`User ${client.name} disconnected`);
  }
}

message.service.ts

import { EntityManager } from "@mikro-orm/postgresql";
import { Injectable } from "@nestjs/common";
import { OnEvent } from "@nestjs/event-emitter";
import { Server, Socket } from "socket.io";

import { ConversationProfile } from "@/chat/entities/conversation-profile.entity";
import { MessageSentEvent } from "@/common/events/message-sent.event";
import { Profile } from "@/profile/entities/profile.entity";

@Injectable()
export class MessageService {
  public server: Server;

  constructor(private readonly em: EntityManager) {}

  async getProfile(id: string) {
    return this.em.fork().findOne(Profile, {
      id,
    });
  }

  async getConversationIds(profileId: string) {
    const conversationProfiles = await this.em
      .fork()
      .find(ConversationProfile, {
        profile: {
          $in: [profileId],
        },
      });
    const conversationIds = conversationProfiles.map(
      (conversationProfile) => conversationProfile.conversation.id as string,
    );
    return conversationIds;
  }

  async syncConversationRooms(profileId: string, client: Socket) {
    const conversationIds = await this.getConversationIds(profileId);
    const rooms = conversationIds.map((id) => `conversation:${id}`);

    client.join(rooms);

    client.rooms.forEach((room) => {
      if (room.startsWith("conversation:") && !rooms.includes(room)) {
        console.log("leaving", room);
        client.leave(room);
      }
    });
  }

  @OnEvent(MessageSentEvent.EVENT_NAME)
  async handleMessageSent(event: MessageSentEvent) {
    const { message } = event;
    const { conversation } = message;
    const room = `conversation:${conversation.id}`;
    this.server
      .to(room)
      // .except(message.senderProfile.id)
      .emit("message:created", message);
  }
}