Web Socket #

npm install ws
npm install --save-dev @types/ws typescript ts-node

// tsconfig.json — minimal configuration for a WebSocket project
{
  "compilerOptions": {
    "target": "ES2022",
    "module": "CommonJS",
    "strict": true,
    "esModuleInterop": true,
    "outDir": "./dist",
    "rootDir": "./src"
  }
}

// src/types/ws-messages.ts

// Messages from Client to Server
export type PesanMasuk =
  | { tipe: "AUTH"; token: string }
  | { tipe: "LANGGANAN"; topik: string }
  | { tipe: "BATAL_LANGGANAN"; topik: string }
  | { tipe: "KIRIM"; topik: string; isi: unknown }
  | { tipe: "PING" };

// Messages from Server to Client
export type PesanKeluar =
  | { tipe: "AUTH_OK"; sessionId: string }
  | { tipe: "AUTH_GAGAL"; alasan: string }
  | { tipe: "PESAN"; topik: string; dari: string; isi: unknown; waktu: string }
  | { tipe: "ERROR"; kode: string; pesan: string }
  | { tipe: "PONG" }
  | { tipe: "DITUTUP"; alasan: string };

// Client connection status stored by the server
export interface StatusKoneksi {
  sessionId: string;
  penggunaId: string | null;
  terautentikasi: boolean;
  topikDilanggani: Set<string>;
  terhubungPada: Date;
  lastPingPada: Date;
}

// Type guard — validate data from the network before processing
export function isPesanMasuk(data: unknown): data is PesanMasuk {
  if (typeof data !== "object" || data === null) return false;
  const d = data as Record<string, unknown>;
  if (typeof d.tipe !== "string") return false;

  switch (d.tipe) {
    case "AUTH":
      return typeof d.token === "string";
    case "LANGGANAN":
    case "BATAL_LANGGANAN":
      return typeof d.topik === "string" && d.topik.length > 0;
    case "KIRIM":
      return typeof d.topik === "string" && "isi" in d;
    case "PING":
      return true;
    default:
      return false;
  }
}

// src/server/ws-server.ts
import { WebSocketServer, WebSocket, RawData } from "ws";
import { IncomingMessage } from "http";
import { parse as parseUrl } from "url";
import type { PesanMasuk, PesanKeluar, StatusKoneksi } from "../types/ws-messages";
import { isPesanMasuk } from "../types/ws-messages";

// Extend WebSocket with additional metadata
interface WebSocketDenganMeta extends WebSocket {
  status: StatusKoneksi;
  isAlive: boolean; // For the ping-pong heartbeat
}

class WebSocketServerApp {
  private wss: WebSocketServer;
  private koneksi = new Map<string, WebSocketDenganMeta>();
  private intervalHeartbeat: NodeJS.Timer | undefined;

  constructor(private readonly port: number) {
    this.wss = new WebSocketServer({
      port,
      // Verify the client during the handshake — reject before the connection forms
      verifyClient: this.verifikasiClient.bind(this),
    });

    this.wss.on("connection", this.tanganiKoneksiBaru.bind(this));
    this.wss.on("error", (err) => console.error("[WS Server] Error:", err));

    this.mulaiHeartbeat();
    console.log(`[WS Server] Berjalan di ws://localhost:${port}`);
  }

  // Verify during the HTTP upgrade — can check Origin, rate limits, etc.
  private verifikasiClient(
    info: { origin: string; req: IncomingMessage; secure: boolean },
    callback: (result: boolean, code?: number, message?: string) => void
  ): void {
    const allowedOrigins = [
      "http://localhost:3000",
      "https://muslimapps.id",
    ];

    if (!allowedOrigins.includes(info.origin)) {
      callback(false, 403, "Origin tidak diizinkan");
      return;
    }

    callback(true);
  }

