[DEVELOP] 실시간 채팅 아키텍처 설계 — WebSocket부터 푸시 알림까지

1
2
3
4
5
6
// Socket.IO 내부 구조 (개념적)
const rooms = {
  "room-1": Set(["socketId-유저A", "socketId-유저B"]),
  "room-2": Set(["socketId-유저C", "socketId-유저D"]),
  "room-3": Set(["socketId-유저A", "socketId-유저E"]),
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
const io = new Server(3000);

io.on("connection", (socket) => {
  socket.on("join_room", (roomId: string) => {
    socket.join(roomId);
  });

  socket.on("leave_room", (roomId: string) => {
    socket.leave(roomId);
  });

  socket.on("send_message", (data) => {
    io.to(data.roomId).emit("new_message", data);
    // io.to()는 해당 Room에 join한 소켓들에게만 전달
  });
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
socket.on("send_message", async (data) => {
  // 1) DB에 영구 저장
  const saved = await db.collection("messages").insertOne({
    roomId: data.roomId,
    senderId: socket.userId,
    content: data.content,
    type: "text",
    isRead: false,
    createdAt: new Date(),
  });

  // 2) Redis에 최근 메시지 캐싱 (채팅방 목록 미리보기용)
  await redis.set(`room:${data.roomId}:lastMessage`, data.content);

  // 3) 상대방에게 실시간 전달
  io.to(data.roomId).emit("new_message", saved);
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
// hooks/useChat.ts
import { useEffect, useRef, useCallback, useState } from "react";
import { io, Socket } from "socket.io-client";
import { useQuery } from "@tanstack/react-query";

interface Message {
  id: string;
  roomId: string;
  senderId: string;
  content: string;
  type: "text" | "image" | "system";
  isRead: boolean;
  createdAt: string;
}

export const useChat = (roomId: string) => {
  const socketRef = useRef<Socket | null>(null);
  const [realtimeMessages, setRealtimeMessages] = useState<Message[]>([]);
  const [isConnected, setIsConnected] = useState(false);

  // 과거 메시지: TanStack Query로 캐싱 + 무한스크롤
  const { data: history } = useQuery({
    queryKey: ["chat", roomId, "messages"],
    queryFn: () => fetchMessages(roomId, { limit: 30 }),
  });

  useEffect(() => {
    socketRef.current = io(CHAT_SERVER_URL, {
      auth: { token: getAccessToken() },
      transports: ["websocket"],
    });

    socketRef.current.on("connect", () => setIsConnected(true));
    socketRef.current.on("disconnect", () => setIsConnected(false));

    socketRef.current.emit("join_room", roomId);

    socketRef.current.on("new_message", (message: Message) => {
      setRealtimeMessages((prev) => [...prev, message]);
    });

    return () => {
      socketRef.current?.disconnect();
    };
  }, [roomId]);

  const sendMessage = useCallback(
    (content: string) => {
      socketRef.current?.emit("send_message", { roomId, content });
    },
    [roomId],
  );

  const messages = [...(history ?? []), ...realtimeMessages];

  return { messages, sendMessage, isConnected };
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
// 온라인 상태 관리
const onlineUsers = new Map<string, string>(); // userId → socketId
const userFocusRoom = new Map<string, string>(); // userId → 현재 보고 있는 roomId

io.on("connection", (socket) => {
  onlineUsers.set(socket.userId, socket.id);

  socket.on("focus_room", (roomId: string) => {
    userFocusRoom.set(socket.userId, roomId);
  });

  socket.on("unfocus_room", () => {
    userFocusRoom.delete(socket.userId);
  });

  socket.on("disconnect", () => {
    onlineUsers.delete(socket.userId);
    userFocusRoom.delete(socket.userId);
  });
});

// 메시지 발송 시 분기
socket.on("send_message", async (data) => {
  const message = await saveMessage(data);

  for (const userId of room.participants) {
    if (userId === data.senderId) continue;

    const isOnline = onlineUsers.has(userId);
    const focusedRoom = userFocusRoom.get(userId);

    if (!isOnline) {
      // 앱 꺼져있음 → OS 푸시
      await sendPushNotification(userId, {
        title: senderName,
        body: data.content,
        data: { roomId: data.roomId },
      });
    } else if (focusedRoom !== data.roomId) {
      // 앱 열려있지만 다른 화면 → 인앱 알림
      io.to(onlineUsers.get(userId)!).emit("notification", {
        type: "new_message",
        roomId: data.roomId,
        preview: data.content,
      });
    } else {
      // 해당 채팅방에 있음 → 메시지만 전달, 알림 없음, 즉시 읽음 처리
      io.to(onlineUsers.get(userId)!).emit("new_message", message);
    }
  }
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import admin from "firebase-admin";

const sendPushNotification = async (userId: string, payload: any) => {
  const deviceToken = await getDeviceToken(userId);

  await admin.messaging().send({
    token: deviceToken,
    notification: {
      title: payload.title,
      body: payload.body,
    },
    data: payload.data,
    android: { priority: "high" },
    apns: { payload: { aps: { badge: 1, sound: "default" } } },
  });
};

1
2
3
4
5
6
7
import { createAdapter } from "@socket.io/redis-adapter";
import { createClient } from "redis";

const pubClient = createClient({ url: REDIS_URL });
const subClient = pubClient.duplicate();

io.adapter(createAdapter(pubClient, subClient));

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import { Queue } from "bullmq";

const notificationQueue = new Queue("notifications");

socket.on("send_message", async (data) => {
  const message = await saveMessage(data);
  io.to(data.roomId).emit("new_message", message);

  // 푸시 알림은 큐에 넣고 즉시 반환
  await notificationQueue.add("push", {
    recipientId: targetUserId,
    message: data.content,
  });
});

1
2
3
4
5
6
// 서버: 방마다 atomic counter (Redis INCR)
const seq = await redis.incr(`room:${roomId}:seq`);
const message = { ...data, seq, createdAt: new Date() };

// 클라이언트: seq 기준 정렬
setMessages((prev) => [...prev, message].sort((a, b) => a.seq - b.seq));

1
2
3
4
5
6
7
8
9
10
11
// 클라이언트: 디바운스로 타이핑 이벤트 전송
const handleTyping = useDebouncedCallback(() => {
  socket.emit("typing", { roomId });
}, 300);

// 서버: 해당 방 다른 유저에게 전달
socket.on("typing", (data) => {
  socket.to(data.roomId).emit("user_typing", {
    userId: socket.userId,
  });
});

1
2
3
4
5
6
7
8
9
10
11
12
13
socket.on("connect", () => {
  // 마지막으로 받은 메시지의 타임스탬프 전송
  socket.emit("sync", { lastReceivedAt: getLastMessageTime() });
});

// 서버: 해당 시점 이후 메시지를 한번에 전달
socket.on("sync", async (data) => {
  const missed = await db.messages.find({
    roomId: { $in: userRooms },
    createdAt: { $gt: data.lastReceivedAt },
  });
  socket.emit("sync_messages", missed);
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import { RateLimiterMemory } from "rate-limiter-flexible";

const rateLimiter = new RateLimiterMemory({
  points: 10, // 10개 메시지
  duration: 1, // 1초당
});

socket.on("send_message", async (data) => {
  try {
    await rateLimiter.consume(socket.userId);
    // 정상 처리
  } catch {
    socket.emit("error", {
      code: "RATE_LIMITED",
      message: "메시지를 너무 빨리 보내고 있습니다.",
    });
  }
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
// 채팅방
interface ChatRoom {
  id: string;
  productId: string; //  상품 ID
  participants: string[]; // [판매자ID, 구매자ID]
  lastMessage: string;
  lastMessageAt: Date;
  unreadCount: Record<string, number>; // { 유저ID: 안읽은수 }
  createdAt: Date;
  status: "active" | "archived" | "blocked";
}

// 메시지
interface ChatMessage {
  id: string;
  roomId: string;
  senderId: string;
  type: "text" | "image" | "location" | "product" | "system";
  content: string;
  metadata?: {
    imageUrl?: string;
    thumbnailUrl?: string;
    productId?: string;
    latitude?: number;
    longitude?: number;
  };
  seq: number;
  isRead: boolean;
  createdAt: Date;
  deletedAt?: Date;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 1. 배치 처리: 메시지를 모아서 한번에 DB에 저장
const messageBuffer: Message[] = [];
setInterval(async () => {
  if (messageBuffer.length > 0) {
    await db.collection('messages').insertMany(messageBuffer.splice(0));
  }
}, 1000);

// 2. 방 단위 전달: 전체 브로드캐스트 절대 금지
io.to(roomId).emit(...)  // ✅ 해당 방만
io.emit(...)             // ❌ 전체 유저 (절대 금지)

// 3. DB 커넥션 풀링
mongoose.connect(MONGO_URI, {
  maxPoolSize: 50,
  socketTimeoutMS: 30000,
});

// 4. 읽음 처리는 디바운스 (스크롤할 때마다 호출하면 서버 죽음)
const debouncedMarkRead = useDebouncedCallback((roomId: string) => {
  socket.emit('read_messages', roomId);
}, 500);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
// handshake에서 인증
io.use(async (socket, next) => {
  const token = socket.handshake.auth.token;
  try {
    const decoded = await verifyJWT(token);
    socket.userId = decoded.userId;
    next();
  } catch {
    next(new Error("Authentication failed"));
  }
});

// Room join 시 권한 검증
socket.on("join_room", async (roomId) => {
  const room = await db.chatRooms.findById(roomId);
  if (!room.participants.includes(socket.userId)) {
    socket.emit("error", { code: "FORBIDDEN" });
    return;
  }
  socket.join(roomId);
});