Урок 05

Real-time уведомления

Что такое real-time

Real-time — сервер сам присылает данные, когда что-то изменилось, без опроса.

Три способа:

  • Polling — клиент спрашивает каждые N секунд.
  • SSE — сервер держит однонаправленный поток.
  • WebSocket — двунаправленное соединение.

Сравнение

СпособНаправлениеНагрузкаКогда
PollingЗапрос-ответВысокаяРедко меняется
Long pollingЗапрос-ответСредняяПростая реализация
SSEСервер → клиентНизкаяУведомления, стрим
WebSocketОбаНизкаяЧат, игры, курсоры

Правило:

  • Уведомления — SSE.
  • Чат / интерактив — WebSocket.
  • AI-стрим — SSE.

Архитектура real-time

Laravel бизнес-событие
    ↓
Redis PUBLISH
    ↓
Node.js подписан
    ↓
Socket.IO / SSE
    ↓
Клиенты (браузер, мобильный)

Разделение:

  • Laravel — источник событий.
  • Redis — шина.
  • Node.js — доставка клиентам.

Redis Pub/Sub

Laravel публикует:

use Illuminate\Support\Facades\Redis;

public function notify(int $userId, array $payload): void
{
    Redis::publish('app:events', json_encode([
        'channel' => "user:{$userId}",
        'event' => 'notification',
        'data' => $payload,
    ]));
}

Node.js подписан:

import { Redis } from 'ioredis';

const pub = new Redis(process.env.REDIS_URL);
const sub = new Redis(process.env.REDIS_URL);

sub.subscribe('app:events');

sub.on('message', (channel, message) => {
    const { channel: userChannel, event, data } = JSON.parse(message);

    io.to(userChannel).emit(event, data);
});

Что даёт: Laravel не знает про WebSocket, Node.js не знает про бизнес-логику.

Socket.IO сервер

import { Server } from 'socket.io';
import { createAdapter } from '@socket.io/redis-adapter';

const io = new Server(httpServer, {
    cors: { origin: process.env.FRONTEND_URL, credentials: true },
    adapter: createAdapter(pub, sub),
});

io.use(async (socket, next) => {
    const token = socket.handshake.auth.token;

    try {
        const user = await verifyToken(token);
        socket.data.user = user;
        next();
    } catch (err) {
        next(new Error('Unauthorized'));
    }
});

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

    socket.join(`user:${user.id}`);

    if (user.isAdmin) socket.join('admins');

    socket.on('disconnect', () => {
        // auto leave rooms
    });
});

Аутентификация

Три подхода:

1. Sanctum-токен в handshake:

io.use(async (socket, next) => {
    const token = socket.handshake.auth.token;
    const user = await verifySanctumToken(token);
    socket.data.user = user;
    next();
});

2. JWT:

import jwt from 'jsonwebtoken';

io.use((socket, next) => {
    const { token } = socket.handshake.auth;
    try {
        const payload = jwt.verify(token, process.env.JWT_SECRET);
        socket.data.user = payload;
        next();
    } catch {
        next(new Error('Invalid token'));
    }
});

3. Проверка в Redis:

io.use(async (socket, next) => {
    const token = socket.handshake.auth.token;
    const session = await redis.get(`session:${token}`);

    if (!session) return next(new Error('Unauthorized'));

    socket.data.user = JSON.parse(session);
    next();
});

Правило: первое сообщение — только после аутентификации.

Rooms

Что даёт: группировка соединений.

// Подписать клиента на его комнату
socket.join(`user:${user.id}`);

// Отправить в комнату
io.to(`user:${user.id}`).emit('notification', data);

// Всем в комнате, кроме отправителя
socket.to(`room:${roomId}`).emit('message', data);

// Всем
io.emit('broadcast', data);

Использование:

  • user:{id} — личные уведомления.
  • order:{id} — статус заказа.
  • chat:{id} — сообщения чата.
  • admins — для админки.

Presence (онлайн-статус)

Простой вариант:

const onlineUsers = new Map<number, Set<string>>();

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

    if (!onlineUsers.has(userId)) onlineUsers.set(userId, new Set());
    onlineUsers.get(userId).add(socket.id);

    socket.broadcast.emit('user:online', userId);

    socket.on('disconnect', () => {
        onlineUsers.get(userId).delete(socket.id);

        if (onlineUsers.get(userId).size === 0) {
            onlineUsers.delete(userId);
            socket.broadcast.emit('user:offline', userId);
        }
    });
});

С Redis (несколько инстансов):

await redis.sadd(`online:${userId}`, socket.id);
await redis.expire(`online:${userId}`, 300);

Масштабирование

Проблема: 2+ инстанса Node.js — клиент на инстансе A, событие на инстансе B.

Решение 1 — Redis adapter:

import { createAdapter } from '@socket.io/redis-adapter';

io.adapter(createAdapter(pubClient, subClient));

Что даёт: io.to(room).emit() работает между инстансами.

Решение 2 — sticky sessions:

upstream nodejs {
    ip_hash;
    server 127.0.0.1:3001;
    server 127.0.0.1:3002;
}

Правило: для WebSocket — Redis adapter + sticky.

SSE — Server-Sent Events

Проще, чем WebSocket:

app.get('/notifications', async (req, res) => {
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');

    const userId = req.user.id;
    const channel = `user:${userId}`;

    const sub = redis.duplicate();
    await sub.subscribe(channel);

    sub.on('message', (_, message) => {
        res.write(`data: ${message}\n\n`);
    });

    req.on('close', () => {
        sub.unsubscribe(channel);
        sub.quit();
    });
});

Клиент:

const es = new EventSource('/notifications');

es.onmessage = (e) => {
    const data = JSON.parse(e.data);
    console.log(data);
};

Что даёт: однонаправленный поток, авто-переподключение.

Когда SSE лучше WS:

  • Только сервер → клиент.
  • Простая реализация.
  • Работает через HTTP/2.

Laravel Broadcasting

Альтернатива своей реализации:

composer require pusher/pusher-php-server

// config/broadcasting.php
'default' => env('BROADCAST_DRIVER', 'redis'),

'connections' => [
    'redis' => [
        'driver' => 'redis',
        'connection' => 'default',
    ],
],

Событие:

class OrderShipped implements ShouldBroadcast
{
    use Dispatchable, InteractsWithSockets, SerializesModels;

    public function __construct(public Order $order) {}

    public function broadcastOn(): array
    {
        return [
            new PrivateChannel('user.' . $this->order->user_id),
        ];
    }

    public function broadcastAs(): string
    {
        return 'order.shipped';
    }
}

Клиент (Laravel Echo):

import Echo from 'laravel-echo';

window.Echo = new Echo({
    broadcaster: 'socket.io',
    host: 'wss://api.example.com',
    authEndpoint: '/broadcasting/auth',
});

Echo.private(`user.${userId}`)
    .listen('.order.shipped', (e) => console.log(e));

Реализация с laravel-echo-server

Проще, чем свой Socket.IO:

npm install -g laravel-echo-server

laravel-echo-server init
laravel-echo-server start

Что даёт: готовый сервер для Broadcasting.

Когда использовать: нужны стандартные broadcast-события.

Когда свой Socket.IO:

  • Сложная логика комнат.
  • Presence.
  • Кастомные события.

Порядок и доставка

Проблемы:

  • Событие пропало — клиент не получил.
  • Дубликат — пришло дважды.
  • Порядок нарушен — важные события перепутались.

Решение:

  • Event ID — каждое событие с UUID.
  • Клиент хранит последний — при reconnect запрашивает с after_id.
  • Идемпотентный обработчик — повторная обработка безопасна.
io.to(user).emit('event', {
    id: uuid(),
    type: 'order.shipped',
    data: { orderId: 123 },
});

Reconnect:

socket.on('connect', async () => {
    const lastId = localStorage.getItem('lastEventId');
    const events = await fetch(`/api/events?after=${lastId}`);
    events.forEach(processEvent);
});

Heartbeat

Проблема: прокси рвут idle-соединения через 60 сек.

Решение:

// Server
const interval = setInterval(() => {
    io.emit('ping', Date.now());
}, 25000);

io.on('connection', () => {
    clearInterval(interval);
});

// Client
socket.on('ping', () => socket.emit('pong'));

Что даёт: соединение активно.

Мониторинг

Что смотреть:

  • Connected sockets — сколько онлайн.
  • Rooms count — сколько комнат.
  • Messages/sec — сколько событий.
  • Redis lag — задержка pub/sub.
  • Reconnects — частота переподключений.

Grafana dashboard:

  • Prometheus exporter для Socket.IO.
  • Метрики из Redis.

Практика: уведомления о заказах

1. Laravel при смене статуса:

class OrderStatusChanged
{
    public function __construct(public Order $order) {}

    public function handle(): void
    {
        Redis::publish('app:events', json_encode([
            'channel' => "user:{$this->order->user_id}",
            'event' => 'order.status',
            'data' => [
                'orderId' => $this->order->id,
                'status' => $this->order->status,
                'updatedAt' => $this->order->updated_at->toIso8601String(),
            ],
        ]));
    }
}

2. Node.js:

sub.on('message', (_, message) => {
    const { channel, event, data } = JSON.parse(message);
    io.to(channel).emit(event, data);
});

3. Фронт:

socket.on('order.status', (data) => {
    updateOrderInUI(data);
});

Частые ошибки

1. WebSocket без авторизации:

Любой может подписаться на user:5

Решение: middleware в io.use.

2. Один инстанс без Redis adapter:

События теряются при масштабировании

Решение: createAdapter.

3. Нет heartbeat:

Соединения рвутся через прокси

Решение: ping каждые 25 сек.

4. Нет reconnection:

После обрыва клиент не переподключается

Решение: Socket.IO сам, для SSE — свой.

5. Событие без ID:

При reconnect теряются события

Решение: событие с UUID + запрос пропущенных.

6. Все комнаты в одном процессе:

Память кончается

Решение: Redis для presence.

Итог

  • Real-time — через SSE или WebSocket.
  • Уведомления — SSE, чат — WebSocket.
  • Redis — шина между Laravel и Node.js.
  • Socket.IO rooms — группировка.
  • Redis adapter — масштабирование.
  • Heartbeat — против прокси.
  • Event ID — против потерь.
  • Laravel Broadcasting — готовое решение.
  • Presence — через Redis.

Дальше — мониторинг и алерты.