Интеграция с WebSocket

TanStack Query изначально проектировался как библиотека для работы с асинхронными запросами и серверным состоянием. Основная модель библиотеки строится вокруг HTTP-запросов, кеширования, фоновых обновлений и повторной синхронизации данных.

Однако современные приложения всё чаще требуют реактивного обновления данных в режиме реального времени:

  • чаты;
  • торговые терминалы;
  • совместное редактирование;
  • игровые интерфейсы;
  • системы мониторинга;
  • панели аналитики;
  • уведомления;
  • live-ленты.

В подобных системах стандартного механизма polling недостаточно. Постоянные HTTP-запросы создают лишнюю нагрузку и увеличивают задержки.

Интеграция WebSocket позволяет:

  • мгновенно обновлять кеш;
  • синхронизировать данные между клиентами;
  • избегать лишних refetch;
  • уменьшать сетевой трафик;
  • строить реактивные интерфейсы.

TanStack Query не содержит встроенного WebSocket-клиента, однако предоставляет мощные механизмы управления кешем, идеально подходящие для real-time архитектуры.

Почему WebSocket не заменяет TanStack Query

Распространённая ошибка — попытка полностью отказаться от TanStack Query после подключения WebSocket.

Это создаёт множество проблем:

  • отсутствие нормализованного кеша;
  • ручное управление загрузкой;
  • сложная обработка ошибок;
  • дублирование состояния;
  • потеря механизмов stale/fresh;
  • отсутствие retry;
  • отсутствие background sync.

Правильная архитектура выглядит иначе:

Ответственность Инструмент
Начальная загрузка данных TanStack Query
Кеширование TanStack Query
Инвалидация TanStack Query
Синхронизация WebSocket
Push-обновления WebSocket
Локальное обновление кеша QueryClient

WebSocket становится источником событий, а TanStack Query остаётся системой управления серверным состоянием.

Базовая схема интеграции

Типичная последовательность работы:

  1. useQuery загружает первоначальные данные.
  2. WebSocket подключается к серверу.
  3. Сервер отправляет события.
  4. Клиент обновляет Query Cache.
  5. Компоненты автоматически перерисовываются.

Создание WebSocket-соединения

Простейший клиент

const socket = new WebSocket("ws://localhost:3000")

Обработка подключения:

socket.addEventListener("open", () => {
  console.log("connected")
})

Получение сообщений:

socket.addEventListener("message", (event) => {
  const data = JSON.parse(event.data)

  console.log(data)
})

Интеграция с QueryClient

Главная идея — обновлять кеш напрямую.

Создание QueryClient

import { QueryClient } from "@tanstack/react-query"

export const queryClient = new QueryClient()

Обновление кеша через setQueryData

Наиболее распространённый сценарий.

Сервер отправляет событие

{
  "type": "NEW_MESSAGE",
  "payload": {
    "id": 10,
    "text": "Hello"
  }
}

Обработка сообщения

socket.addEventListener("message", (event) => {
  const message = JSON.parse(event.data)

  if (message.type === "NEW_MESSAGE") {
    queryClient.setQueryData(
      ["messages"],
      (old = []) => {
        return [...old, message.payload]
      }
    )
  }
})

Как работает setQueryData

Метод:

queryClient.setQueryData(queryKey, updater)

делает следующее:

  1. Находит запись в кеше.
  2. Передаёт текущее значение в updater.
  3. Сохраняет новое состояние.
  4. Вызывает обновление подписанных компонентов.

Почему setQueryData лучше invalidateQueries

Многие разработчики после WebSocket-события делают:

queryClient.invalidateQueries({
  queryKey: ["messages"]
})

Это вызывает новый HTTP-запрос.

При активном real-time потоке подобный подход создаёт:

  • лишнюю нагрузку;
  • скачки UI;
  • сетевой шум;
  • повторные запросы.

Если сервер уже прислал новые данные через WebSocket, повторный HTTP-запрос часто не нужен.

Когда invalidateQueries всё же нужен

Есть сценарии, где сервер присылает только уведомление об изменении:

{
  "type": "POST_UPDATED",
  "id": 15
}

Данных недостаточно для обновления кеша.

