Deepdive into Tanstack/React Query - 4. 重试与网络管理

export function createRetryer<TData = unknown, TError = DefaultError>(
  config: RetryerConfig<TData, TError>,
): Retryer<TData> {
  let isRetryCancelled = false
  let failureCount = 0
  let continueFn: ((value?: unknown) => void) | undefined

  const thenable = pendingThenable<TData>()

  const isResolved = () =>
    (thenable.status as Thenable<TData>['status']) !== 'pending'

  // ... 省略具体实现

  return {
    promise: thenable,
    status: () => thenable.status,
    cancel,
    continue: () => {
      continueFn?.()
      return thenable
    },
    cancelRetry,
    continueRetry,
    canStart,
    start: () => {
      if (canStart()) {
        run()
      } else {
        pause().then(run)
      }
      return thenable
    },
  }
}

interface RetryerConfig<TData = unknown, TError = DefaultError> {
  fn: () => TData | Promise<TData>       // 要执行的异步函数
  initialPromise?: Promise<TData>        // 可复用的初始 Promise
  onCancel?: (error: TError) => void     // 取消回调
  onFail?: (failureCount: number, error: TError) => void  // 失败回调
  onPause?: () => void                   // 暂停回调
  onContinue?: () => void                // 继续回调
  retry?: RetryValue<TError>             // 重试策略
  retryDelay?: RetryDelayValue<TError>   // 重试延迟
  networkMode: NetworkMode | undefined   // 网络模式
  canRun: () => boolean                  // 是否可运行
}

export interface Retryer<TData = unknown> {
  promise: Promise<TData>                        // 最终的 Promise
  cancel: (cancelOptions?: CancelOptions) => void // 取消请求
  continue: () => Promise<unknown>               // 继续执行
  cancelRetry: () => void                        // 取消后续重试
  continueRetry: () => void                      // 恢复重试
  canStart: () => boolean                        // 是否可以开始
  start: () => Promise<TData>                    // 启动执行
  status: () => 'pending' | 'resolved' | 'rejected'  // 当前状态
}

export type RetryValue<TError> = boolean | number | ShouldRetryFunction<TError>

type ShouldRetryFunction<TError = DefaultError> = (
  failureCount: number,
  error: TError,
) => boolean

// 从 run() 函数中提取
const retry = config.retry ?? (isServer ? 0 : 3)  // 默认客户端重试3次，服务端0次

const shouldRetry =
  retry === true ||                                  // true: 无限重试
  (typeof retry === 'number' && failureCount < retry) ||  // number: 重试指定次数
  (typeof retry === 'function' && retry(failureCount, error))  // function: 自定义逻辑

export type RetryDelayValue<TError> = number | RetryDelayFunction<TError>

type RetryDelayFunction<TError = DefaultError> = (
  failureCount: number,
  error: TError,
) => number

// 默认的重试延迟函数：指数退避策略
function defaultRetryDelay(failureCount: number) {
  return Math.min(1000 * 2 ** failureCount, 30000)
}

// 失败回调 - 每次重试失败后触发
config.onFail?.(failureCount, error)

// 暂停回调 - 当请求因网络或焦点问题暂停时触发
const pause = () => {
  return new Promise((continueResolve) => {
    continueFn = (value) => {
      if (isResolved() || canContinue()) {
        continueResolve(value)
      }
    }
    config.onPause?.()  // 触发暂停回调
  }).then(() => {
    continueFn = undefined
    if (!isResolved()) {
      config.onContinue?.()  // 触发继续回调
    }
  })
}

// Query.fetch 中的使用示例
this.#retryer = createRetryer({
  // ...
  onFail: (failureCount, error) => {
    this.#dispatch({ type: 'failed', failureCount, error })
  },
  onPause: () => {
    this.#dispatch({ type: 'pause' })
  },
  onContinue: () => {
    this.#dispatch({ type: 'continue' })
  },
})

const run = () => {
  // 1. 如果已经解决，直接返回
  if (isResolved()) {
    return
  }

  let promiseOrValue: any

  // 2. 首次执行可以复用 initialPromise
  const initialPromise = failureCount === 0 ? config.initialPromise : undefined

  // 3. 执行查询函数
  try {
    promiseOrValue = initialPromise ?? config.fn()
  } catch (error) {
    promiseOrValue = Promise.reject(error)
  }

  // 4. 处理结果
  Promise.resolve(promiseOrValue)
    .then(resolve)  // 成功：解决 Promise
    .catch((error) => {
      // 5. 失败处理
      if (isResolved()) return

      // 6. 计算重试策略
      const retry = config.retry ?? (isServer ? 0 : 3)
      const retryDelay = config.retryDelay ?? defaultRetryDelay
      const delay = typeof retryDelay === 'function'
        ? retryDelay(failureCount, error)
        : retryDelay
      
      const shouldRetry =
        retry === true ||
        (typeof retry === 'number' && failureCount < retry) ||
        (typeof retry === 'function' && retry(failureCount, error))

      // 7. 不重试则直接拒绝
      if (isRetryCancelled || !shouldRetry) {
        reject(error)
        return
      }

      // 8. 增加失败计数并触发回调
      failureCount++
      config.onFail?.(failureCount, error)

      // 9. 延迟后重试
      sleep(delay)
        .then(() => {
          // 检查是否可以继续，否则暂停
          return canContinue() ? undefined : pause()
        })
        .then(() => {
          if (isRetryCancelled) {
            reject(error)
          } else {
            run()  // 递归重试
          }
        })
    })
}