  private tanganiKoneksiBaru(ws: WebSocket, req: IncomingMessage): void {
    const wsMeta = ws as WebSocketDenganMeta;
    const sessionId = crypto.randomUUID();
    const { query } = parseUrl(req.url ?? "", true);

    wsMeta.status = {
      sessionId,
      penggunaId: null,
      terautentikasi: false,
      topikDilanggani: new Set(),
      terhubungPada: new Date(),
      lastPingPada: new Date(),
    };
    wsMeta.isAlive = true;

    this.koneksi.set(sessionId, wsMeta);
    console.log(`[WS Server] Klien terhubung: ${sessionId} (${req.socket.remoteAddress})`);

    // Handle incoming messages
    wsMeta.on("message", (data: RawData) => {
      this.prosesPesanMasuk(wsMeta, data);
    });

    // Heartbeat — respond to pong
    wsMeta.on("pong", () => {
      wsMeta.isAlive = true;
      wsMeta.status.lastPingPada = new Date();
    });

    wsMeta.on("close", (code, reason) => {
      console.log(
        `[WS Server] Klien terputus: ${sessionId} (kode: ${code}, alasan: ${reason.toString()})`
      );
      this.koneksi.delete(sessionId);
    });

    wsMeta.on("error", (err) => {
      console.error(`[WS Server] Error pada ${sessionId}:`, err.message);
      this.koneksi.delete(sessionId);
    });
  }

  private prosesPesanMasuk(ws: WebSocketDenganMeta, rawData: RawData): void {
    // Parse and validate
    let data: unknown;
    try {
      data = JSON.parse(rawData.toString());
    } catch {
      this.kirim(ws, { tipe: "ERROR", kode: "PARSE_ERROR", pesan: "Data bukan JSON yang valid" });
      return;
    }

    if (!isPesanMasuk(data)) {
      this.kirim(ws, { tipe: "ERROR", kode: "INVALID_MESSAGE", pesan: "Format pesan tidak dikenali" });
      return;
    }

    // Route by message type
    switch (data.tipe) {
      case "AUTH":
        this.tanganiAuth(ws, data.token);
        break;

      case "LANGGANAN":
        if (!ws.status.terautentikasi) {
          this.kirim(ws, { tipe: "ERROR", kode: "UNAUTH", pesan: "Autentikasi diperlukan" });
          return;
        }
        ws.status.topikDilanggani.add(data.topik);
        console.log(`[WS Server] ${ws.status.sessionId} berlangganan: ${data.topik}`);
        break;

      case "KIRIM":
        if (!ws.status.terautentikasi) {
          this.kirim(ws, { tipe: "ERROR", kode: "UNAUTH", pesan: "Autentikasi diperlukan" });
          return;
        }
        this.siarkanKeTopic(data.topik, ws.status.penggunaId!, data.isi);
        break;

      case "PING":
        this.kirim(ws, { tipe: "PONG" });
        break;

      case "BATAL_LANGGANAN":
        ws.status.topikDilanggani.delete(data.topik);
        break;
    }
  }

  private tanganiAuth(ws: WebSocketDenganMeta, token: string): void {
    // Simulated token verification
    try {
      const payload = verifikasiJWT(token);
      ws.status.penggunaId = payload.sub;
      ws.status.terautentikasi = true;
      this.kirim(ws, { tipe: "AUTH_OK", sessionId: ws.status.sessionId });
      console.log(`[WS Server] Autentikasi berhasil: ${payload.sub}`);
    } catch {
      this.kirim(ws, { tipe: "AUTH_GAGAL", alasan: "Token tidak valid atau kedaluwarsa" });
      ws.close(1008, "Autentikasi gagal"); // 1008 = Policy Violation
    }
  }