В этом случае корректно:

queryClient.invalidateQueries({
  queryKey: ["post", 15]
})

Гибридный подход

Очень распространённая архитектура:

Тип события Действие
Полные данные setQueryData
Сигнал об изменении invalidateQueries
Крупная синхронизация refetchQueries

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

Обычно WebSocket подключается внутри эффекта.

import { useEffect } from "react"

function MessagesSocket() {
  useEffect(() => {
    const socket = new WebSocket("ws://localhost:3000")

    socket.addEventListener("message", (event) => {
      const data = JSON.parse(event.data)

      queryClient.setQueryData(
        ["messages"],
        (old = []) => [...old, data]
      )
    })

    return () => {
      socket.close()
    }
  }, [])

  return null
}

Централизованный Socket Provider

В больших приложениях нельзя создавать WebSocket в каждом компоненте.

Это приводит к:

  • множественным подключениям;
  • дублированию событий;
  • утечкам памяти;
  • проблемам синхронизации.

Правильнее использовать единый provider.

Архитектура SocketProvider

import {
  createContext,
  useContext,
  useEffect,
  useRef
} from "react"

const SocketContext = createContext(null)

export function SocketProvider({ children }) {
  const socketRef = useRef(null)

  useEffect(() => {
    const socket = new WebSocket("ws://localhost:3000")

    socketRef.current = socket

    return () => {
      socket.close()
    }
  }, [])

  return (
    <SocketContext.Provider value={socketRef}>
      {children}
    </SocketContext.Provider>
  )
}

export function useSocket() {
  return useContext(SocketContext)
}

Интеграция через custom hooks

Удобно выносить обработку событий в отдельные хуки.

Пример

import { useEffect } from "react"
import { useQueryClient } from "@tanstack/react-query"

export function useMessagesSocket(socket) {
  const queryClient = useQueryClient()

  useEffect(() => {
    const handler = (event) => {
      const message = JSON.parse(event.data)

      queryClient.setQueryData(
        ["messages"],
        (old = []) => [...old, message]
      )
    }

    socket.addEventListener("message", handler)

    return () => {
      socket.removeEventListener("message", handler)
    }
  }, [socket, queryClient])
}

Частичное обновление объектов

Не всегда требуется заменять весь массив.

Обновление одного элемента

queryClient.setQueryData(
  ["posts"],
  (old = []) => {
    return old.map(post => {
      if (post.id === updatedPost.id) {
        return {
          ...post,
          ...updatedPost
        }
      }

      return post
    })
  }
)

Обновление детального запроса

Если существует запрос:

["post", id]

то обновление выглядит так:

queryClient.setQueryData(
  ["post", updatedPost.id],
  updatedPost
)

Синхронизация списка и деталей

Частая проблема — обновление только одной части кеша.

Например:

  • список постов обновился;
  • детальная страница осталась устаревшей.

Правильный подход:

queryClient.setQueryData(
  ["posts"],
  updatePosts
)

queryClient.setQueryData(
  ["post", updatedPost.id],
  updatedPost
)

Обработка удаления

Удаление записи из кеша

queryClient.setQueryData(
  ["messages"],
  (old = []) => {
    return old.filter(
      message => message.id !== deletedId
    )
  }
)

Обработка добавления

queryClient.setQueryData(
  ["notifications"],
  (old = []) => {
    return [newNotification, ...old]
  }
)

Работа с бесконечными списками

Интеграция WebSocket с infinite queries требует отдельной логики.

Структура infinite query:

{
  pages: [],
  pageParams: []
}

Обновление infinite query

queryClient.setQueryData(
  ["feed"],
  (oldData) => {
    if (!oldData) return oldData

    return {
      ...oldData,
      pages: oldData.pages.map(page => {
        return {
          ...page,
          items: page.items.map(item => {
            if (item.id === updated.id) {
              return updated
            }

            return item
          })
        }
      })
    }
  }
)

Добавление новой записи в первую страницу

queryClient.setQueryData(
  ["feed"],
  (oldData) => {
    if (!oldData) return oldData

    const firstPage = oldData.pages[0]

    return {
      ...oldData,
      pages: [
        {
          ...firstPage,
          items: [newItem, ...firstPage.items]
        },
        ...oldData.pages.slice(1)
      ]
    }
  }
)