export class CancelledError extends Error {
  revert?: boolean   // 是否回滚乐观更新
  silent?: boolean   // 是否静默取消（不触发错误处理）
  
  constructor(options?: CancelOptions) {
    super('CancelledError')
    this.revert = options?.revert
    this.silent = options?.silent
  }
}

const cancel = (cancelOptions?: CancelOptions): void => {
  if (!isResolved()) {
    const error = new CancelledError(cancelOptions) as TError
    reject(error)
    config.onCancel?.(error)
  }
}

// Query.fetch 方法中
const abortController = new AbortController()

// 添加取消监听
const unsubscribe = this.#cache.config.onAbort?.(queryFnContext.signal)

this.#abortSignalConsumed = false
queryFnContext.signal.addEventListener('abort', () => {
  if (!this.#abortSignalConsumed) {
    this.#retryer?.cancel({ revert: true })
  }
})

type NetworkMode = 'online' | 'always' | 'offlineFirst'

// canStart: 判断是否可以开始执行
const canStart = () => canFetch(config.networkMode) && config.canRun()

// canContinue: 判断暂停后是否可以继续
const canContinue = () =>
  focusManager.isFocused() &&
  (config.networkMode === 'always' || onlineManager.isOnline()) &&
  config.canRun()

// 基本用法
import { onlineManager } from '@tanstack/react-query'

// 检查当前网络状态
const isOnline = onlineManager.isOnline()

// 手动设置网络状态（用于测试或特殊场景）
onlineManager.setOnline(true)
onlineManager.setOnline(false)
onlineManager.setOnline(undefined)  // 恢复默认检测

// 订阅网络状态变化
const unsubscribe = onlineManager.subscribe((isOnline) => {
  console.log('Network status:', isOnline ? 'online' : 'offline')
})

// 默认使用 navigator.onLine 和 online/offline 事件
// 可以自定义监听逻辑
onlineManager.setEventListener((setOnline) => {
  // 使用自定义的网络检测逻辑
  const handleOnline = () => setOnline(true)
  const handleOffline = () => setOnline(false)
  
  window.addEventListener('online', handleOnline)
  window.addEventListener('offline', handleOffline)
  
  return () => {
    window.removeEventListener('online', handleOnline)
    window.removeEventListener('offline', handleOffline)
  }
})

// packages/query-core/src/subscribable.ts
export class Subscribable<TListener extends Function = Listener> {
  protected listeners: Set<TListener>

  constructor() {
    this.listeners = new Set()
  }

  subscribe(listener: TListener): () => void {
    this.listeners.add(listener)
    
    // 首次订阅时可能需要初始化
    this.onSubscribe()

    return () => {
      this.listeners.delete(listener)
      this.onUnsubscribe()
    }
  }

  // 子类可重写这些钩子
  protected onSubscribe(): void {}
  protected onUnsubscribe(): void {}
}

// 核心实现
class FocusManager extends Subscribable<FocusListener> {
  #focused: boolean | undefined
  #cleanup?: () => void
  #setup?: (setFocused: (focused?: boolean) => void) => (() => void) | void

  protected onSubscribe(): void {
    if (!this.#cleanup) {
      this.setEventListener(this.#setup)
    }
  }

  protected onUnsubscribe(): void {
    if (!this.listeners.size) {
      this.#cleanup?.()
      this.#cleanup = undefined
    }
  }

  setEventListener(setup?: typeof this.#setup): void {
    this.#setup = setup
    this.#cleanup?.()
    
    this.#cleanup = setup?.((focused) => {
      if (typeof focused === 'boolean') {
        this.setFocused(focused)
      } else {
        this.onFocus()
      }
    })
  }

  setFocused(focused?: boolean): void {
    const changed = this.#focused !== focused
    if (changed) {
      this.#focused = focused
      this.onFocus()
    }
  }

  onFocus(): void {
    const isFocused = this.isFocused()
    this.listeners.forEach((listener) => listener(isFocused))
  }

  isFocused(): boolean {
    // 优先使用手动设置的值
    if (typeof this.#focused === 'boolean') {
      return this.#focused
    }
    
    // 服务端默认返回 true
    if (typeof document === 'undefined') {
      return true
    }
    
    // 浏览器端检查 visibilityState
    return document.visibilityState === 'visible'
  }
}

