실시간 채팅: WebSocket + 메시지 저장

CREATE TABLE conversations (
  id UUID PRIMARY KEY,
  type VARCHAR NOT NULL,  -- 'direct', 'group'
  name VARCHAR,
  last_message_at TIMESTAMP,
  created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE conversation_members (
  conversation_id UUID,
  user_id UUID,
  role VARCHAR DEFAULT 'member',  -- owner, admin, member
  joined_at TIMESTAMP DEFAULT NOW(),
  last_read_at TIMESTAMP,        -- 읽음 처리
  notification_enabled BOOLEAN DEFAULT true,
  PRIMARY KEY (conversation_id, user_id)
);

CREATE TABLE messages (
  id UUID PRIMARY KEY,
  conversation_id UUID NOT NULL,
  sender_id UUID NOT NULL,
  content TEXT,
  type VARCHAR DEFAULT 'text',  -- text, image, video, file, system
  reply_to_id UUID REFERENCES messages(id),
  attachments JSONB,
  edited_at TIMESTAMP,
  deleted_at TIMESTAMP,
  created_at TIMESTAMP DEFAULT NOW()
);

CREATE INDEX ON messages(conversation_id, created_at DESC);

import { Server } from 'socket.io';
import { createAdapter } from '@socket.io/redis-adapter';

const io = new Server(httpServer, {
  cors: { origin: FRONTEND_URL },
});

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

io.use(authMiddleware);

io.on('connection', (socket) => {
  const userId = socket.data.userId;

  // 사용자가 속한 모든 대화방 자동 join
  socket.on('init', async () => {
    const memberships = await db.conversationMember.findMany({
      where: { userId },
      select: { conversationId: true },
    });
    for (const m of memberships) {
      socket.join(`conv:${m.conversationId}`);
    }
    socket.join(`user:${userId}`);

    // 온라인 상태
    await redis.sadd('online_users', userId);
    socket.broadcast.emit('user:online', userId);
  });


  // 메시지 전송
  socket.on('message:send', async (data, ack) => {
    const { conversationId, content, replyToId } = data;

    // 권한 검증
    const member = await db.conversationMember.findFirst({
      where: { conversationId, userId },
    });
    if (!member) return ack({ error: 'not_a_member' });

    const message = await db.message.create({
      data: {
        conversationId,
        senderId: userId,
        content,
        replyToId,
      },
      include: { sender: { select: { id: true, name: true, avatarUrl: true } } },
    });

    // 대화방의 last_message_at 갱신
    await db.conversation.update({
      where: { id: conversationId },
      data: { lastMessageAt: message.createdAt },
    });

    // 같은 대화방 모두에게 브로드캐스트
    io.to(`conv:${conversationId}`).emit('message:new', message);

    // 오프라인 멤버에게 푸시
    await sendOfflinePush(conversationId, message, userId);

    ack({ ok: true, message });
  });


  // 타이핑 표시
  socket.on('typing:start', ({ conversationId }) => {
    socket.to(`conv:${conversationId}`).emit('typing:start', { userId });
  });
  socket.on('typing:stop', ({ conversationId }) => {
    socket.to(`conv:${conversationId}`).emit('typing:stop', { userId });
  });


  // 읽음 처리
  socket.on('message:read', async ({ conversationId }) => {
    await db.conversationMember.update({
      where: { conversationId_userId: { conversationId, userId } },
      data: { lastReadAt: new Date() },
    });
    socket.to(`conv:${conversationId}`).emit('message:read', { userId, conversationId });
  });


  socket.on('disconnect', async () => {
    await redis.srem('online_users', userId);
    socket.broadcast.emit('user:offline', userId);
  });
});

'use client';
import { useEffect, useState } from 'react';
import { io as ioClient } from 'socket.io-client';

const socket = ioClient(API_URL, {
  auth: { token: getAccessToken() },
});


export function ChatRoom({ conversationId }: { conversationId: string }) {
  const [messages, setMessages] = useState<Message[]>([]);
  const [typing, setTyping] = useState<string[]>([]);

  useEffect(() => {
    // 초기 메시지 로드
    fetch(`/api/conversations/${conversationId}/messages?limit=50`)
      .then(r => r.json())
      .then(setMessages);

    // 새 메시지 수신
    socket.on('message:new', (msg) => {
      if (msg.conversationId === conversationId) {
        setMessages(prev => [...prev, msg]);
      }
    });

    // 타이핑
    socket.on('typing:start', ({ userId }) => {
      setTyping(prev => [...new Set([...prev, userId])]);
    });
    socket.on('typing:stop', ({ userId }) => {
      setTyping(prev => prev.filter(id => id !== userId));
    });

    // 읽음 처리
    socket.emit('message:read', { conversationId });

    return () => {
      socket.off('message:new');
      socket.off('typing:start');
      socket.off('typing:stop');
    };
  }, [conversationId]);

  const sendMessage = (content: string) => {
    socket.emit('message:send', { conversationId, content }, (res: any) => {
      if (res.error) toast.error(res.error);
    });
  };

  return (
    <>
      <MessageList messages={messages} />
      {typing.length > 0 && <TypingIndicator users={typing} />}
      <MessageInput onSend={sendMessage} onTyping={() => {
        socket.emit('typing:start', { conversationId });
        setTimeout(() => socket.emit('typing:stop', { conversationId }), 3000);
      }} />
    </>
  );
}

// 직접 INSERT 대신 큐로
socket.on('message:send', async (data, ack) => {
  const tempId = crypto.randomUUID();

  // 즉시 응답 (낙관적)
  ack({ ok: true, tempId });

  // 큐에 추가 (비동기 영속화)
  await messageQueue.add('persist', {
    tempId,
    conversationId: data.conversationId,
    senderId: userId,
    content: data.content,
  }, { jobId: tempId });
});


// Worker
const worker = new Worker('messages', async (job) => {
  const message = await db.message.create({ data: job.data });
  io.to(`conv:${job.data.conversationId}`).emit('message:persisted', {
    tempId: job.data.tempId,
    message,
  });
});

-- 사용자별·대화방별 읽지 않은 수
SELECT
  m.conversation_id,
  COUNT(*) as unread_count
FROM messages m
JOIN conversation_members cm
  ON cm.conversation_id = m.conversation_id
  AND cm.user_id = ?
WHERE m.created_at > COALESCE(cm.last_read_at, '1970-01-01')
  AND m.sender_id != ?
GROUP BY m.conversation_id;