Personal Chats

Personal chats are direct conversations between two users. This article explains how personal (one-to-one) chats work in the system.

Database Schema for Personal Chats

Schema diagram for personal chats

        
erDiagram
    USERS ||--o{ CHATS : "sender_settings"
    USERS ||--o{ CHATS : "receiver_settings"
    USERS ||--o{ MESSAGES : "author"
    USERS ||--o{ MESSAGE_RECIPIENTS : "recipient"
    MESSAGES ||--o{ MESSAGE_RECIPIENTS : "distributed_to"
    USERS {
        int id PK
        string username
    }
    CHATS {
        int sender_id PK, FK
        int receiver_id PK, FK
        timestamp cleared_at
        boolean muted
        boolean archived
        boolean pinned
    }
    MESSAGES {
        int id PK
        string content
        int sender_id FK
        int channel_id FK
        timestamp created_at
        timestamp updated_at
    }
    MESSAGE_RECIPIENTS {
        int id PK
        int message_id FK
        int receiver_id FK
        enum status
        timestamp created_at
        timestamp updated_at
    }

Personal Chat Flow

Personal chat message flow

        
sequenceDiagram
    participant Client as Client A
    participant API as Backend API
    participant RabbitMQ as RabbitMQ
    participant Worker as Chat Worker
    participant DB as PostgreSQL
    participant Socket as Socket Server
    participant Client2 as Client B
    Client->>Client: Create temp message (in memory)
    Client->>API: HTTP POST /api/messages/send/personal
    API->>RabbitMQ: Publish "personal.message" event
    API-->>Client: Return 200 OK
    RabbitMQ->>Worker: Dequeue "personal.message" event
    Worker->>DB: Insert message + message_recipients (status=SENT)
    Worker->>RabbitMQ: Publish "personal.message.persisted" event
    RabbitMQ->>Socket: Route to recipient's user ID
    Socket-->>Client: Notify "sent" via socket
    Client->>Client: Replace temp message with actual (Single tick)
    Socket->>Client2: Forward message to recipient

Implementation

The API server provides HTTP endpoints for message sending, and immediately offloads the task to RabbitMQ, so that it remains unblocked.

const sendMessage = async (receiverId: number, content: string, hash: string) => { await $api('/api/messages/send/personal', { method: 'POST', body: { receiverId, content, hash, }, }) }

func (s *MessageService) SendPersonalMessage(ctx context.Context, userID int64, dto *SendPersonalMessageDTO) error { // Publish to RabbitMQ for worker to process message := RabbitMQMessage{ Type: "MESSAGE_SEND", Payload: map[string]any{ "senderId": userID, "receiverId": dto.ReceiverID, "content": dto.Content, "hash": dto.Hash, }, } return s.rabbitmqService.PublishToIncoming("personal.message", message) }

The chat worker handles the "heavy" database operations and notifies the system when the message is persisted.

consumer.on('personal.incoming.message', async event => { const { hash, content, senderId, receiverId } = event.payload const message = await messageRepository.save({ content, sender: { id: senderId }, channel: null, }) await messageRecipientRepository.save({ message: { id: message.id }, receiver: { id: receiverId }, status: MessageStatus.SENT, }) // Publish event for socket server to deliver to sender await rabbitmq.publishToOutgoing(senderId.toString(), { event: 'PERSONAL_MESSAGE_SENT', data: { hash, messageId: message.id, status: MessageStatus.SENT, createdAt: message.createdAt, }, }) // Publish event for socket server to deliver to recipient await rabbitmq.publishToOutgoing(receiverId.toString(), { event: 'PERSONAL_MESSAGE_RECEIVE', data: { messageId: message.id, content, senderId, createdAt: message.createdAt, }, }) })

The socket servers keep consuming events from their dedicated queues (discussed in Scalable WebSockets with RabbitMQ). When the "sent" (for sender) and "receive" (for recipient) events arrive from OUTGOING_EXCHANGE, the server forwards them to their respective clients. This finalizes the lifecycle for both the sender and the receiver.

// Setup RabbitMQ consumer for server queue (Socket.IO server in Node.js) consumeFromServerQueue((message, ack) => { try { const { event, userId, data } = message // Find socket ID for the user const socketId = chatsStore.getClient(userId) if (socketId) { io.to(socketId).emit(event, data) } ack() } catch (error) { console.error('Error processing message from RabbitMQ:', error) ack() } })