Событийная архитектура

Практически все real-time системы используют event-driven подход.

Типичная структура

{
  "type": "MESSAGE_CREATED",
  "payload": {}
}

Router для событий

const handlers = {
  MESSAGE_CREATED: handleMessageCreated,
  MESSAGE_UPDATED: handleMessageUpdated,
  MESSAGE_DELETED: handleMessageDeleted
}

socket.addEventListener("message", (event) => {
  const data = JSON.parse(event.data)

  const handler = handlers[data.type]

  if (handler) {
    handler(data.payload)
  }
})

Дедупликация сообщений

WebSocket-системы иногда отправляют повторяющиеся события.

Без дедупликации интерфейс может содержать дубликаты.

Проверка перед вставкой

queryClient.setQueryData(
  ["messages"],
  (old = []) => {
    const exists = old.some(
      message => message.id === newMessage.id
    )

    if (exists) {
      return old
    }

    return [...old, newMessage]
  }
)

Out-of-order события

Иногда сообщения приходят в неправильном порядке.

Например:

  1. UPDATE
  2. CREATE

Это особенно часто возникает при reconnect.

Защита через timestamps

queryClient.setQueryData(
  ["posts"],
  (old = []) => {
    return old.map(post => {
      if (post.id !== updated.id) {
        return post
      }

      if (
        post.updatedAt >
        updated.updatedAt
      ) {
        return post
      }

      return updated
    })
  }
)

Reconnect логика

WebSocket-соединения могут разрываться:

  • потеря сети;
  • sleep mode;
  • смена вкладки;
  • мобильные сети;
  • рестарт сервера;
  • proxy timeout.

Автоматическое переподключение

function connect() {
  const socket = new WebSocket(
    "ws://localhost:3000"
  )

  socket.oncl ose = () => {
    setTimeout(() => {
      connect()
    }, 3000)
  }
}

Exponential Backoff

Постоянные reconnect-попытки могут перегрузить сервер.

Используется backoff:

let retries = 0

function connect() {
  const socket = new WebSocket(
    "ws://localhost:3000"
  )

  socket.oncl ose = () => {
    retries += 1

    const timeout =
      Math.min(1000 * 2 ** retries, 30000)

    setTimeout(connect, timeout)
  }

  socket.ono pen = () => {
    retries = 0
  }
}

Повторная синхронизация после reconnect

После восстановления соединения кеш может устареть.

Часто выполняется:

socket.ono pen = () => {
  queryClient.invalidateQueries()
}

selective invalidation

Иногда выгоднее обновлять только часть кеша.

queryClient.invalidateQueries({
  queryKey: ["notifications"]
})

queryClient.invalidateQueries({
  queryKey: ["messages"]
})

Heartbeat и ping/pong

Некоторые серверы закрывают неактивные соединения.

Используется heartbeat:

setInterval(() => {
  socket.send(
    JSON.stringify({
      type: "PING"
    })
  )
}, 30000)

Работа с авторизацией

Обычно токен передаётся:

  • в query params;
  • через cookie;
  • через handshake;
  • через subprotocols.

Пример передачи токена

const socket = new WebSocket(
  `ws://localhost:3000?token=${token}`
)

Обновление токена

При refresh токена может понадобиться reconnect.

authStore.subscribe(() => {
  socket.close()
  connect()
})

WebSocket и optimistic updates

TanStack Query отлично поддерживает optimistic updates.

Но при WebSocket возможен конфликт:

  1. optimistic update обновил UI;
  2. сервер отправил событие;
  3. данные дублировались.

Предотвращение дублирования

Часто используются:

  • временные client IDs;
  • correlation IDs;
  • reconciliation;
  • merge logic.

Пример clientId

const optimisticMessage = {
  id: crypto.randomUUID(),
  pending: true,
  text
}

После ответа сервера:

queryClient.setQueryData(
  ["messages"],
  (old = []) => {
    return old.map(message => {
      if (message.id === optimisticId) {
        return serverMessage
      }

      return message
    })
  }
)

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

