Omni-channel: Go, MongoDB, RabbitMQ và WebSocket

Omni-channel là hệ thống xử lý hội thoại theo sự kiện, với API Go/Gin, adapter kênh, worker RabbitMQ và CMS Next.js. Thay vì chỉ mô tả một inbox hợp nhất, bài này phân tích cách webhook trở thành message, cách hệ thống giữ thứ tự xử lý theo hội thoại, vị trí của idempotency và cách event tới đúng người dùng qua WebSocket. Dữ liệu chat cốt lõi nằm trong MongoDB; source còn có kết nối MySQL cho phần nghiệp vụ quan hệ, không dùng TypeORM hoặc Lucid trong backend Go.

Ranh giới API, adapter, worker và giao diện

Thành phần	Trách nhiệm	Không nên đồng nhất với
Go/Gin API	HTTP routes, auth, query hội thoại và nhận webhook	Adapter riêng cho từng nền tảng
Channel adapters	Chuyển đổi payload và giao tiếp API/session của kênh	Quyền xem dữ liệu trong CMS
RabbitMQ workers	Dispatcher, partition consumer, outbound và retry	Database lưu lịch sử lâu dài
MongoDB	Users/RBAC, conversations, messages và event records	Cache Redis ngắn hạn
Redis	Idempotency gate và phối hợp realtime	Bằng chứng đã lưu message bền vững
WebSocket hub	Phân phối cập nhật có xét context người nhận	Quyền truy cập mặc định mọi hội thoại
Next.js CMS	Inbox, khách hàng, kênh và quản trị	Nguồn chân lý của trạng thái gửi tin

Kênh / webhook → Gin gateway → inbound exchange
  → dispatcher → conversation partition → processor → MongoDB
  → event / Redis → WebSocket hub → CMS
CMS gửi tin → API → outbound queue → adapter → nền tảng kênh

Backend có chế độ chạy worker riêng và đường khởi động worker trong tiến trình ứng dụng tùy cấu hình. Điều này hỗ trợ tách tải xử lý khỏi HTTP khi triển khai, nhưng cũng tạo yêu cầu vận hành: không khởi tạo thêm consumer theo cảm tính mà chưa hiểu số partition, chế độ đang dùng và phạm vi khóa/idempotency.

Cấu trúc source và hợp đồng event

api/
  routes/       # Gin route composition
  controllers/  # HTTP actions
  channel/      # normalized channel contracts
  queue/        # topology, publisher, payloads
  workers/      # dispatcher, conversation và outbound processing
  models/       # BSON/JSON data structures
  database/     # MongoDB indexes, MySQL connection
  store/        # Redis idempotency
  socket/       # hub và clients
cms/            # Next.js inbox và channel administration

InboundEventPayload có event_id, idempotency_key, channel, channel_account_id, conversation_id, raw_payload, các mốc thời gian, attempt và trace_id. Payload có thể mang normalized message để phần xử lý sau không phải hiểu lại mọi định dạng kênh. Việc giữ cả event time lẫn thời điểm gateway nhận và enqueue giúp phân biệt trễ từ nguồn, trễ truyền và trễ hàng đợi.

OutboundEventPayload tham chiếu message_id và outbound_event_id, kèm khóa idempotency, lần thử, queued_at và expires_at. Một công việc gửi có thể hết hạn trước khi được thực thi; source có hàm kiểm tra cửa sổ này. Thiết kế event contract như vậy cho phép theo dõi vòng đời xử lý, nhưng không thể suy từ việc có trace_id rằng hệ thống đã triển khai đầy đủ distributed tracing hoặc dashboard quan sát.

MongoDB: định danh hội thoại, unique index và lịch sử message

Collection conversations có unique index ghép (channel_account_id, external_conversation_id). Hai tài khoản kênh có thể nhận cùng một external ID, nên chỉ dùng external ID làm khóa toàn hệ thống sẽ tạo va chạm sai. Các index khác phục vụ assigned_user_id/status, assigned_team_id/status và last_message_at cho cách inbox lọc dữ liệu.

Messages có sparse unique index trên channel_message_key và các index theo conversation_id kết hợp event_time, created_at hoặc _id. Inbound/outbound events có unique index trên idempotency_key. Các index này thuộc lớp lưu trữ bền vững và hỗ trợ chống trùng theo khóa đã chọn; hiệu quả còn phụ thuộc cách xây khóa, điều kiện query và dữ liệu thiếu trường. Cơ chế unique compound index được mô tả trong tài liệu MongoDB.

Source model dùng struct tags BSON để lưu document và JSON để trả qua API. PasswordHash bị loại khỏi JSON bằng tag riêng. MongoDB không làm mất nhu cầu về schema contract: API, worker và CMS vẫn phải thống nhất trạng thái, ID và quan hệ tham chiếu. Kết nối MySQL trong cùng backend dùng database/sql với pool có giới hạn; sự tồn tại của nó không có nghĩa lịch sử chat được lưu đồng thời ở cả hai database.

Partition theo hội thoại và ý nghĩa thứ tự xử lý

Dispatcher ánh xạ conversation ID vào partition bằng FNV-1a 32 bit rồi lấy modulo số partition. Với cùng số partition và cùng ID, các event của hội thoại đi vào cùng nhóm xử lý. Runner tạo consumer cho từng conversation queue, với mức prefetch được truyền là 1 trong các luồng đã đối chiếu. Mục tiêu là hạn chế xử lý chồng chéo trong từng luồng, thay vì chạy mọi tin độc lập rồi hy vọng timestamp tự sửa được thứ tự.

