Research & Learning · feature explainer

Cơ chế Pusher cho chat realtime trong `threads-api-v2`

TL;DR — Luồng đúng là: POST /message-groups/:publicId/messages lưu message vào DB trước, sau đó trigger Pusher event message:new lên channel private-chat-{groupPublicId}. FE subscribe channel này để nhận message realtime, còn lịch sử vẫn lấy từ API để chống miss event.

Flow gửi message end-to-end

Bạn đã có bước tạo tin nhắn rồi, nên phần cần thêm chỉ là đoạn phát event sau khi transaction commit thành công.

1 · REST nhận request gửi tin message-group.controller.ts

Client gọi POST /message-groups/:publicId/messages với body { content }. API đã xác thực JWT và lấy senderId từ req.user.sub.

2 · Lưu DB bằng transaction message-group-facade.service.ts

Service validate member của group, insert bản ghi Message, update lastMessageAt, rồi trả về payload message.

3 · Trigger Pusher event message-group-facade.service.ts

Sau khi transaction thành công, trigger event lên channel của cuộc trò chuyện. Nếu Pusher lỗi thì chỉ log, không nên rollback request đã gửi tin thành công trong DB.

4 · FE nhận event và cập nhật UI web/mobile client

FE subscribe đúng channel của group, bind event message:new, append message vào state nếu chưa tồn tại. Khi mở lại phòng chat vẫn gọi API GET /message-groups/:publicId/messages để đồng bộ.

Channel naming khuyến nghị

Chat group: private-chat-{groupPublicId} — chỉ member group được subscribe.
User riêng: private-user-{userId} — dùng cho notification cá nhân.
Event chat mới: message:new.
Event đã xem (phase sau): message:seen.

★

Không dùng public channel cho chat production vì ai biết channel name cũng subscribe được. Dùng private channel + endpoint auth để kiểm tra member.

Snippet triển khai nhanh (BE + FE)

Bộ snippet dưới đây bám theo cấu trúc module hiện tại trong repo của bạn.

// src/modules/message-group/service/message-group-facade.service.ts
import { pusherService } from "@/modules/pusher/service/pusher.service";

async sendMessage(groupPublicId: string, senderId: string, content: string) {
  const messageGroup = await this.findMessageGroupOrThrow(groupPublicId);
  await messageMemberService.assertMemberOrThrow(messageGroup.id, senderId);

  const messagePayload = await transactionService.doInTransaction(async (tx) => {
    const message = await messageService.createMessage(
      { messageGroupId: messageGroup.id, senderId, content },
      tx,
    );

    await messageGroupService.updateLastMessageAt(messageGroup.id, new Date(), tx);
    return mapMessageResponse(message);
  });

  try {
    await pusherService.trigger(
      `private-chat-${messageGroup.publicId}`,
      "message:new",
      {
        groupPublicId: messageGroup.publicId,
        message: messagePayload,
      },
    );
  } catch (error) {
    baseLogger.error("Failed to trigger chat realtime event: %o", error);
  }

  return messagePayload;
}

// route: POST /pusher/auth (authorization middleware bắt buộc)
// body: { socket_id, channel_name }

if (channel_name.startsWith("private-chat-")) {
  const groupPublicId = channel_name.replace("private-chat-", "");
  const group = await messageGroupService.findByPublicId(groupPublicId);
  if (!group) throw new ForbiddenException("Channel is not accessible");

  await messageMemberService.assertMemberOrThrow(group.id, userId);
  return res.json(await pusherService.authorizeChannel(socket_id, channel_name));
}

if (channel_name === `private-user-${userId}`) {
  return res.json(await pusherService.authorizeChannel(socket_id, channel_name));
}

throw new ForbiddenException("Unsupported private channel");

// FE (pusher-js)
import Pusher from "pusher-js";

const pusher = new Pusher(import.meta.env.VITE_PUSHER_KEY, {
  wsHost: import.meta.env.VITE_PUSHER_HOST,
  wsPort: Number(import.meta.env.VITE_PUSHER_PORT),
  forceTLS: false,
  enabledTransports: ["ws", "wss"],
  channelAuthorization: {
    endpoint: `${import.meta.env.VITE_API_URL}/pusher/auth`,
    headers: {
      Authorization: `Bearer ${accessToken}`,
    },
  },
});

const channel = pusher.subscribe(`private-chat-${groupPublicId}`);
channel.bind("message:new", ({ message }) => {
  upsertMessage(message);
});

Checklist test nhanh

Mở 2 tab với 2 user khác nhau, cùng vào 1 message group.
Tab A gửi message qua API hiện tại.
Tab B phải nhận message:new gần như ngay lập tức.
Tắt mạng tab B, gửi tin từ tab A, bật lại mạng và load lại history từ API để đảm bảo không mất message.
Thử user không thuộc group subscribe private-chat-{groupPublicId} phải bị từ chối ở endpoint auth.

FAQ

Nếu Pusher lỗi thì có nên fail API gửi tin nhắn không?: Không nên. Message đã lưu DB là nguồn sự thật. Realtime chỉ là kênh phát nhanh cho UX.
Vì sao vẫn cần API get messages nếu đã có realtime?: Realtime có thể miss event khi user offline/reconnect. API history giúp đồng bộ trạng thái chuẩn.
Nên đặt event tên new-message hay message:new?: Chọn một convention và giữ nhất quán toàn hệ thống. Với repo này, pattern resource:action khá dễ mở rộng.