TanStack Query позволяет подписываться на изменения кеша.

Это полезно для интеграции сложных real-time систем.

Подписка на Query Cache

const unsubscribe =
  queryClient
    .getQueryCache()
    .subscribe((event) => {
      console.log(event)
    })

Типы событий Query Cache

Можно отслеживать:

  • создание query;
  • удаление query;
  • обновление query;
  • observer changes;
  • garbage collection.

WebSocket и multi-tab синхронизация

Если приложение открыто в нескольких вкладках:

  • каждая вкладка может создать свой socket;
  • сервер получит множество подключений;
  • возрастёт нагрузка.

BroadcastChannel

Современный подход — использовать:

const channel =
  new BroadcastChannel("app")

Одна вкладка получает WebSocket-события и рассылает их остальным.

Синхронизация через broadcast

socket.onmess age = (event) => {
  const data = JSON.parse(event.data)

  channel.postMessage(data)
}

Другие вкладки:

channel.onmess age = (event) => {
  handleSocketEvent(event.data)
}

WebSocket и staleTime

При постоянных push-обновлениях часто увеличивают staleTime.

useQuery({
  queryKey: ["feed"],
  queryFn: fetchFeed,
  staleTime: Infinity
})

Причина:

  • данные приходят через socket;
  • постоянные refetch становятся ненужными.

Опасности staleTime Infinity

Полный отказ от refetch может привести к:

  • рассинхронизации;
  • пропущенным событиям;
  • stale cache после reconnect;
  • потерянным данным.

Поэтому даже с WebSocket обычно оставляют:

  • refetchOnReconnect;
  • background sync;
  • periodic refetch.

Комбинация polling и WebSocket

Некоторые системы используют гибрид:

Механизм Назначение
WebSocket Мгновенные обновления
Polling Проверка целостности
Refetch Восстановление синхронизации

Интеграция с Socket.IO

TanStack Query легко работает с Socket.IO.

Подключение

import { io } from "socket.io-client"

const socket = io("http://localhost:3000")

Подписка на события

socket.on("message_created", (message) => {
  queryClient.setQueryData(
    ["messages"],
    (old = []) => [...old, message]
  )
})

Socket.IO acknowledgements

Socket.IO поддерживает acknowledgements.

socket.emit(
  "create_message",
  payload,
  (response) => {
    console.log(response)
  }
)

WebSocket и SSR

На сервере WebSocket недоступен.

Поэтому соединение создаётся только на клиенте.

Проверка окружения

if (typeof window !== "undefined") {
  connectSocket()
}

Очистка ресурсов

Наиболее распространённая причина memory leaks:

  • забытый removeEventListener;
  • незакрытый socket;
  • множественные reconnect timers.

Безопасная очистка

useEffect(() => {
  const socket = new WebSocket(url)

  const handler = (event) => {}

  socket.addEventListener(
    "message",
    handler
  )

  return () => {
    socket.removeEventListener(
      "message",
     handler
    )

    socket.close()
  }
}, [])

Типичная production-схема

В production-приложениях интеграция обычно включает:

Компонент Назначение
TanStack Query Серверное состояние
WebSocket Real-time события
Event Router Маршрутизация событий
QueryClient Обновление кеша
Reconnect Layer Восстановление соединения
Auth Layer Авторизация
BroadcastChannel Multi-tab sync
Periodic Refetch Проверка целостности

Главные принципы интеграции

WebSocket не хранит состояние

Состояние должно храниться в Query Cache.

Query Cache — единый источник истины

Компоненты должны читать данные только из TanStack Query.

Socket-события должны быть идемпотентными

Повторное событие не должно ломать состояние.

Reconnect — обязательная часть архитектуры

Разрывы соединений происходят постоянно.

invalidateQueries — не универсальное решение

Если данные уже пришли через WebSocket, лучше обновлять кеш напрямую.

Real-time архитектура требует контроля консистентности

Необходимо учитывать:

  • порядок событий;
  • дедупликацию;
  • повторную доставку;
  • конфликт optimistic updates;
  • восстановление после reconnect;
  • multi-tab синхронизацию;
  • stale cache.
Оглавление