Синхронизация данных в реальном времени

Синхронизация данных в реальном времени — одна из наиболее сложных задач при работе с клиентским кешем. TanStack Query предоставляет инфраструктуру для управления серверным состоянием, однако библиотека не ограничивается обычными HTTP-запросами. Query Cache может обновляться из WebSocket-соединений, Server-Sent Events, polling-механизмов, Background Sync, push-событий и любых пользовательских транспортов.

Ключевая особенность TanStack Query заключается в том, что сервер остаётся единственным источником истины, а клиентский кеш выполняет роль синхронизированного представления серверного состояния.

Основные задачи real-time синхронизации:

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

Модель серверного состояния

Перед реализацией real-time синхронизации важно понимать разницу между локальным и серверным состоянием.

Локальное состояние:

const [isOpen, setIsOpen] = useState(false)

Серверное состояние:

const { data } = useQuery({
  queryKey: ['messages'],
  queryFn: fetchMessages
})

Серверное состояние обладает несколькими особенностями:

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

TanStack Query решает эти проблемы через:

  • Query Cache;
  • автоматический refetch;
  • invalidation;
  • background updates;
  • stale/fresh модели;
  • подписки на обновления.

Источники real-time обновлений

Polling

Наиболее простой способ синхронизации.

const { data } = useQuery({
  queryKey: ['notifications'],
  queryFn: fetchNotifications,
  refetchInterval: 5000
})

Каждые 5 секунд выполняется повторный запрос.

Преимущества:

  • простая реализация;
  • работает везде;
  • не требует WebSocket.

Недостатки:

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

Smart Polling

Polling можно адаптировать под состояние приложения.

const { data } = useQuery({
  queryKey: ['chat'],
  queryFn: fetchChat,
  refetchInterval: (data) => {
    if (!data?.hasNewMessages) {
      return 30000
    }

    return 3000
  }
})

Интервал изменяется динамически.

Polling только в активной вкладке

const { data } = useQuery({
  queryKey: ['feed'],
  queryFn: fetchFeed,
  refetchInterval: 10000,
  refetchIntervalInBackground: false
})

При неактивной вкладке polling останавливается.

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

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

Наиболее эффективный способ real-time обновлений — WebSocket.

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

function useChatSocket() {
  const queryClient = useQueryClient()

  useEffect(() => {
    const socket = new WebSocket('ws://localhost:3001')

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

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

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

При получении события кеш обновляется напрямую.

Инвалидация через WebSocket

Не всегда требуется обновлять кеш вручную. Часто достаточно инвалидировать запрос.

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

  if (payload.type === 'NEW_COMMENT') {
    queryClient.invalidateQueries({
      queryKey: ['comments']
    })
  }
}

После invalidation TanStack Query выполнит повторный запрос.

Такой подход особенно полезен:

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

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

Полный refetch не всегда эффективен.

Например, обновление одного элемента:

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

  queryClient.setQueryData(
    ['todos'],
    (old = []) => {
      return old.map(todo => {
        if (todo.id === updatedTodo.id) {
          return updatedTodo
        }

        return todo
      })
    }
  )
}

Преимущества:

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

Синхронизация Infinite Query

Infinite Query требует особого обновления страниц.

queryClient.setQueryData(
  ['messages'],
  (oldData) => {
    if (!oldData) {
      return oldData
    }

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

          return message
        })
      }))
    }
  }
)

Структура InfiniteData должна сохраняться полностью.

Добавление новых элементов в Infinite Query

queryClient.setQueryData(
  ['messages'],
  (oldData) => {
    if (!oldData) {
      return oldData
    }

    return {
      ...oldData,
      pages: [
        {
          ...oldData.pages[0],
          items: [newMessage, ...oldData.pages[0].items]
        },
        ...oldData.pages.slice(1)
      ]
    }
  }
)

Новые элементы обычно вставляются в первую страницу.

Server-Sent Events

SSE подходит для однонаправленного потока событий.