const focusManager = new FocusManager()

// 默认监听 visibilitychange 事件
focusManager.setEventListener((handleFocus) => {
  if (typeof window !== 'undefined' && window.addEventListener) {
    window.addEventListener('visibilitychange', handleFocus, false)
  }
  
  return () => {
    window.removeEventListener('visibilitychange', handleFocus)
  }
})

class OnlineManager extends Subscribable<OnlineListener> {
  #online: boolean | undefined
  #cleanup?: () => void
  #setup?: (setOnline: (online?: boolean) => void) => (() => void) | void

  setOnline(online?: boolean): void {
    const changed = this.#online !== online
    if (changed) {
      this.#online = online
      this.onOnline()
    }
  }

  onOnline(): void {
    const isOnline = this.isOnline()
    this.listeners.forEach((listener) => listener(isOnline))
  }

  isOnline(): boolean {
    if (typeof this.#online === 'boolean') {
      return this.#online
    }
    
    // 服务端或无 navigator 时默认在线
    if (typeof navigator === 'undefined' || typeof navigator.onLine === 'undefined') {
      return true
    }
    
    return navigator.onLine
  }
}

const onlineManager = new OnlineManager()

onlineManager.setEventListener((setOnline) => {
  if (typeof window !== 'undefined' && window.addEventListener) {
    window.addEventListener('online', () => setOnline(true))
    window.addEventListener('offline', () => setOnline(false))
  }
  
  return () => {
    window.removeEventListener('online', () => setOnline(true))
    window.removeEventListener('offline', () => setOnline(false))
  }
})

import NetInfo from '@react-native-community/netinfo'
import { onlineManager } from '@tanstack/react-query'

onlineManager.setEventListener((setOnline) => {
  return NetInfo.addEventListener((state) => {
    setOnline(!!state.isConnected)
  })
})

import { AppState, Platform } from 'react-native'
import { focusManager } from '@tanstack/react-query'

focusManager.setEventListener((handleFocus) => {
  const subscription = AppState.addEventListener('change', (state) => {
    if (Platform.OS !== 'web') {
      handleFocus(state === 'active')
    }
  })
  
  return () => {
    subscription.remove()
  }
})

// QueryCache 构造函数中
class QueryCache extends Subscribable<QueryCacheListener> {
  #queries: QueryStore

  constructor(public config: QueryCacheConfig = {}) {
    super()
    this.#queries = new Map()
  }

  onFocus(): void {
    // 窗口聚焦时，通知所有 Query
    this.#queries.forEach((query) => {
      query.observers.forEach((observer) => {
        if (observer.shouldFetchOnWindowFocus()) {
          observer.fetch()
        }
      })
    })
  }

  onOnline(): void {
    // 网络恢复时，恢复暂停的请求
    this.#queries.forEach((query) => {
      query.observers.forEach((observer) => {
        if (observer.shouldFetchOnReconnect()) {
          observer.fetch()
        }
      })
    })
    // 同时恢复 Retryer
    this.#queries.forEach((query) => {
      query.retryer?.continue()
    })
  }
}

// 场景：用户在弱网环境下请求数据
const { data } = useQuery({
  queryKey: ['user', userId],
  queryFn: () => fetchUser(userId),
  retry: 3,
  retryDelay: (attempt) => Math.min(1000 * 2 ** attempt, 30000),
  networkMode: 'online',
})

// 标准 API 请求（默认）
useQuery({
  queryKey: ['data'],
  queryFn: fetchData,
  networkMode: 'online',  // 默认值
})

// 本地数据源
useQuery({
  queryKey: ['localData'],
  queryFn: getFromIndexedDB,
  networkMode: 'always',
})

// PWA 缓存优先
useQuery({
  queryKey: ['posts'],
  queryFn: fetchPosts,
  networkMode: 'offlineFirst',
})

useQuery({
  queryKey: ['data'],
  queryFn: fetchData,
  retry: (failureCount, error) => {
    // 4xx 错误不重试
    if (error instanceof HTTPError && error.status >= 400 && error.status < 500) {
      return false
    }
    // 最多重试 3 次
    return failureCount < 3
  },
  retryDelay: (attempt, error) => {
    // 429 (Rate Limit) 使用更长的延迟
    if (error instanceof HTTPError && error.status === 429) {
      return 60000  // 1 分钟
    }
    // 默认指数退避
    return Math.min(1000 * 2 ** attempt, 30000)
  },
})

// 在测试中禁用重试
const queryClient = new QueryClient({
  defaultOptions: {
    queries: {
      retry: false,
      networkMode: 'always',  // 测试中不检查网络
    },
  },
})

// 或手动控制网络状态
import { onlineManager } from '@tanstack/react-query'

beforeEach(() => {
  onlineManager.setOnline(true)
})

test('handles offline state', async () => {
  onlineManager.setOnline(false)
  // ... 测试逻辑
})