Copy page

WebSocket

Koneksi WebSocket

Koneksi WebSocket

ws-server adalah binary terpisah (crates/ws-server) yang mengekspos WebSocket endpoint untuk mendorong event real-time ke browser. Ia berlangganan empat Redis pub/sub channel dan meneruskan pesan ke koneksi klien yang sesuai berdasarkan peran dan assignment.

Halaman ini menjelaskan cara klien membangun koneksi, cara autentikasi bekerja, dan siklus hidup koneksi dari handshake sampai tutup.

Endpoint

Item	Nilai
Protokol	ws:// (dev) atau wss:// (produksi)
Path	/ws
Host	ws-server di WS_SERVER_PORT (default 3002)
Auth	Query string ?token=<JWT>

URL lengkap dev: ws://localhost:3002/ws?token=eyJhbGciOiJIUzI1NiIs...

Kenapa token lewat query string?

WebSocket API browser (new WebSocket(url, protocols)) tidak dapat menambahkan header HTTP custom. Karena itu kami tidak dapat memakai Authorization: Bearer seperti pada REST API. Tiga alternatif yang dipertimbangkan:

1. Query string (pilihan kami) — sederhana, cukup untuk single-tenant OmniStream. Risiko: URL tercatat di log reverse proxy.
2. Sec-WebSocket-Protocol — workaround umum, tetapi bergantung pada parsing non-standar di server.
3. Cookie-only — butuh same-origin yang kadang bentrok dengan deployment cross-subdomain.

Untuk memitigasi risiko logging, konfigurasi reverse proxy Anda (nginx, Caddy) agar tidak menulis query string untuk path /ws ke access log. Pemakaian JWT jangka-pendek (24 jam default) juga mengurangi dampak jika token bocor.

Kode koneksi dari browser

Frontend SvelteKit OmniStream menggunakan pola berikut:

Code
async function connectWs(): Promise<WebSocket> {
  // 1. Ambil token segar dari api-gateway.
  //    Cookie httpOnly access_token sudah terkirim otomatis.
  const res = await fetch("/api/auth/token", { credentials: "include" });
  const { token } = await res.json();

  // 2. Konek ke ws-server dengan token di query string.
  const ws = new WebSocket(`ws://localhost:3002/ws?token=${token}`);

  ws.addEventListener("open", () => console.log("ws connected"));
  ws.addEventListener("message", (e) => handleMessage(JSON.parse(e.data)));
  ws.addEventListener("close", (e) => scheduleReconnect(e.code));
  ws.addEventListener("error", (e) => console.error("ws error", e));

  return ws;
}

GET /api/auth/token mengembalikan token baru berdasarkan cookie yang sudah ada — alasannya ada di Cookie httpOnly.

Alur autentikasi di server

Ketika request masuk, ws-server menjalankan handler di crates/ws-server/src/handler.rs:

1. Upgrade handshake — Axum membaca query WsQuery { token } dari URL.
2. Validasi token — JwtValidator::validate(&query.token) memakai JWT_SECRET yang sama dengan api-gateway. Gagal validasi (kedaluwarsa, signature salah) menghasilkan 401 Unauthorized dan koneksi ditolak sebelum upgrade.
3. Decode claims — token valid menghasilkan AgentClaims { sub, email, role, exp, iat }. sub adalah UUID agent yang dipakai sebagai key di peta koneksi aktif.
4. Register ke registry — koneksi ditambahkan ke shared HashMap<Uuid, Vec<WsSender>>. Satu agent bisa punya beberapa koneksi (mis. dua tab browser); semua menerima event yang sama.
5. Broadcast presence (jika ini koneksi pertama) — jika ini satu-satunya koneksi aktif untuk agent tersebut, server mempublikasikan AgentPresenceEvent { agent_id, status: "online" } ke Redis channel agent_presence.

Kode referensi: crates/ws-server/src/handler.rs::handle_socket.

Siklus hidup koneksi

Code
[Browser]                                       [ws-server]
   |                                                  |
   |-- GET /ws?token=... -------------------------->  |
   |                                           (validate JWT)
   |<----------------- 101 Switching Protocols -------|
   |                                         (register in registry)
   |                                   (first conn? publish presence online)
   |                                                  |
   |<-- {"channel":"conversation_updates", "data":...}|   (pushes from subscriber)
   |<-- {"channel":"typing_indicators", "data":...}   |
   |                                                  |
   |-- {"type":"ping"} ----------------------------->  |  (client keepalive)
   |-- {"type":"typing", "conversation_id":"..."} -->  |  (client-originated)
   |                                                  |
   |-- close(1000) --------------------------------->  |
   |                                         (remove from registry)
   |                                 (last conn? publish presence offline)
   |                                                  |