  // Broadcast to all subscribers of a given topic
  private siarkanKeTopic(topik: string, dariPenggunaId: string, isi: unknown): void {
    const pesan: PesanKeluar = {
      tipe: "PESAN",
      topik,
      dari: dariPenggunaId,
      isi,
      waktu: new Date().toISOString(),
    };

    let jumlahPenerima = 0;
    for (const koneksi of this.koneksi.values()) {
      if (
        koneksi.status.topikDilanggani.has(topik) &&
        koneksi.readyState === WebSocket.OPEN
      ) {
        this.kirim(koneksi, pesan);
        jumlahPenerima++;
      }
    }

    console.log(`[WS Server] Siaran ke topik "${topik}": ${jumlahPenerima} penerima`);
  }

  // Send a message to a single client with type safety
  private kirim(ws: WebSocket, pesan: PesanKeluar): void {
    if (ws.readyState !== WebSocket.OPEN) return;
    ws.send(JSON.stringify(pesan));
  }

  // Heartbeat: ping all clients every 30 seconds, disconnect unresponsive ones
  private mulaiHeartbeat(): void {
    this.intervalHeartbeat = setInterval(() => {
      for (const [sessionId, ws] of this.koneksi) {
        if (!ws.isAlive) {
          console.log(`[WS Server] Heartbeat gagal, memutus: ${sessionId}`);
          ws.terminate(); // Force close without a close handshake
          this.koneksi.delete(sessionId);
          continue;
        }
        ws.isAlive = false;
        ws.ping(); // Send a ping, wait for the pong
      }
    }, 30_000);
  }

  // Broadcast to ALL connected clients
  siarkanSemua(pesan: PesanKeluar): void {
    for (const ws of this.koneksi.values()) {
      if (ws.readyState === WebSocket.OPEN) {
        this.kirim(ws, pesan);
      }
    }
  }

  get statistik() {
    return {
      totalKoneksi: this.koneksi.size,
      terautentikasi: [...this.koneksi.values()].filter((ws) => ws.status.terautentikasi).length,
    };
  }

  berhenti(): void {
    clearInterval(this.intervalHeartbeat as unknown as number);
    for (const ws of this.koneksi.values()) {
      ws.close(1001, "Server ditutup"); // 1001 = Going Away
    }
    this.wss.close();
  }
}

// Mock JWT verifier
function verifikasiJWT(token: string): { sub: string } {
  if (token.startsWith("valid-")) return { sub: token.replace("valid-", "usr-") };
  throw new Error("Token tidak valid");
}

// Run the server
const server = new WebSocketServerApp(8080);

// Graceful shutdown
process.on("SIGTERM", () => server.berhenti());
process.on("SIGINT", () => server.berhenti());

// src/client/ws-client.ts — runs in the browser
import type { PesanKeluar, PesanMasuk } from "../types/ws-messages";

class WebSocketClient {
  private ws: WebSocket | null = null;
  private reconnectDelay = 1000;
  private readonly maxDelay = 30_000;
  private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
  private harus_terhubung = true;

  constructor(
    private readonly url: string,
    private readonly token: string,
    private readonly onPesan?: (pesan: PesanKeluar) => void
  ) {}

  hubungkan(): void {
    if (this.ws?.readyState === WebSocket.OPEN) return;

    this.ws = new WebSocket(this.url);

    this.ws.addEventListener("open", () => {
      console.log("[WSClient] Terhubung");
      this.reconnectDelay = 1000; // Reset the backoff

      // Authenticate immediately after connecting
      this.kirim({ tipe: "AUTH", token: this.token });
    });

    this.ws.addEventListener("message", (event: MessageEvent<string>) => {
      try {
        const pesan = JSON.parse(event.data) as PesanKeluar;
        this.tanganiPesanMasuk(pesan);
        this.onPesan?.(pesan);
      } catch {
        console.error("[WSClient] Gagal parse pesan:", event.data);
      }
    });

    this.ws.addEventListener("close", (event) => {
      console.log(`[WSClient] Terputus (kode: ${event.code}, alasan: ${event.reason})`);

      // Don't reconnect for certain codes (auth failed, etc.)
      const tidakReconnect = [1008, 1003];
      if (this.harus_terhubung && !tidakReconnect.includes(event.code)) {
        this.jadwalkanReconnect();
      }
    });

    this.ws.addEventListener("error", () => {
      console.error("[WSClient] Error koneksi");
    });
  }