useEffect(() => {
  const events = new EventSource('/events')

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

    queryClient.setQueryData(
      ['stats'],
      data
    )
  }

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

SSE проще WebSocket:

  • встроенная поддержка reconnect;
  • обычный HTTP;
  • меньше накладных расходов;
  • идеально для live-дашбордов.

Background Refetch

TanStack Query автоматически синхронизирует данные при:

  • фокусе окна;
  • восстановлении соединения;
  • повторном монтировании компонентов.
const { data } = useQuery({
  queryKey: ['profile'],
  queryFn: fetchProfile,
  refetchOnWindowFocus: true,
  refetchOnReconnect: true
})

Это создаёт эффект «живого» приложения даже без WebSocket.

Управление stale состоянием

Синхронизация тесно связана с концепцией stale/fresh.

const { data } = useQuery({
  queryKey: ['dashboard'],
  queryFn: fetchDashboard,
  staleTime: 10000
})

В течение 10 секунд данные считаются свежими.

Слишком маленький staleTime:

  • увеличивает количество запросов;
  • создаёт лишние обновления.

Слишком большой:

  • приводит к устаревшему UI.

Реальное время и optimistic updates

Optimistic update позволяет мгновенно отобразить изменение до ответа сервера.

const mutation = useMutation({
  mutationFn: sendMessage,

  onMutate: async (newMessage) => {
    await queryClient.cancelQueries({
      queryKey: ['messages']
    })

    const previous = queryClient.getQueryData(['messages'])

    queryClient.setQueryData(
      ['messages'],
      (old = []) => [
        ...old,
        {
          ...newMessage,
          optimistic: true
        }
      ]
    )

    return { previous }
  },

  onError: (error, variables, context) => {
    queryClient.setQueryData(
      ['messages'],
      context.previous
    )
  },

  onSettled: () => {
    queryClient.invalidateQueries({
      queryKey: ['messages']
    })
  }
})

В real-time системах optimistic updates особенно важны, поскольку задержки становятся визуально заметными.

Дедупликация real-time событий

При использовании optimistic updates и WebSocket часто возникает дублирование.

Сценарий:

  1. сообщение добавлено оптимистично;
  2. сервер сохранил сообщение;
  3. WebSocket прислал новое событие;
  4. сообщение появилось дважды.

Решение — дедупликация.

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

    if (exists) {
      return old
    }

    return [...old, message]
  }
)

Нормализация кеша

При большом количестве связанных сущностей полезна ручная нормализация.

Плохой вариант:

['posts']
['post', id]
['feed']
['profile']

Один и тот же объект может существовать в нескольких местах.

При обновлении потребуется синхронизация всех копий.

Централизованное обновление сущностей

function updatePostEverywhere(updatedPost) {
  queryClient.setQueriesData(
    { queryKey: ['posts'] },
    (old) => {
      if (!old) {
        return old
      }

      return old.map(post => {
        if (post.id === updatedPost.id) {
          return updatedPost
        }

        return post
      })
    }
  )
}

setQueriesData обновляет несколько query одновременно.

Синхронизация между вкладками

TanStack Query поддерживает broadcast-механизм.

import { broadcastQueryClient } from '@tanstack/query-broadcast-client-experimental'

broadcastQueryClient({
  queryClient,
  broadcastChannel: 'app-cache'
})

Изменения кеша распространяются между вкладками браузера.

Это особенно полезно для:

  • уведомлений;
  • чатов;
  • онлайн-статусов;
  • админ-панелей.

Синхронизация после reconnect

WebSocket может временно отключаться.

Необходимо восстанавливать состояние.

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

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

Обработка out-of-order событий

Real-time события могут приходить в неправильном порядке.

Например:

  • обновление версии 5;
  • затем обновление версии 4.

Решение — versioning.

queryClient.setQueryData(
  ['document', doc.id],
  (old) => {
    if (!old) {
      return doc
    }

    if (doc.version < old.version) {
      return old
    }

    return doc
  }
)

Event-driven архитектура

Крупные приложения обычно используют типизированные события.

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

  switch (payload.type) {
    case 'POST_CREATED':
      handlePostCreated(payload.data)
      break

    case 'POST_UPDATED':
      handlePostUpdated(payload.data)
      break

    case 'POST_DELETED':
      handlePostDeleted(payload.data)
      break
  }
}

Такой подход:

  • упрощает масштабирование;
  • улучшает поддержку;
  • снижает связанность кода.

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

Для сложных real-time сценариев можно использовать QueryObserver.

import { QueryObserver } from '@tanstack/react-query'

const observer = new QueryObserver(
  queryClient,
  {
    queryKey: ['messages'],
    queryFn: fetchMessages
  }
)

const unsubscribe = observer.subscribe(result => {
  console.log(result.data)
})