// Rút gọn đúng thuật toán trong workers/routing.go
func PartitionForConversation(id string, partitions int) int {
    if partitions <= 1 { return 0 }
    h := fnv.New32a()
    _, _ = h.Write([]byte(id))
    return int(h.Sum32() % uint32(partitions))
}

Partition không bảo đảm thứ tự toàn cục giữa mọi khách hàng. Nó cũng không tự bảo đảm thứ tự khi triển khai nhiều consumer cạnh tranh cho cùng queue hoặc khi retry quay lại sau một event mới. Đổi số partition làm thay đổi phép ánh xạ, vì vậy đây là thay đổi cần xem xét cùng backlog và chiến lược chuyển tải. Muốn đánh giá đúng phải quan sát topology và số consumer đang chạy, không chỉ nhìn hàm hash.

Ack, retry, DLQ và idempotency nhiều lớp

RabbitMQ topology có inbound, outbound, retry và dead-letter exchanges. Các retry queue dùng TTL theo các mốc 5 giây, 30 giây và 5 phút rồi đưa công việc về queue xử lý. Runner thực hiện Ack sau các nhánh xử lý phù hợp, Nack với hoặc không requeue tùy loại lỗi. Về khái niệm, consumer acknowledgement xác nhận phía xử lý với broker; nó khác publisher confirm ở chiều publisher gửi vào broker, như RabbitMQ giải thích.

Redis IdempotencyGate dùng SetNX với TTL và timeout để xác định lần thấy đầu tiên của một key. Redis hỗ trợ điều kiện chỉ ghi khi key chưa tồn tại, nhưng TTL khiến bộ nhớ chống trùng có cửa sổ thời gian, không phải sổ cái vĩnh viễn. Unique index và event records trong MongoDB vẫn có vai trò riêng. Bản thân việc request vượt gate cũng không chứng minh xử lý cuối đã thành công.

Một cửa sổ lỗi cần đánh giá là đã ghi gate nhưng chưa publish thành công sang bước kế tiếp. Retry, Ack/Nack và trạng thái event phải được xem cùng nhau để biết hệ thống có thể bỏ qua hay chạy lặp công việc nào. Bài viết không gắn nhãn exactly-once cho luồng này. Thiết kế sử dụng retry và chống trùng nhằm kiểm soát hệ quả của delivery lặp, trong khi những điểm giao giữa Redis, MongoDB và RabbitMQ vẫn là ranh giới không có một transaction SQL chung.

DLQ giữ lại công việc vượt chính sách thử lại để có thể điều tra. Đưa tin vào DLQ không phải đã giải quyết sự cố. Trace ID, account ID, attempt và trạng thái message cần đủ để xác định nên sửa dữ liệu, cấu hình adapter hay quyền ở nền tảng kênh trước khi chạy lại.

WebSocket, phân quyền và trạng thái gửi tin

Hub được khởi tạo với MongoDB, RBAC checker và Redis client. Source có logic xét người được gán hội thoại và membership nhóm khi xác định người nhận event. Điều này quan trọng vì endpoint đọc có ACL chưa đủ: nếu WebSocket broadcast mọi message tới mọi client, dữ liệu vẫn có thể bị lộ qua luồng realtime.

CMS tiếp tục dùng REST để truy vấn danh sách, lịch sử và thao tác quản trị; WebSocket bổ sung cập nhật theo event. Sau reconnect, lịch sử bền vững ở database là điểm đối chiếu cho những cập nhật UI có thể bỏ lỡ. Không nên dùng trạng thái một socket đang mở để kết luận kênh bên ngoài còn session hoặc người nhận đã đọc tin.

Cần phân biệt yêu cầu gửi đã được API tiếp nhận, job đã enqueue, adapter đã chuyển đi và xác nhận phía kênh. Mỗi mức trả lời một câu hỏi khác nhau. Ảnh preview phía dưới cố ý cho thấy kênh chưa kết nối; đó là một trạng thái giao diện thật, không được dùng như bằng chứng tích hợp production đã hoạt động với mọi nền tảng.

Quan sát và kiểm thử các ranh giới hệ thống

Gửi lặp cùng inbound key và kiểm tra số message thực sự được lưu.
Làm gián đoạn publish sau bước Redis gate để kiểm tra nhánh phục hồi và khả năng mất xử lý.
Đẩy nhiều event cùng conversation và event thuộc conversation khác để kiểm tra phạm vi ordering.
Giả lập lỗi adapter tạm thời, lỗi vĩnh viễn và job hết hạn để đối chiếu retry/DLQ.
Kiểm tra người dùng không thuộc nhóm qua cả REST và WebSocket, kể cả sau đổi phân công.

Các unit test về routing, outbound status và conversation processor hiện diện trong source là cơ sở để định hướng kiểm thử. Chúng không thay thế kiểm thử tích hợp nhiều tiến trình hoặc phép đo độ trễ. Các chỉ số phù hợp để bổ sung gồm tuổi công việc cũ nhất, retry count, DLQ depth và khoảng cách giữa event_time với thời điểm xử lý, thay vì một con số throughput không có bối cảnh.

Giao diện ứng dụng và dữ liệu minh họa

Các ảnh dưới đây được chụp trực tiếp từ FE hoặc trang quản trị của ứng dụng. Với các màn hình chạy trong phiên preview, dữ liệu mẫu được tách khỏi hệ thống vận hành; bố cục và thành phần giao diện do chính source của dự án render.

Không gian inbox đa kênh

Quản lý tài khoản kênh

Tài liệu kỹ thuật đối chiếu

Xem thêm các dự án khác hoặc trao đổi về kiến trúc và phạm vi triển khai.