  private tanganiPesanMasuk(pesan: PesanKeluar): void {
    switch (pesan.tipe) {
      case "AUTH_OK":
        console.log(`[WSClient] Autentikasi berhasil, session: ${pesan.sessionId}`);
        break;
      case "AUTH_GAGAL":
        console.error(`[WSClient] Autentikasi gagal: ${pesan.alasan}`);
        this.harus_terhubung = false; // Don't retry if auth fails
        break;
      case "PONG":
        // The server responded to our ping
        break;
      case "ERROR":
        console.error(`[WSClient] Error dari server [${pesan.kode}]: ${pesan.pesan}`);
        break;
    }
  }

  private jadwalkanReconnect(): void {
    if (this.reconnectTimer) return;

    console.log(`[WSClient] Reconnect dalam ${this.reconnectDelay}ms...`);

    this.reconnectTimer = setTimeout(() => {
      this.reconnectTimer = null;
      this.hubungkan();
    }, this.reconnectDelay);

    // Exponential backoff with jitter
    this.reconnectDelay = Math.min(
      this.reconnectDelay * 2 + Math.random() * 1000,
      this.maxDelay
    );
  }

  kirim(pesan: PesanMasuk): void {
    if (this.ws?.readyState !== WebSocket.OPEN) {
      console.warn("[WSClient] Tidak terhubung, pesan dibuang");
      return;
    }
    this.ws.send(JSON.stringify(pesan));
  }

  langganan(topik: string): void {
    this.kirim({ tipe: "LANGGANAN", topik });
  }

  batalLangganan(topik: string): void {
    this.kirim({ tipe: "BATAL_LANGGANAN", topik });
  }

  kirimPesan(topik: string, isi: unknown): void {
    this.kirim({ tipe: "KIRIM", topik, isi });
  }

  putuskan(): void {
    this.harus_terhubung = false;
    if (this.reconnectTimer) {
      clearTimeout(this.reconnectTimer);
    }
    this.ws?.close(1000, "Ditutup oleh pengguna");
  }

  get statusKoneksi(): string {
    switch (this.ws?.readyState) {
      case WebSocket.CONNECTING: return "menghubungkan";
      case WebSocket.OPEN: return "terhubung";
      case WebSocket.CLOSING: return "menutup";
      case WebSocket.CLOSED: return "terputus";
      default: return "tidak diinisialisasi";
    }
  }
}

// Usage in a web application
const klien = new WebSocketClient(
  "wss://api.muslimapps.id/ws",
  "valid-user-token",
  (pesan) => {
    if (pesan.tipe === "PESAN") {
      console.log(`[${pesan.topik}] ${pesan.dari}: ${JSON.stringify(pesan.isi)}`);
    }
  }
);

klien.hubungkan();
klien.langganan("jadwal-sholat");
klien.kirimPesan("jadwal-sholat", { kota: "Jakarta" });

// src/server/sse-server.ts — using Express
import express from "express";

const app = express();

app.get("/notifikasi", (req, res) => {
  // Set SSE headers
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");
  res.setHeader("X-Accel-Buffering", "no"); // Important for Nginx

  // Send the first event
  res.write("data: {\"tipe\":\"TERHUBUNG\"}\n\n");

  // Send updates periodically
  const interval = setInterval(() => {
    const data = {
      tipe: "UPDATE",
      waktu: new Date().toISOString(),
      pesan: "Data terbaru dari server",
    };
    res.write(`data: ${JSON.stringify(data)}\n\n`);
  }, 5000);

  // Clean up when the client disconnects
  req.on("close", () => {
    clearInterval(interval);
    console.log("SSE client terputus");
  });
});

// SSE client (browser)
// const source = new EventSource("/notifikasi");
// source.onmessage = (event) => {
//   const data = JSON.parse(event.data);
//   console.log(data);
// };