Observer позволяет подписываться на изменения query вне React-компонентов.

Реальное время и garbage collection

По умолчанию query удаляются из кеша через некоторое время.

const queryClient = new QueryClient({
  defaultOptions: {
    queries: {
      gcTime: 1000 * 60 * 30
    }
  }
})

Для real-time приложений часто увеличивают gcTime.

Иначе кеш может очищаться слишком агрессивно.

Live-обновление списков пользователей

Пример системы присутствия пользователей:

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

  if (payload.type === 'USER_ONLINE') {
    queryClient.setQueryData(
      ['online-users'],
      (old = []) => {
        const exists = old.includes(payload.userId)

        if (exists) {
          return old
        }

        return [...old, payload.userId]
      }
    )
  }

  if (payload.type === 'USER_OFFLINE') {
    queryClient.setQueryData(
      ['online-users'],
      (old = []) => {
        return old.filter(
          id => id !== payload.userId
        )
      }
    )
  }
}

Интеграция с React Context

WebSocket-подключение обычно централизуется.

const SocketContext = createContext(null)

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

  useEffect(() => {
    socketRef.current = new WebSocket('ws://localhost:3001')

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

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

Это предотвращает создание множества соединений.

Push-based invalidation

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

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

  queryClient.invalidateQueries({
    queryKey: payload.queryKey
  })
}

Клиент самостоятельно загружает новые данные.

Такой подход:

  • уменьшает размер событий;
  • снижает связанность клиента и сервера;
  • упрощает backend.

Batch-обновления

Большое количество событий может вызывать лишние ререндеры.

import { notifyManager } from '@tanstack/react-query'

notifyManager.batch(() => {
  updates.forEach(update => {
    queryClient.setQueryData(
      ['items', update.id],
      update
    )
  })
})

Batching уменьшает нагрузку на React.

Согласованность списка и детали

Типичная проблема:

['posts']
['post', 5]

Обновляется detail query, но список остаётся старым.

Решение:

function syncPost(post) {
  queryClient.setQueryData(
    ['post', post.id],
    post
  )

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

        return item
      })
    }
  )
}

Offline-first синхронизация

TanStack Query поддерживает offline-сценарии.

const queryClient = new QueryClient({
  defaultOptions: {
    mutations: {
      networkMode: 'offlineFirst'
    }
  }
})

Mutation будут ожидать восстановления соединения.

Persisted cache

Для real-time приложений часто используется persistent cache.

persistQueryClient({
  queryClient,
  persister
})

После перезагрузки страницы данные восстанавливаются мгновенно.

Race conditions

При высокой частоте обновлений возможны гонки данных.

Например:

  1. выполняется refetch;
  2. приходит WebSocket update;
  3. refetch перезаписывает новое состояние старым ответом.

Распространённые решения:

  • versioning;
  • timestamps;
  • merge-логика;
  • event sourcing;
  • серверные revision numbers.

CRDT и collaborative editing

Для совместного редактирования документов обычного кеша недостаточно.

TanStack Query можно использовать как транспортный слой:

  • хранение snapshot;
  • синхронизация revisions;
  • background refetch;
  • reconnect logic.

Но разрешение конфликтов обычно реализуется отдельно через:

  • CRDT;
  • Operational Transformation;
  • event sourcing.

Управление частотой обновлений

Слишком частые обновления создают проблемы производительности.

Throttle:

const throttledUpdate = throttle((data) => {
  queryClient.setQueryData(
    ['chart'],
    data
  )
}, 1000)

Debounce:

const debouncedInvalidate = debounce(() => {
  queryClient.invalidateQueries({
    queryKey: ['search']
  })
}, 500)

Streaming и progressive updates

Иногда данные поступают частями.

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

  queryClient.setQueryData(
    ['stream'],
    (old = '') => old + chunk.text
  )
}

Подход используется:

  • в AI-чатах;
  • live-логах;
  • терминалах;
  • потоковой аналитике.

Диагностика real-time кеша

Для отладки используются:

  • React Query Devtools;
  • Query Cache inspection;
  • mutation tracking;
  • observer tracing;
  • network monitoring.

Полезно логировать:

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

Это помогает анализировать:

  • invalidation;
  • refetch;
  • cache updates;
  • observer lifecycle;
  • удаление query;
  • race conditions.
Оглавление