WebSocket — добавление real-time событий

Как добавить WebSocket-уведомления в существующий модуль.

Архитектура

sequenceDiagram
    participant C1 as Client 1
    participant API as FastAPI Router
    participant CM as ConnectionManager
    participant PS as PubSub (PG/Redis)
    participant C2 as Client 2

    C1->>API: POST /tickets/1/assign
    API->>API: business logic
    API->>CM: send_to_user(user_id, event)
    CM->>PS: publish("send_to_user", {...})
    PS->>CM: notification (all workers)
    CM->>C2: WebSocket: {"type": "ticket_assigned"}

Backend — отправка события

1. Где лежит chat_manager

backend/base/crm/tickets/routers/tickets.py

env.apps.chat.chat_manager

2. Отправка после действия

@router_private.post("/{ticket_id}/assign")
async def assign_ticket(req: Request, ticket_id: int, body: AssignBody):
    env: "Environment" = req.app.state.env

    ticket = await env.models.ticket.get(ticket_id)
    await ticket.update(env.models.ticket(
        assigned_to=body.user_id,
        status="in_progress",
    ))

    # Уведомить назначенного пользователя
    await env.apps.chat.chat_manager.send_to_user(             # (1)!
        user_id=body.user_id,
        message={
            "type": "ticket_assigned",
            "ticket_id": ticket_id,
            "title": ticket.title,
        },
    )

    return {"success": True}

1. send_to_user — отправляет событие конкретному пользователю через все его WebSocket-подключения (включая другие вкладки и worker-процессы).

Методы отправки

Метод	Описание
send_to_chat(chat_id, message, exclude_user=None)	Всем участникам чата — активным пользователям из chat_member, на любом воркере. Подписываться клиенту не нужно.
send_to_user(user_id, message)	Конкретному пользователю

Транзакции

Вызов внутри async with env.apps.db.get_transaction() доставляется на COMMIT: pg_notify идёт через соединение транзакции, поэтому клиент никогда не получит событие раньше, чем увидит данные, а откат транзакции отменит и событие. Вне транзакции событие уходит сразу.

Размер события

Лимит pg_notify — около 8 КБ, событие больше выбрасывается с ошибкой в лог. Для new_message тело режет ChatMessage.serialize_for_ws (флаг body_truncated, фронт дочитывает сообщение из REST); своим событиям держите payload маленьким — id и короткие поля.

Frontend — обработка события

frontend/src/fara_tickets/hooks/useTicketEvents.ts

import { useWebSocket } from '@fara_chat/hooks/useWebSocket';
import { notifications } from '@mantine/notifications';

export function useTicketEvents() {
    const handleMessage = useCallback((data: any) => {
        switch (data.type) {
            case 'ticket_assigned':
                // Показать уведомление
                notifications.show({
                    title: 'Новый тикет',
                    message: `Вам назначен тикет: ${data.title}`,
                    color: 'blue',
                });

                // Инвалидировать кэш списка тикетов
                dispatch(
                    crudApi.util.invalidateTags([
                        { type: 'tickets', id: 'LIST' }
                    ])
                );
                break;

            case 'ticket_status_changed':
                // Обновить конкретный тикет в кэше
                dispatch(
                    crudApi.util.updateQueryData(
                        'read',
                        { model: 'tickets', id: data.ticket_id },
                        (draft) => {
                            draft.data.status = data.status;
                        }
                    )
                );
                break;
        }
    }, [dispatch]);

    useWebSocket(token, handleMessage);
}

Тестирование

В тестах мокай send_to_chat / send_to_user фикстурой mock_chat_ws (tests/conftest.py) — она подменяет методы на реальном chat_manager приложения:

async def test_assign_ticket_sends_notification(
    self, authenticated_client, mock_chat_ws
):
    client, user_id, token = authenticated_client
    ticket_id = await self._create_ticket()

    response = await client.post(
        f"/tickets/{ticket_id}/assign",
        json={"user_id": user_id},
    )

    assert response.status_code == 200

    # Проверяем что WS-уведомление отправлено
    mock_chat_ws.send_to_user.assert_called_once_with(
        user_id=user_id,
        message={
            "type": "ticket_assigned",
            "ticket_id": ticket_id,
            "title": "Test Ticket",
        },
    )

Не забудь замокать WebSocket

Без мока send_to_user / send_to_chat тест уйдёт в реальный PubSub и может зависнуть (если PubSub listener занял все соединения пула).