Реалізація Real-Time Notifications (WebSocket/SSE) на сайті

Реалізація Real-Time Notifications (WebSocket/SSE) на веб-сайті

Сповіщення в реальному часі — один з найбільш частих запитів. Технічно між WebSocket та SSE принципова різниця: SSE — одностороння потокова передача від сервера до клієнта через звичайний HTTP, WebSocket — двосторонній канал. Для сповіщень часто достатньо SSE.

SSE vs WebSocket: коли що вибирати

SSE підходить, якщо потрібно тільки отримувати події з сервера: нові повідомлення, оновлення статусу, оповіщення. Працює через стандартний HTTP/2, підтримує автоматичне переподключення, не потребує додаткових бібліотек на клієнті.

WebSocket потрібен, якщо клієнт також надсилає дані в реальному часі: чат, ігри, спільне редагування.

SSE: Server → Client (одностороння, HTTP)
WebSocket: Server ↔ Client (двостороння, WS protocol)

SSE: реалізація на Node.js/Express

// server/routes/notifications.ts
import { Router, Request, Response } from 'express';
import { authMiddleware } from '../middleware/auth';

const router = Router();

// Map userId -> Set<Response>
const clients = new Map<string, Set<Response>>();

router.get('/stream', authMiddleware, (req: Request, res: Response) => {
  const userId = req.user!.id;

  res.writeHead(200, {
    'Content-Type':  'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection':    'keep-alive',
    'X-Accel-Buffering': 'no',  // важливо для nginx
  });

  // Heartbeat кожні 30 секунд — інакше proxy/браузер розірве з'єднання
  const heartbeat = setInterval(() => {
    res.write(':heartbeat\n\n');
  }, 30_000);

  // Реєстрація клієнта
  if (!clients.has(userId)) clients.set(userId, new Set());
  clients.get(userId)!.add(res);

  // Початковий снімок непрочитаних
  getUnreadNotifications(userId).then((notifications) => {
    res.write(sseEvent('init', notifications));
  });

  req.on('close', () => {
    clearInterval(heartbeat);
    clients.get(userId)?.delete(res);
    if (clients.get(userId)?.size === 0) clients.delete(userId);
  });
});

function sseEvent(type: string, data: unknown, id?: string): string {
  let msg = '';
  if (id) msg += `id: ${id}\n`;
  msg += `event: ${type}\n`;
  msg += `data: ${JSON.stringify(data)}\n\n`;
  return msg;
}

// Публічна функція для надсилання сповіщення користувачу
export function pushNotification(userId: string, notification: Notification) {
  const userClients = clients.get(userId);
  if (!userClients) return;

  const msg = sseEvent('notification', notification, notification.id);
  userClients.forEach((res) => res.write(msg));
}

export default router;

Клієнтська частина: EventSource

class NotificationService {
  private es: EventSource | null = null;
  private reconnectDelay = 1000;

  connect() {
    this.es = new EventSource('/api/notifications/stream', {
      withCredentials: true,
    });

    this.es.addEventListener('init', (e) => {
      const notifications = JSON.parse(e.data);
      notificationStore.setAll(notifications);
    });

    this.es.addEventListener('notification', (e) => {
      const notification = JSON.parse(e.data);
      notificationStore.add(notification);
      this.showToast(notification);
    });

    this.es.addEventListener('error', () => {
      this.es?.close();
      // Експоненціальний backoff
      setTimeout(() => {
        this.reconnectDelay = Math.min(this.reconnectDelay * 2, 30_000);
        this.connect();
      }, this.reconnectDelay);
    });

    this.es.addEventListener('open', () => {
      this.reconnectDelay = 1000; // скидання при успішному з'єднанні
    });
  }

  disconnect() {
    this.es?.close();
    this.es = null;
  }

  private showToast(notification: Notification) {
    // інтеграція з бібліотекою toast на вибір (sonner, react-hot-toast, тощо)
    toast(notification.title, {
      description: notification.body,
      action: notification.actionUrl
        ? { label: 'Відкрити', onClick: () => navigate(notification.actionUrl!) }
        : undefined,
    });
  }
}

WebSocket-варіант: інтеграція з чергою

Для production сповіщення не надсилаються прямо з запиту до SSE-клієнта — між прикладним шаром та доставкою стоїть черга:

HTTP Request → DB збереження → Redis Publish → WebSocket Server → Client

// Публікація події (з будь-якого сервісу/робітника)
import { createClient } from 'redis';

const pub = createClient({ url: process.env.REDIS_URL });
await pub.connect();

async function emitNotification(userId: string, notification: Notification) {
  await db.notifications.create({ data: notification });
  await pub.publish(
    `notifications:${userId}`,
    JSON.stringify(notification)
  );
}

// WebSocket сервер підписується через окремий Redis subscriber
const sub = createClient({ url: process.env.REDIS_URL });
await sub.connect();

io.on('connection', (socket) => {
  const userId = socket.data.userId;

  // Підписка на особистий канал
  sub.subscribe(`notifications:${userId}`, (message) => {
    socket.emit('notification', JSON.parse(message));
  });

  socket.on('notification:read', async (notificationId: string) => {
    await db.notifications.update({
      where: { id: notificationId },
      data:  { readAt: new Date() },
    });
  });

  socket.on('disconnect', () => {
    sub.unsubscribe(`notifications:${userId}`);
  });
});

Структура сповіщення

interface Notification {
  id:         string;
  userId:     string;
  type:       'comment' | 'mention' | 'order' | 'system' | 'alert';
  title:      string;
  body:       string;
  actorId?:   string;      // хто ініціював
  entityType?: string;     // 'post' | 'order' | ...
  entityId?:  string;
  actionUrl?: string;
  imageUrl?:  string;
  readAt?:    Date;
  createdAt:  Date;
}

Групування та батчинг

Якщо за секунду надходить багато подій (масова розсилка, потік даних), клієнт отримує окремо SSE-event для кожного. Краще групувати:

// Серверний буфер: 200ms debounce на flush
const pendingByUser = new Map<string, Notification[]>();

function bufferNotification(userId: string, notification: Notification) {
  if (!pendingByUser.has(userId)) {
    pendingByUser.set(userId, []);
    setTimeout(() => flushUser(userId), 200);
  }
  pendingByUser.get(userId)!.push(notification);
}

function flushUser(userId: string) {
  const batch = pendingByUser.get(userId) ?? [];
  pendingByUser.delete(userId);

  if (batch.length === 1) {
    pushToClient(userId, sseEvent('notification', batch[0]));
  } else {
    pushToClient(userId, sseEvent('notifications:batch', batch));
  }
}

Масштабування SSE

SSE тримає HTTP-з'єднання відкритим — кожний підключений користувач займає один file descriptor. Node.js комфортно тримає 10k+ з'єднань, але при горизонтальному масштабуванні (кілька інстансів) користувач може бути підключений до інстансу A, а сповіщення згенеровано інстансом B. Redis Pub/Sub вирішує це — кожний інстанс підписаний на всі канали й доставляє тільки своїм клієнтам.

Для nginx: проксування SSE потребує відключення буферизації:

location /api/notifications/stream {
    proxy_pass http://app_backend;
    proxy_buffering       off;
    proxy_cache           off;
    proxy_read_timeout    3600s;
    proxy_set_header Connection '';
    chunked_transfer_encoding on;
}

Реалізація SSE-сповіщень з Redis: 2–3 дні. Додавання WebSocket з двостороннією логікою (read receipts, typing): ще один день.