Heartbeat dan ping

Klien mengirim {"type":"ping"} setiap 30 detik untuk menjaga koneksi melewati proxy yang agresif menutup koneksi idle. Server hanya melakukan no-op — tidak ada response. Jika klien tidak mengirim apa pun dan proxy menutup koneksi, klien akan mencoba reconnect via event close.

Lihat Client messages untuk detail formatnya.

Reconnect dan backoff

Klien sebaiknya reconnect dengan backoff eksponensial untuk menghindari badai request ketika ws-server restart:

Code
let reconnectDelay = 1000;

function scheduleReconnect(closeCode: number) {
  // 1008 = Policy Violation (umumnya token invalid/expired)
  if (closeCode === 1008) {
    // Ambil token baru lalu reconnect
    reconnectDelay = 1000;
  } else {
    reconnectDelay = Math.min(reconnectDelay * 2, 30000); // cap 30s
  }

  setTimeout(connectWs, reconnectDelay);
}

Reset delay ke 1000 ms ketika open event muncul.

Kode close yang umum

Kode	Arti	Aksi klien
1000	Normal closure (logout, tab ditutup)	Jangan reconnect
1001	Going away (server shutdown)	Reconnect setelah backoff
1006	Abnormal closure (network drop)	Reconnect dengan backoff
1008	Policy violation (token invalid/expired)	Ambil token baru, lalu reconnect
1011	Internal error	Reconnect dengan backoff

Kode 1008 dipakai ws-server saat JWT gagal validasi setelah upgrade. Sebelum upgrade, penolakan muncul sebagai 401 HTTP biasa.

Multi-koneksi per agent

Satu agent dapat membuka beberapa tab dan semuanya dianggap koneksi aktif. Registry menyimpan Vec<WsSender> per agent_id. Konsekuensinya:

• Semua tab menerima event yang sama.
• Presence online dikirim hanya saat koneksi pertama muncul.
• Presence offline dikirim hanya saat koneksi terakhir tutup.
• Ini mencegah flapping saat user berganti tab.

Peran dan routing event

Event yang Anda terima bergantung pada peran JWT:

• Admin / Supervisor — menerima event conversation_updates untuk semua percakapan.
• Agent — menerima event hanya untuk percakapan yang assigned_agent_id-nya adalah agent tersebut, ditambah event ConversationCreated untuk percakapan baru yang belum di-assign (supaya bisa mengklaim).

Lihat Event routing untuk matriks lengkap per channel.

Contoh lengkap: klien CLI dengan websocat

Code
# 1. Ambil token
TOKEN=$(curl -sS -X POST http://localhost:3000/api/auth/login \
  -H "Content-Type: application/json" \
  -d '{"email":"admin@omnistream.com","password":"admin123"}' \
  | jq -r '.token')

# 2. Konek dan print semua event
websocat "ws://localhost:3002/ws?token=$TOKEN"

Setiap event akan muncul sebagai satu baris JSON.

Troubleshooting

• 401 Unauthorized saat upgrade — token tidak valid atau kedaluwarsa. Ambil token baru via GET /api/auth/token.
• Koneksi tiba-tiba ditutup dengan code 1008 — JWT kedaluwarsa di tengah sesi. Klien harus refetch token dan reconnect.
• Tidak menerima event meskipun koneksi open — periksa peran Anda. Agent hanya menerima event untuk percakapan yang di-assign. Lihat Event routing.
• Koneksi berputar-putar reconnecting setiap 30 detik — reverse proxy menutup koneksi idle. Pastikan klien mengirim {"type":"ping"} atau konfigurasi proxy_read_timeout nginx ke nilai yang lebih tinggi.
• Duplikasi event di frontend — Anda mungkin membuka beberapa tab. Setiap tab menerima event secara terpisah; frontend harus idempoten berdasarkan message_id atau event_id.

File terkait

• Handler upgrade dan registry: crates/ws-server/src/handler.rs
• Subscriber Redis: crates/ws-server/src/subscriber.rs
• Main binary: crates/ws-server/src/main.rs
• JWT validator: crates/omni-common/src/jwt.rs