Интеграция Centrifugo для real-time сообщений на сайте

Интеграция Centrifugo для real-time сообщений на сайте

Centrifugo — self-hosted real-time сервер, совместимый с любым бэкендом через HTTP API. Написан на Go, потребляет минимум ресурсов, поддерживает WebSocket, SSE, HTTP streaming и SockJS.

Архитектура интеграции

Ваш бэкенд (любой язык)
    │── HTTP API → Centrifugo (публикация)
    │── gRPC API → Centrifugo
    │
Клиент (браузер) ◄── WebSocket ──► Centrifugo

Установка через Docker

# docker-compose.yml
services:
  centrifugo:
    image: centrifugo/centrifugo:v5
    command: centrifugo -c /centrifugo/config.json
    ports:
      - "8000:8000"
    volumes:
      - ./centrifugo-config.json:/centrifugo/config.json

// centrifugo-config.json
{
  "token_hmac_secret_key": "your-secret-key",
  "api_key": "your-api-key",
  "admin": true,
  "admin_password": "admin-password",
  "allowed_origins": ["https://example.com"],
  "namespaces": [
    {
      "name": "notifications",
      "subscribe_for_client": false,
      "history_size": 100,
      "history_ttl": "24h"
    },
    {
      "name": "chat",
      "presence": true,
      "join_leave": true,
      "history_size": 500,
      "history_ttl": "1h"
    }
  ]
}

Сервер: генерация JWT и публикация

import jwt from 'jsonwebtoken';
import axios from 'axios';

// Токен для клиента
function generateCentrifugoToken(userId: string): string {
  return jwt.sign(
    {
      sub: userId,
      exp: Math.floor(Date.now() / 1000) + 3600
    },
    process.env.CENTRIFUGO_SECRET,
    { algorithm: 'HS256' }
  );
}

// Публикация сообщения через HTTP API
async function publishMessage(channel: string, data: unknown): Promise<void> {
  await axios.post(`${process.env.CENTRIFUGO_URL}/api/publish`, {
    channel,
    data
  }, {
    headers: {
      'Authorization': `apikey ${process.env.CENTRIFUGO_API_KEY}`,
      'Content-Type': 'application/json'
    }
  });
}

// Отправить уведомление пользователю
await publishMessage(`notifications#${userId}`, {
  type: 'order_shipped',
  orderId: '12345',
  message: 'Ваш заказ отправлен'
});

// Batch publish
await axios.post(`${process.env.CENTRIFUGO_URL}/api/batch`, {
  commands: userIds.map(id => ({
    publish: {
      channel: `notifications#${id}`,
      data: notification
    }
  }))
});

Клиент (JavaScript)

import { Centrifuge } from 'centrifuge';

const centrifuge = new Centrifuge(process.env.NEXT_PUBLIC_CENTRIFUGO_URL, {
  token: await fetchCentrifugoToken(),  // от вашего бэкенда
  debug: process.env.NODE_ENV === 'development'
});

centrifuge.on('connect', (ctx) => {
  console.log('Connected to Centrifugo', ctx.transport);
});

centrifuge.on('disconnect', (ctx) => {
  console.log('Disconnected', ctx.reason);
});

// Подписка на персональные уведомления
const sub = centrifuge.newSubscription(`notifications#${userId}`, {
  getToken: async (ctx) => {
    // Получить subscription токен для private канала
    const resp = await fetch('/api/centrifugo/subscription-token', {
      method: 'POST',
      body: JSON.stringify({ channel: ctx.channel })
    });
    return (await resp.json()).token;
  }
});

sub.on('publication', (ctx) => {
  const notification = ctx.data;
  showNotification(notification.message);
});

sub.on('subscribing', (ctx) => console.log('Subscribing...'));
sub.on('subscribed', (ctx) => console.log('Subscribed'));

sub.subscribe();
centrifuge.connect();

Presence — онлайн-статусы

// Кто сейчас читает документ (collaborative editing)
const docSub = centrifuge.newSubscription(`chat:doc-${documentId}`);

docSub.presence().then((presence) => {
  const onlineUsers = Object.values(presence.clients);
  setOnlineUsers(onlineUsers);
});

docSub.on('join', ({ info }) => {
  console.log(`${info.user} joined`);
});

docSub.on('leave', ({ info }) => {
  console.log(`${info.user} left`);
});

Сроки

Centrifugo + Node.js SDK + клиент с private каналами — 3–5 дней.