import {
  createBuilder,
  defineQueries,
  defineQuery,
  Zero as ZeroClient,
} from '@rocicorp/zero'
import {
  useConnectionState,
  useZero,
  ZeroContext,
  ZeroProvider,
} from '@rocicorp/zero/react'
import {
  createContext,
  memo,
  useContext,
  useEffect,
  useInsertionEffect,
  useLayoutEffect,
  useMemo,
  useRef,
  useState,
  type Context,
  type ReactNode,
} from 'react'

import { createPermissions } from './createPermissions'
import {
  createUseQuery,
  type QueryControlMode,
  type UseQueryHook,
} from './createUseQuery'
import { clearZeroClientData } from './helpers/clearZeroClientData'
import { createMutators } from './helpers/createMutators'
import { createEmitter, type Emitter } from './helpers/emitter'
import { getAuth } from './helpers/getAuth'
import { readHostStorageScope } from './helpers/hostStorageScope'
import { createMutationLifecycle } from './helpers/mutationLifecycle'
import { IS_SERVER_RUNTIME } from './helpers/platform'
import {
  composeRecoveryLogSink,
  isRecoverableZeroStalePokeMessage,
  makeZeroRecovery,
  type RecoveryGuardStorage,
  type ScheduleReloadContext,
  type ZeroLogPattern,
  type ZeroRecoveryDeps,
} from './helpers/recoverZeroClient'
import { observeMutation, reportMutationInvocationError } from './helpers/useMutation'
import { registerClientInstance } from './instanceRegistry'
import { getAllMutationsPermissions, getMutationsPermissions } from './modelRegistry'
import { registerQuery } from './queryRegistry'
import { resolveQuery, type PlainQueryFn } from './resolveQuery'
import { setCustomQueries } from './run'
import { getEnvironment, setAuthData, setEnvironment, setSchema } from './state'
import { getRawWhere, setEvaluatingPermission } from './where'
import { setRunner, type ZeroRunner } from './zeroRunner'
import { zql } from './zql'

import type {
  AuthData,
  GenericModels,
  GetZeroMutators,
  ZeroEvent,
  ZeroEventsEmitter,
  ZeroReconnectReasonKey,
} from './types'
import type {
  AnyQueryRegistry,
  Query,
  Row,
  Zero,
  ZeroOptions,
  Schema as ZeroSchema,
} from '@rocicorp/zero'
import type { AggregateSet } from 'orez-lite/aggregate'
import type { HttpPullLifecycleEvent, HttpPullTransport } from 'orez-lite/client'

type PreloadOptions = { ttl?: 'always' | 'never' | number | undefined }

const MAX_AUTH_REFRESH_ATTEMPTS = 3

export type GroupedQueries = Record<string, Record<string, (...args: any[]) => any>>

// controls how usePermission behaves before the server responds:
//  - 'optimistic': evaluate the permission query on the client (default)
//  - 'optimistic-deny': return false until server confirms
//  - 'optimistic-allow': return true until server confirms
export type PermissionStrategy = 'optimistic' | 'optimistic-deny' | 'optimistic-allow'

export type ZeroProviderTransport = {
  install(serverURL: string): Pick<HttpPullTransport, 'subscribeLifecycle'> | undefined
  logClassifications?: {
    benign?: readonly ZeroLogPattern[]
  }
}

export type ZeroProviderGeneration = {
  isCurrent: () => boolean
}

export type WaitForZeroOptions = {
  signal?: AbortSignal
}

export type CreateZeroClientOptions<
  Schema extends ZeroSchema,
  Models extends GenericModels,
> = {
  schema: Schema
  models: Models
  groupedQueries: GroupedQueries
  aggregates?: AggregateSet<Schema>
  permissionStrategy?: PermissionStrategy
  // repeated server acknowledgement timeouts reconnect with the existing local
  // state. one timeout remains a normal slow-server failure.
  serverAckTimeoutRecoveryThreshold?: number
  // names this client instance so multiple instances can coexist on one page.
  // each query/mutator namespace is claimed by exactly one instance, and the
  // ambient run() + the combineZeroClients facade dispatch by that claim.
  instanceName?: string
}

export type DirectQueryAdapter<Schema extends ZeroSchema> = (props: {
  DisabledContext: Context<QueryControlMode>
  customQueries: AnyQueryRegistry
  getZero: () => any
  zeroVersion: Emitter<number>
}) => UseQueryHook<Schema>

function getZeroProxyValue(instance: object, key: PropertyKey) {
  const value = Reflect.get(instance, key, instance)
  if (typeof value !== 'function') return value

  const bound = value.bind(instance)
  for (const property of Reflect.ownKeys(value)) {
    if (property === 'length' || property === 'name' || property === 'prototype') {
      continue
    }
    const descriptor = Object.getOwnPropertyDescriptor(value, property)
    if (descriptor) {
      Object.defineProperty(bound, property, descriptor)
    }
  }
  return bound
}

function createUnavailableDirectUseQuery<
  Schema extends ZeroSchema,
>(): UseQueryHook<Schema> {
  function useQueryDirect(): never {
    throw new Error(
      `[on-zero] direct queries are optional. Import createZeroClientWithDirectQueries from 'on-zero/multi' for clients used outside the innermost ZeroProvider.`
    )
  }

  return useQueryDirect as UseQueryHook<Schema>
}

export function createZeroClient<Schema extends ZeroSchema, Models extends GenericModels>(
  options: CreateZeroClientOptions<Schema, Models>
) {
  return createZeroClientInternal(options)
}

export function createZeroClientInternal<
  Schema extends ZeroSchema,
  Models extends GenericModels,
>({
  schema,
  models,
  groupedQueries,
  aggregates,
  permissionStrategy = 'optimistic',
  instanceName = 'default',
  serverAckTimeoutRecoveryThreshold = 2,
  createDirectUseQuery,
}: CreateZeroClientOptions<Schema, Models> & {
  createDirectUseQuery?: DirectQueryAdapter<Schema>
}) {
  type ZeroMutators = GetZeroMutators<Models>
  type ZeroInstance = Zero<Schema, ZeroMutators>
  type TableName = keyof Schema['tables'] & string

  setSchema(schema, createBuilder(schema))

  // only set environment to 'client' if server hasn't already claimed it
  // server bindings may set this first during SSR
  if (getEnvironment() === null) {
    setEnvironment('client')
  }

  const permissionsHelpers = createPermissions<Schema>({
    schema,
    environment: 'client',
  })

  // build query registry from grouped queries
  // this creates ONE shared defineQueries registry that matches the server's structure
  const wrappedNamespaces: Record<
    string,
    Record<string, ReturnType<typeof defineQuery>>
  > = {}

  for (const [namespace, queries] of Object.entries(groupedQueries)) {
    wrappedNamespaces[namespace] = {}
    for (const [name, fn] of Object.entries(queries)) {
      registerQuery(fn, `${namespace}.${name}`)
      // wrap each plain function in defineQuery
      wrappedNamespaces[namespace][name] = defineQuery(({ args }: { args: any }) =>
        fn(args)
      )
    }
  }

  // register per-model permission queries so each table gets its own materialized view
  // client: evaluates raw permission condition for optimistic result
  // server: evaluates real permission condition authoritatively
  const permissionCheckFns: Record<
    string,
    (args: { objOrId: string | Record<string, any> }) => any
  > = {}

  const createPermissionCheckFn = (table: string) => {
    const fn = (args: { objOrId: string | Record<string, any> }) => {
      const perm = getMutationsPermissions(table)
      const base = (zql as any)[table]

      if (!args.objOrId) {
        return base.where((eb: any) => eb.cmpLit(true, '=', false)).one()
      }

      if (permissionStrategy === 'optimistic') {
        // unwrap serverWhere so conditions actually evaluate on client
        // set flag so nested serverWhere calls also bypass the client no-op
        const rawPerm = perm ? getRawWhere(perm) || perm : perm
        return base
          .where((eb: any) => {
            setEvaluatingPermission(true)
            try {
              return permissionsHelpers.buildPermissionQuery(
                getAuth(),
                eb,
                rawPerm || ((e: any) => e.and()),
                args.objOrId,
                table
              )
            } finally {
              setEvaluatingPermission(false)
            }
          })
          .one()
      }

      if (permissionStrategy === 'optimistic-deny') {
        // client query always returns false, server corrects authoritatively
        return base.where((eb: any) => eb.cmpLit(true, '=', false)).one()
      }

      // optimistic-allow: pass wrapped perm directly
      // serverWhere is a no-op on client → eb.and() → always true → row exists check
      // server evaluates real condition and corrects authoritatively
      return base
        .where((eb: any) => {
          return permissionsHelpers.buildPermissionQuery(
            getAuth(),
            eb,
            perm || ((e: any) => e.and()),
            args.objOrId,
            table
          )
        })
        .one()
    }
    permissionCheckFns[table] = fn
    registerQuery(fn, `permission.${table}`)
    return fn
  }

  wrappedNamespaces['permission'] = {}
  for (const [table] of getAllMutationsPermissions()) {
    const fn = createPermissionCheckFn(table)
    wrappedNamespaces['permission'][table] = defineQuery(({ args }: any) => fn(args))
  }

  // create the single shared CustomQuery registry
  const customQueries = defineQueries(wrappedNamespaces)

  // claim this instance's query/mutator namespaces so the ambient run() and
  // the combineZeroClients facade dispatch to the owning instance. the
  // auto-generated 'permission' namespace stays unclaimed (per-instance).
  const instance = registerClientInstance({
    name: instanceName,
    namespaces: Object.keys(models),
    customQueries,
    queryNames: Object.entries(groupedQueries).flatMap(([namespace, queries]) =>
      Object.keys(queries).map((name) => `${namespace}.${name}`)
    ),
  })
  const zeroRuntime = instance.runtime as typeof instance.runtime & {
    zero: ZeroInstance | null
    readyWaiters: Set<(instance: ZeroInstance) => void>
  }

  // register for global run() helper
  setCustomQueries(customQueries)

  const DisabledContext = createContext<QueryControlMode>(false)
  const ZeroProviderGenerationContext = createContext<ZeroProviderGeneration | null>(null)
  const zeroProviderGenerations = new WeakMap<object, ZeroProviderGeneration>()

  function useZeroProviderGeneration(): ZeroProviderGeneration | null {
    return useContext(ZeroProviderGenerationContext)
  }

  function getZeroProviderGeneration(zeroInstance: ZeroInstance): ZeroProviderGeneration {
    const existing = zeroProviderGenerations.get(zeroInstance)
    if (existing) return existing
    const generation = {
      isCurrent: () => zeroRuntime.zero === zeroInstance,
    }
    zeroProviderGenerations.set(zeroInstance, generation)
    return generation
  }

  // mutators never vary per mount: auth is read dynamically through
  // getAuthData() at mutation time. built once, lazily, so the provider and
  // connectHeadless construct identical instances without either one forcing
  // the work at createZeroClient() time.
  let clientMutators: ReturnType<typeof createMutators> | null = null
  const installedTransports = new WeakMap<
    ZeroInstance,
    Pick<HttpPullTransport, 'subscribeLifecycle'>
  >()
  function getClientMutators() {
    clientMutators ??= createMutators({
      models,
      environment: 'client',
      authData: null,
      bindCan: permissionsHelpers.bindCan,
      aggregates,
    })
    return clientMutators
  }

  type ConstructZeroInstanceArgs = {
    options: Omit<ZeroOptions<Schema, ZeroMutators>, 'schema' | 'mutators'>
    transport?: ZeroProviderTransport
    beforeReload?: () => Promise<void>
    scheduleReload?: (ctx: ScheduleReloadContext) => void
    guardStorage?: RecoveryGuardStorage
    benignLogPatterns?: readonly ZeroLogPattern[]
    recoverInPlace?: () => Promise<boolean>
    reload?: () => void | boolean
  }

  // build a Zero instance with on-zero's recovery wiring. shared by the
  // provider's rotation effect and by connectHeadless so a non-react host gets
  // the same instance the app gets, rather than a parallel construction that
  // can drift.
  function constructZeroInstance({
    options,
    transport,
    beforeReload,
    scheduleReload,
    guardStorage,
    benignLogPatterns,
    recoverInPlace,
    reload,
  }: ConstructZeroInstanceArgs): ZeroInstance {
    let installedTransport: Pick<HttpPullTransport, 'subscribeLifecycle'> | undefined
    // install before construction so the instance's first connect goes through
    // HTTP. ensureHttpPullTransport is per-origin idempotent by design (a
    // rotation would otherwise chain shims), so repeat calls reuse.
    if (transport) {
      // same precedence as zero's own getServer (cacheURL is the current
      // option name; server is its deprecated alias)
      const serverURL = options.cacheURL ?? options.server
      if (typeof serverURL !== 'string') {
        throw new Error(`client transport requires a server URL`)
      }
      installedTransport = transport.install(serverURL)
    }
    // recovery closures reach the instance through this ref so they always
    // delete the CURRENT instance's own store (set right after construction;
    // the handlers only fire post-mount).
    const instanceRef: { current: ZeroInstance | null } = { current: null }
    const recoveryDeps: ZeroRecoveryDeps = {
      deleteLocalState: () => deleteZeroInstance(instanceRef.current),
      zeroEvents,
      beforeReload,
      scheduleReload,
      guardStorage,
      benignLogPatterns: [
        ...(transport?.logClassifications?.benign ?? []),
        ...(benignLogPatterns ?? []),
      ],
      onRecovery: () => mutationLifecycle.fence(),
      recoverInPlace: recoverInPlace ?? (() => remint({ dropLocalState: false })),
      reload,
    }
    const recovery = makeZeroRecovery(recoveryDeps)
    const createdInstance = new ZeroClient<Schema, ZeroMutators>({
      kvStore: 'mem',
      ...options,
      schema,
      // @ts-expect-error same erasure ZeroProvider needed
      mutators: getClientMutators(),
      // when the consumer brings no logSink, install ours: it preserves Zero's
      // console output AND watches for the local-store-lost signature. a
      // consumer with its own logSink owns log-based recovery (no double-fire
      // with e.g. soot's origin-gated recovery).
      logSink: options.logSink ?? composeRecoveryLogSink(recoveryDeps),
      // consumer handlers win; otherwise on-zero's default self-healing
      // recovery covers EVERY reason (drop local state + reload, guarded) —
      // passing these to Zero disables its built-in reload, so any reason we
      // left unhandled would fatal-blank the app forever.
      onUpdateNeeded: options.onUpdateNeeded ?? recovery.onUpdateNeeded,
      onClientStateNotFound:
        options.onClientStateNotFound ?? recovery.onClientStateNotFound,
    })
    instanceRef.current = createdInstance
    if (installedTransport) installedTransports.set(createdInstance, installedTransport)
    return createdInstance
  }

  // publish the active instance through the stable facade and query runner.
  // the provider calls this during render (before descendant effects do
  // imperative work) and connectHeadless calls it directly — the `zero` proxy,
  // run(), and waitForZero() resolve identically either way.
  function publishZeroInstance(zeroInstance: ZeroInstance): boolean {
    if (zeroInstance === zeroRuntime.zero) return false
    // retiring the outgoing client and activating the replacement is one step:
    // a write queued against the client being replaced must not land on its
    // replacement, and fencing from a separate effect let the two orders drift.
    if (zeroRuntime.zero) mutationLifecycle.fence()
    zeroRuntime.zero = zeroInstance
    mutationLifecycle.activate()
    const runner: ZeroRunner = (query, options) => zeroInstance.run(query as any, options)
    // the instance-keyed runner is what run() dispatches owned namespaces to;
    // the global runner stays as the ambient fallback (inline zql)
    instance.runner = runner
    setRunner(runner)
    const waiters = [...zeroRuntime.readyWaiters]
    zeroRuntime.readyWaiters.clear()
    for (const onReady of waiters) onReady(zeroInstance)
    return true
  }

  function waitForZero({ signal }: WaitForZeroOptions = {}): Promise<ZeroInstance> {
    if (zeroRuntime.zero) return Promise.resolve(zeroRuntime.zero)
    if (signal?.aborted) {
      return Promise.reject(
        signal.reason ?? new Error('Waiting for the Zero instance was aborted')
      )
    }

    return new Promise((resolve, reject) => {
      const onReady = (instance: ZeroInstance) => {
        signal?.removeEventListener('abort', onAbort)
        resolve(instance)
      }
      const onAbort = () => {
        zeroRuntime.readyWaiters.delete(onReady)
        reject(signal?.reason ?? new Error('Waiting for the Zero instance was aborted'))
      }
      zeroRuntime.readyWaiters.add(onReady)
      signal?.addEventListener('abort', onAbort, { once: true })
    })
  }

  // the documented `useMutation(zero.mutate.x.y)` pattern dereferences
  // `mutate` wherever it is written, including above the provider and in
  // module scope where no instance exists yet — hand back a lazy path that
  // resolves the live instance at CALL time instead of throwing at property
  // access. a call that fires with still-no instance throws the same error, so
  // real misuse stays loud.
  function lazyMutatePath(path: string[]): any {
    const resolve = () => {
      if (zeroRuntime.zero === null) {
        throw new Error(
          `Zero instance not initialized. Ensure ZeroProvider is mounted before accessing 'zero'.`
        )
      }
      let target: any = zeroRuntime.zero.mutate
      for (const key of path) {
        if (target == null) break
        target = target[key]
      }
      if (typeof target !== 'function') {
        const label = path.length > 0 ? path.join('.') : '<callable mutation>'
        throw new Error(
          `[on-zero] mutation '${label}' is not registered on the active Zero client.`
        )
      }
      return target
    }
    return new Proxy(function lazyMutator() {} as any, {
      get(_, key) {
        if (typeof key === 'symbol') return undefined
        return lazyMutatePath([...path, key])
      },
      apply(_, __, args) {
        // a queued background write pins the instance it was queued against;
        // this throws StaleGenerationError (before the catch below, so it is
        // never reported as a mutation failure) when that instance is gone.
        mutationLifecycle.assertWritable()
        try {
          const result = resolve()(...args)
          mutationLifecycle.claimMutation(result)
          void observeMutation(result)
          return result
        } catch (error) {
          reportMutationInvocationError(error)
          throw error
        }
      },
    })
  }

  // Proxy allows swapping the Zero instance on login without breaking existing references.
  // Ideally rocicorp/zero would support .setAuth() natively, but for now we swap instances.
  const zero: ZeroInstance = new Proxy({} as never, {
    get(_, key) {
      // Always resolve mutation paths at call time. Besides surviving auth
      // rotations, this is the one boundary that can observe raw
      // fire-and-forget calls before callers discard Zero's result promises.
      if (key === 'mutate') return lazyMutatePath([])
      // `zero` is a module export, and tooling reads identity properties off
      // module exports without ever intending to touch Zero. metro's fast
      // refresh registers every export as a family and then reads `.prototype`
      // on the next hot update (registerExportsForReactRefresh ->
      // canPreserveStateBetween -> isReactClass); symbol keys arrive the same
      // way from Object.prototype.toString, structuredClone and console
      // inspection. those reads land in the ordinary window where the provider
      // has not created the instance yet, and throwing there aborted the whole
      // refresh pass and dropped every pending update. no such key can be Zero
      // API, so answer them without needing an instance; real API access below
      // still throws loudly.
      if (typeof key === 'symbol' || key === 'prototype') return undefined
      if (zeroRuntime.zero === null) {
        throw new Error(
          `Zero instance not initialized. Ensure ZeroProvider is mounted before accessing 'zero'.`
        )
      }
      if (key === 'delete') {
        const instanceToDelete = zeroRuntime.zero
        return () => deleteZeroInstance(instanceToDelete)
      }
      return getZeroProxyValue(zeroRuntime.zero, key)
    },
  })

  // emitter names are global keys (dev hmr cache) — scope them per instance
  // so two instances never share cached values. the default name stays
  // unchanged for single-instance back-compat.
  const emitterScope = instanceName === 'default' ? '' : `:${instanceName}`

  const zeroEvents: ZeroEventsEmitter = createEmitter<ZeroEvent | null>(
    `zero${emitterScope}`,
    null
  )

  const ackTimeoutRecoveryThreshold =
    Number.isFinite(serverAckTimeoutRecoveryThreshold) &&
    serverAckTimeoutRecoveryThreshold >= 2
      ? Math.floor(serverAckTimeoutRecoveryThreshold)
      : 2
  const mutationLifecycle = createMutationLifecycle({
    ackTimeoutRecoveryThreshold,
    recoverFromAckTimeout: (input) => {
      void reconnectInPlace(
        'server-ack-timeout',
        `${input.label} server acknowledgement timed out ${input.consecutiveTimeouts} consecutive times (${input.timeoutMs}ms each)`
      )
    },
  })
  if (zeroRuntime.zero) mutationLifecycle.activate()

  const zeroInstanceVersion = createDirectUseQuery
    ? createEmitter<number>(`zero-instance-version${emitterScope}`, 0)
    : null

  const AuthDataContext = createContext<AuthData>({} as AuthData)

  const useQuery = createUseQuery<Schema>({
    DisabledContext,
    customQueries,
  })

  const useQueryDirect = createDirectUseQuery
    ? createDirectUseQuery({
        DisabledContext,
        customQueries,
        getZero: () => zeroRuntime.zero,
        zeroVersion: zeroInstanceVersion!,
      })
    : createUnavailableDirectUseQuery<Schema>()

  // permission check uses a per-model synced query so server is authoritative
  // permissionStrategy controls client behavior before server responds.
  // built over a query hook so the facade can route permission checks down
  // the same context vs direct path as the table's other queries.
  // SSG: return an inert hook — see createUseQuery for rationale. checking
  // here (factory-time) instead of per-call keeps hook order stable so
  // rules-of-hooks stays happy.
  const createUsePermission = (useQueryImpl: UseQueryHook<Schema>) => {
    if (IS_SERVER_RUNTIME) {
      return (() => null) as (
        table: TableName | (string & {}),
        objOrId: string | Partial<Row<any>> | undefined,
        enabled?: boolean,
        debug?: boolean
      ) => boolean | null
    }
    return function usePermission(
      table: TableName | (string & {}),
      objOrId: string | Partial<Row<any>> | undefined,
      enabled = typeof objOrId !== 'undefined',
      debug = false
    ): boolean | null {
      const disableMode = useContext(DisabledContext)
      const lastRef = useRef<boolean | null>(null)
      const tableStr = table as string
      const checkFn = permissionCheckFns[tableStr]

      // include auth user ID in query args so zero-cache creates per-user
      // permission views (prevents dedup across different auth contexts)
      const auth = getAuth()
      const _uid = auth?.id || 'anon'

      const [data, status] = useQueryImpl(
        checkFn as any,
        { objOrId: objOrId as any, _uid },
        { enabled: Boolean(!disableMode && enabled && objOrId && checkFn) }
      )

      if (debug) {
        console.info(`usePermission()`, { table, objOrId, data, status })
      }

      if (!objOrId) return false

      // null while loading, then server's authoritative answer
      const result = status.type === 'unknown' ? null : Boolean(data)

      if (!disableMode) {
        lastRef.current = result
        return result
      }

      if (disableMode === 'last-value') {
        return lastRef.current
      }

      return null
    }
  }

  const usePermission = createUsePermission(useQuery)
  const usePermissionDirect = createUsePermission(useQueryDirect)

  // the zero instance lives OUTSIDE the react lifecycle. react destroys and
  // re-fires the effects of a committed tree on any suspense hide/reveal, and
  // consumers also remount the provider (splash -> IDE). when ZeroProvider
  // created zero inside its own effect, every such cycle closed the live
  // instance mid-connect ("Failed to connect / Store is closed") and built a
  // replacement, killing in-flight queries and preloads. instead the instance
  // is created during the provider's render, cached here per identity key,
  // handed to ZeroProvider as an external `zero` (which it never closes), and
  // the previous instance is closed only when the key truly changes — a real
  // identity change (user, storage, server, logged in/out), never a react
  // lifecycle artifact. single-provider assumption: one mounted ProvideZero
  // per client (true of every consumer); two simultaneously mounted providers
  // with different identities would thrash this slot.
  // `auth` is the token the instance last connected with, so a provider that
  // unmounted and remounted around a sign-in can still tell the token changed.
  type CachedZeroEntry = {
    key: string
    instance: ZeroInstance
    auth?: string | null
  }
  let cachedZero: CachedZeroEntry | null = null

  // instances the render phase displaced, waiting for a commit to close them.
  // react can throw a render away, so a rotation is only PROVISIONAL until
  // something commits: closing there would kill an instance the committed tree
  // still holds. keyed so a render that swings back to a retired identity
  // revives that instance instead of constructing a twin beside it.
  const retiredZero = new Map<string, CachedZeroEntry>()

  // in-place re-mint: drop the current instance's local state then reconstruct a
  // fresh client WITHOUT a page reload — the native-safe recovery path (a reload
  // may never land on prod native, wedging the module latch). the mounted
  // provider registers a bump() that changes its instanceKey; remint() drives it
  // through the same rotate effect a real identity change uses. guarded in-memory
  // (Hermes has no sessionStorage) so a client-not-found storm can't reconstruct
  // in a tight loop.
  const REMINT_GUARD_MS = 12_000
  const REMINT_MAX_ATTEMPTS = 5
  const REMINT_ATTEMPT_RESET_MS = 60_000
  const remintControl: { bump: (() => void) | null } = { bump: null }
  let lastRemintAt = 0
  let remintAttempts = 0

  function unpublishZeroInstance(instanceToInvalidate: ZeroInstance): boolean {
    if (zeroRuntime.zero !== instanceToInvalidate) return false
    zeroRuntime.zero = null
    instance.runner = null
    setRunner(null)
    return true
  }

  function clearZeroInstanceReferences(instanceToInvalidate: ZeroInstance): boolean {
    if (cachedZero?.instance === instanceToInvalidate) {
      cachedZero = null
    }
    return unpublishZeroInstance(instanceToInvalidate)
  }

  function invalidateZeroInstance(instanceToInvalidate: ZeroInstance | null): void {
    if (!instanceToInvalidate) return

    mutationLifecycle.fence()
    if (clearZeroInstanceReferences(instanceToInvalidate)) {
      zeroInstanceVersion?.emit(zeroInstanceVersion.value + 1)
    }
    try {
      instanceToInvalidate.close()
    } catch {
      // a deleted client is already unusable; close is best-effort cleanup.
    }
  }

  async function deleteZeroInstance(
    instanceToDelete: ZeroInstance | null
  ): Promise<unknown> {
    mutationLifecycle.fence()
    try {
      return await instanceToDelete?.delete()
    } finally {
      invalidateZeroInstance(instanceToDelete)
    }
  }

  // supported in-place recovery: reconstruct a fresh Zero client without a page
  // reload. by default drops the current instance's local store first (a
  // ClientNotFound / desync means it's unusable), then bumps the provider's
  // instanceKey so the rotate effect mints a clean client. returns false when
  // suppressed by the rate guard or when no provider is mounted.
  async function remint(opts: { dropLocalState?: boolean } = {}): Promise<boolean> {
    // no mounted provider to reconstruct through — bail BEFORE the guard so an
    // unmounted call doesn't burn an attempt or start the cooldown.
    if (!remintControl.bump) return false
    const now = Date.now()
    const sinceLast = now - lastRemintAt
    if (lastRemintAt > 0 && sinceLast < REMINT_GUARD_MS) return false
    if (sinceLast > REMINT_ATTEMPT_RESET_MS) remintAttempts = 0
    if (remintAttempts >= REMINT_MAX_ATTEMPTS) return false
    lastRemintAt = now
    remintAttempts += 1

    const { dropLocalState = true } = opts
    if (dropLocalState && zeroRuntime.zero) {
      await deleteZeroInstance(zeroRuntime.zero).catch(() => {})
    }
    // re-check: the provider may have unmounted during the async delete.
    const bump = remintControl.bump
    if (!bump) return false
    bump()
    return true
  }

  // sign-out teardown: drop every instance this client holds — live, cached, and
  // render-parked — then mint a clean one through the mounted provider.
  // deliberately NOT rate-guarded the way remint() is: remint recovers a
  // desynced client and must not storm, while a sign-out has to land every
  // time. two sign-outs inside remint's guard window would otherwise leave the
  // previous account's local store cached for the next sign-in to revive, and
  // account data must never survive into the wrong account. the recovery budget
  // resets too, so a new session does not inherit the old one's attempts.
  async function retire(): Promise<void> {
    const parked = [...retiredZero.values()]
    retiredZero.clear()
    const live = zeroRuntime.zero ?? cachedZero?.instance ?? null
    cachedZero = null
    lastRemintAt = 0
    remintAttempts = 0
    await deleteZeroInstance(live).catch(() => {})
    // parked instances were never published, so they only need their local
    // store dropped and their socket closed.
    for (const { instance: parkedInstance } of parked) {
      if (parkedInstance === live) continue
      try {
        await parkedInstance.delete()
      } catch {
        // a deleted client is already unusable; this is best-effort cleanup.
      }
      try {
        parkedInstance.close()
      } catch {
        // same — close cannot fail a sign-out.
      }
    }
    remintControl.bump?.()
  }

  function emitReconnectStatus(event: Extract<ZeroEvent, { type: 'reconnect' }>): void {
    const current = zeroEvents.value
    if (
      current?.type === 'reconnect' &&
      current.status === event.status &&
      ('reasonKey' in current ? current.reasonKey : undefined) ===
        ('reasonKey' in event ? event.reasonKey : undefined) &&
      ('reason' in current ? current.reason : undefined) ===
        ('reason' in event ? event.reason : undefined)
    ) {
      return
    }
    zeroEvents.emit(event)
  }

  async function reconnectInPlace(
    reasonKey: ZeroReconnectReasonKey,
    reason: string
  ): Promise<boolean> {
    emitReconnectStatus({ type: 'reconnect', status: 'trying', reasonKey, reason })
    return remint({ dropLocalState: false })
  }

  function reloadPage(): boolean {
    const location = globalThis.location
    if (!location?.reload) return false
    location.reload()
    return true
  }

  // when ProvideZero is rendered without a real Zero instance (SSG, disable=true,
  // or transiently while the active path is still creating its first instance),
  // we want descendants' useZero() / useConnectionState() / on-zero useQuery to
  // NOT throw — but also not run real queries. Hand them a stub Zero plus
  // DisabledContext='empty':
  //   1. zero/react's useZero() reads ZeroContext; the stub is truthy so it
  //      doesn't throw "useZero must be used within a ZeroProvider".
  //   2. on-zero's useQuery wrapper forces enabled=false to the underlying
  //      zero useQuery when DisabledContext is set, so zero's viewStore.getView
  //      returns its disabled-view stub without ever reading zero.clientID or
  //      subscribing through the stub.
  //   3. addContextToQuery(query, zero.context) is the only deep access in the
  //      query path; the stub's .context is a plain object so that call
  //      succeeds harmlessly.
  //   4. useConnectionState reads zero.connection.state.{subscribe,current};
  //      we provide a perma-'closed' state with a no-op subscribe. consumers
  //      handle 'closed' as a normal disconnected state.
  // This stub lets the provider tree render with stable shape regardless of
  // whether Zero is active — so children never re-parent across enable/disable.
  const DISABLED_ZERO_STUB_CONNECTION_STATE = {
    current: { name: 'closed' as const },
    subscribe: () => () => {},
  }
  const DISABLED_ZERO_STUB = {
    clientID: 'disabled',
    context: {},
    connection: { state: DISABLED_ZERO_STUB_CONNECTION_STATE },
    materialize: () => ({
      addListener: () => {},
      destroy: () => {},
      updateTTL: () => {},
    }),
  } as unknown as ZeroInstance

  type ProvideZeroProps = Omit<
    ZeroOptions<Schema, ZeroMutators>,
    'schema' | 'mutators'
  > & {
    children: ReactNode
    authData?: AuthData | null
    // when true, ProvideZero renders a stable shell with stub Zero — no real
    // client is created, no websocket is opened, no IDB store is touched.
    // useQuery descendants receive EMPTY_RESPONSE via DisabledContext='empty'.
    // toggling this on/off NEVER re-parents children: the React tree shape is
    // identical in both modes (active just mounts a sibling lifecycle, which
    // doesn't shift children's position). use this when consumers need the
    // provider tree mounted (e.g. inside a marketing splash that lazily
    // upgrades to the real IDE) without paying for Zero until activation.
    disable?: boolean
    // install a client transport before constructing Zero so its first sync
    // connection uses the transport supplied by the application.
    transport?: ZeroProviderTransport
    // awaited before a self-healing recovery reload — e.g. wait for the dev
    // origin to be reachable so the reload doesn't hit a restarting server.
    beforeReload?: () => Promise<void>
    // take over WHEN/HOW the recovery reload happens (IDE gate + countdown,
    // native expo-updates reload, …) while still driving the same
    // deletes-then-reload work via ctx.performReload. default: immediate reload.
    scheduleReload?: (ctx: ScheduleReloadContext) => void
    // cross-reload loop-guard backing store; defaults to sessionStorage on web.
    // inject a native KV so Hermes gets real cross-reload protection.
    guardStorage?: RecoveryGuardStorage
    // expected recovery signatures supplied as data. transport patterns and app
    // patterns are combined; neither replaces the built-in classification.
    benignLogPatterns?: readonly ZeroLogPattern[]
    // called when the connection needs auth; return a fresh token to reconnect
    // in place. lets an expired token auto-recover without a reload.
    refreshAuth?: () => Promise<string | undefined>
    // when true, mirror this instance's connection state onto
    // document.body.dataset.zero* for e2e/diagnostics. enable on ONE instance
    // (the control/primary) so multiple instances don't fight over the dataset.
    connectionDataset?: boolean
  }

  // providezero keeps the same provider/fiber layout on ssr and client so
  // useId() stays stable across hydration. the server implementation has no
  // hooks because the ssg runtime can load on-zero through a non-deduped react
  // copy whose dispatcher is null; the client implementation always calls its
  // hooks, regardless of `disable`.
  const ProvideZeroServer = ({ children, authData: authDataIn }: ProvideZeroProps) => {
    return (
      <AuthDataContext.Provider value={(authDataIn ?? {}) as AuthData}>
        <DisabledContext.Provider value="empty">
          <ZeroContext.Provider value={DISABLED_ZERO_STUB as any}>
            <ZeroProviderGenerationContext.Provider value={null}>
              {/* match the active path's 3-child layout exactly so descendant
                  useId() lands at the same fiber index in both branches. the
                  two leading nulls reserve the SetZeroInstance + ConnectionMonitor
                  slots; the active path puts those components there only once
                  an instance exists. without these placeholder slots, children
                  would sit at child index 0 here but index 2 in the active
                  path, shifting every descendant useId. */}
              {null}
              {null}
              {children}
            </ZeroProviderGenerationContext.Provider>
          </ZeroContext.Provider>
        </DisabledContext.Provider>
      </AuthDataContext.Provider>
    )
  }

  const ProvideZeroClient = ({
    children,
    authData: authDataIn,
    transport,
    beforeReload,
    scheduleReload,
    guardStorage,
    benignLogPatterns,
    refreshAuth,
    connectionDataset,
    disable,
    ...props
  }: ProvideZeroProps) => {
    // resolve the auth token first: a real logout (token gone) must clear
    // authData so client mutators don't keep running as the old user, while a
    // transient authData blip with the token still present (session refresh, tab
    // wake) keeps the last value so mutations never see null mid-transition.
    const auth = 'auth' in props ? (props as { auth?: string | null }).auth : undefined
    const hasAuth = typeof auth === 'string'

    // stabilize authData across transient gaps, but ONLY while authed — bakes in
    // what consumers hand-rolled with a ref, and additionally clears on logout
    // (the bare ref pattern does not, leaving mutators running as the old user
    // until the instance rotates).
    const stableAuthDataRef = useRef<AuthData | null>(authDataIn ?? null)
    if (authDataIn) {
      stableAuthDataRef.current = authDataIn
    } else if (!hasAuth) {
      stableAuthDataRef.current = null
    }
    const authData = (authDataIn ?? stableAuthDataRef.current ?? null) as AuthData

    // update global authData synchronously during render so mutations always have latest auth
    // (mutations read auth dynamically via getAuthData() to avoid stale closure race condition)
    setAuthData(authData)

    // host-scoped storage: composed here so embedding hosts isolate co-located
    // apps without app code carrying host globals. static per page load — the
    // host injects the scope before the app's module graph evaluates.
    const hostScope = readHostStorageScope()
    const scopedProps = hostScope
      ? { ...props, storageKey: `${hostScope}-${props.storageKey ?? 'zero'}` }
      : props

    // expose the client proxy for harnesses/tests — set from the mounted
    // provider (not module scope) so hot-reloaded trees always point at the
    // live client.
    ;(globalThis as { __testZero?: unknown }).__testZero = zero

    // `?reset` on any URL clears local zero state, then reloads without the
    // param — a universal escape hatch for corrupt local data.
    useEffect(() => {
      if (typeof window === 'undefined' || typeof window.location === 'undefined') return
      if (!new URLSearchParams(window.location.search).has('reset')) return
      const url = new URL(window.location.href)
      url.searchParams.delete('reset')
      void clearZeroClientData({
        closeZero: async () => zero.close(),
        reload: false,
      }).then(() => {
        window.location.replace(url.toString())
      })
    }, [])

    // remint() reconstructs the client in place by bumping this counter, which
    // changes instanceKey and drives the rotate effect exactly as a real
    // identity change does. register the bump so the imperative remint() API can
    // reach this mounted provider.
    const [remintGeneration, setRemintGeneration] = useState(0)
    useEffect(() => {
      remintControl.bump = () => setRemintGeneration((generation) => generation + 1)
      return () => {
        remintControl.bump = null
      }
    }, [])

    // identity = every primitive option except the auth token value (token
    // changes refresh in place below; logged-in <-> logged-out still rotates
    // via hasAuth, matching zero's documented provider semantics). function
    // props (callbacks, logSink, batchViewUpdates) are bound at construction.
    // remintGeneration is included so remint() forces a fresh instance.
    const instanceKey = JSON.stringify([
      Object.entries({ kvStore: 'mem', ...scopedProps })
        .filter(
          ([key, value]) =>
            key !== 'auth' && typeof value !== 'function' && value !== undefined
        )
        .sort(([a], [b]) => (a < b ? -1 : 1)),
      hasAuth,
      transport,
      benignLogPatterns?.map((pattern) =>
        typeof pattern === 'string' ? pattern : `/${pattern.source}/${pattern.flags}`
      ),
      remintGeneration,
    ])

    // create/rotate DURING RENDER. a consumer that reads `zero`, or a
    // transport that has to be installed before the first connect, has to find
    // a live client in its very first render — an effect is one tick too late,
    // and in any environment where passive effects are delayed, discarded, or
    // never flushed (workers, double-rendered roots, concurrent react) that
    // tick may never arrive at all. idempotent by instanceKey, so strictmode's
    // double-invoked render and a suspense hide/reveal both hit the cache: no
    // churn, no second client.
    //
    // disable=true creates nothing. an identity change while disabled still
    // retires the old instance: logout commonly removes auth and disables in
    // the same render, and keeping the authenticated key cached would revive
    // its already-closed connection when the next session signs in. a plain
    // disable with the same identity remains a gate and keeps its warm cache.
    let liveInstance: ZeroInstance | undefined
    if (disable && cachedZero && cachedZero.key !== instanceKey) {
      retiredZero.set(cachedZero.key, cachedZero)
      cachedZero = null
    }
    let activeZero: CachedZeroEntry | null = null
    if (!disable) {
      if (cachedZero?.key !== instanceKey) {
        if (cachedZero) retiredZero.set(cachedZero.key, cachedZero)
        const revived = retiredZero.get(instanceKey)
        retiredZero.delete(instanceKey)
        const nextCachedZero = revived ?? {
          key: instanceKey,
          auth,
          instance: constructZeroInstance({
            options: scopedProps as Omit<
              ZeroOptions<Schema, ZeroMutators>,
              'schema' | 'mutators'
            >,
            transport,
            beforeReload,
            scheduleReload,
            guardStorage,
            benignLogPatterns,
          }),
        }
        cachedZero = nextCachedZero
      }
      activeZero = cachedZero
      liveInstance = activeZero.instance
    }

    // a disabled provider stops being ready before descendant passive effects
    // run. the instance stays cached — disable is a gate, not a teardown — so
    // flipping back on reuses it. a rotation needs nothing here: the render
    // above already published the replacement.
    useLayoutEffect(() => {
      if (disable && zeroRuntime.zero) {
        mutationLifecycle.fence()
        unpublishZeroInstance(zeroRuntime.zero)
      }
    }, [disable])

    // disposal is commit-only, so construction and teardown are deliberately
    // asymmetric. render can be thrown away and a provider can render without
    // ever mounting; closing an instance the committed tree still holds would
    // blank the app, so the render phase only retires and this closes. runs
    // after every commit because a discarded rotation can retire an instance
    // without instanceKey ever changing between two committed renders.
    useEffect(() => {
      if (retiredZero.size === 0) return
      const outgoing = [...retiredZero.values()]
      retiredZero.clear()
      // close only. the replacement is already published and already fenced
      // the outgoing client's writes, and SetZeroInstance's effect emits the
      // one version change a rotation is allowed to produce.
      for (const { instance: zeroInstance } of outgoing) zeroInstance.close()
    })

    // a changed token on the same identity refreshes auth in place — zero
    // sends an auth update over the live connection instead of reconnecting
    // (upstream ZeroProvider does exactly this). string <-> undefined flips
    // rotate the instance via hasAuth in the identity key instead. the compare
    // reads the CACHE rather than a ref, so a provider that unmounted during
    // sign-out and remounted after sign-in still sees the new token; and it
    // happens HERE rather than during render, because render runs twice under
    // strictmode and a render-phase write makes the second pass read its own
    // update and skip the reconnect entirely.
    useEffect(() => {
      if (!activeZero || typeof auth !== 'string' || activeZero.auth === auth) return
      activeZero.auth = auth
      activeZero.instance.connection.connect({ auth })
    }, [activeZero, auth])

    // Always render the same shell shape, with or without an instance, and
    // whether disable is true or false. While disable=true we hand descendants
    // the stub Zero plus DisabledContext='empty' so useZero/useQuery
    // short-circuit instead of throwing. SetZeroInstance + ConnectionMonitor
    // only mount once an active instance exists, as siblings of children —
    // they NEVER wrap children, so toggling them never re-parents.
    return (
      <AuthDataContext.Provider value={authData}>
        <DisabledContext.Provider value={liveInstance ? false : 'empty'}>
          <ZeroContext.Provider value={liveInstance ?? (DISABLED_ZERO_STUB as any)}>
            <ZeroProviderGenerationContext.Provider
              value={liveInstance ? getZeroProviderGeneration(liveInstance) : null}
            >
              {liveInstance ? <SetZeroInstance /> : null}
              {liveInstance ? (
                <ConnectionMonitor
                  zeroEvents={zeroEvents}
                  auth={auth}
                  refreshAuth={refreshAuth}
                  exposeDataset={connectionDataset}
                  datasetCacheUrl={
                    connectionDataset
                      ? ((props as { cacheURL?: string; server?: string }).cacheURL ??
                        (props as { cacheURL?: string; server?: string }).server)
                      : undefined
                  }
                />
              ) : null}
              {children}
            </ZeroProviderGenerationContext.Provider>
          </ZeroContext.Provider>
        </DisabledContext.Provider>
      </AuthDataContext.Provider>
    )
  }

  const ProvideZero = IS_SERVER_RUNTIME ? ProvideZeroServer : ProvideZeroClient

  const SetZeroInstance = () => {
    const zeroInstance = useZero<Schema, ZeroMutators>()

    // publish before descendant effects perform imperative work.
    publishZeroInstance(zeroInstance)

    useInsertionEffect(() => {
      // suspense disconnects layout and passive effects while hiding an
      // already-mounted tree. insertion effects stay attached until the
      // provider actually unmounts, which is the lifecycle boundary that
      // invalidates this generation.
      publishZeroInstance(zeroInstance)
      return () => {
        if (!unpublishZeroInstance(zeroInstance)) return
        mutationLifecycle.fence()
      }
    }, [zeroInstance])

    useEffect(() => {
      zeroInstanceVersion?.emit(zeroInstanceVersion.value + 1)
    }, [zeroInstance])

    return null
  }

  // connect WITHOUT react. same construction and same publication the provider
  // uses, so `zero`, run(), getQuery(), and waitForZero() all resolve exactly as
  // they do in a mounted app — which is what lets code written against the
  // module-global facade run unchanged in a worker, a durable object, or a
  // script. the caller owns the lifetime and must close().
  //
  // there is no react tree here, so nothing owns rotation: an identity change
  // means close this instance and connect a new one. the host is expected to be
  // scoped to a single identity for its lifetime (one project, one user).
  function connectHeadless(
    props: Omit<ZeroOptions<Schema, ZeroMutators>, 'schema' | 'mutators'> & {
      authData?: AuthData | null
      transport?: ZeroProviderTransport
      beforeReload?: () => Promise<void>
      scheduleReload?: (ctx: ScheduleReloadContext) => void
      guardStorage?: RecoveryGuardStorage
      benignLogPatterns?: readonly ZeroLogPattern[]
      // mint a fresh token. REQUIRED for any host that outlives its token: zero
      // parks in needs-auth and will not resume until the auth string changes,
      // so without this a long-lived headless host answers 401 forever.
      refreshAuth?: () => Promise<string | undefined>
    }
  ): { zero: ZeroInstance; close: () => Promise<void> } {
    const {
      authData,
      transport,
      beforeReload,
      scheduleReload,
      guardStorage,
      benignLogPatterns,
      refreshAuth,
      ...options
    } = props
    // mutations read auth dynamically through getAuthData(), so this has to be
    // set before the first mutation exactly as the provider sets it in render.
    setAuthData((authData ?? null) as AuthData)
    let closed = false
    let activeInstance: ZeroInstance
    let unwatch = () => {}
    const construct = () =>
      constructZeroInstance({
        options,
        transport,
        beforeReload,
        scheduleReload,
        guardStorage,
        benignLogPatterns,
        // a headless host has no provider state to bump and no page it owns to
        // reload. replace the disabled instance directly, then republish the
        // stable module facade and move the connection watcher with it.
        recoverInPlace: async () => {
          // an update can arrive after the host intentionally closed this
          // connection during preview turnover. there is no live client left
          // to reconstruct, so recovery is already satisfied.
          if (closed) return true
          const outgoing = activeInstance
          unwatch()
          clearZeroInstanceReferences(outgoing)
          mutationLifecycle.fence()
          try {
            outgoing.close()
          } catch {}
          activeInstance = construct()
          publishZeroInstance(activeInstance)
          zeroInstanceVersion?.emit(zeroInstanceVersion.value + 1)
          unwatch = watchZeroConnection({
            zeroInstance: activeInstance,
            auth: options.auth,
            refreshAuth,
          })
          return true
        },
        reload: () => false,
      })
    activeInstance = construct()
    publishZeroInstance(activeInstance)
    zeroInstanceVersion?.emit(zeroInstanceVersion.value + 1)
    unwatch = watchZeroConnection({
      zeroInstance: activeInstance,
      auth: options.auth,
      refreshAuth,
    })
    return {
      get zero() {
        return activeInstance
      },
      close: async () => {
        closed = true
        unwatch()
        clearZeroInstanceReferences(activeInstance)
        mutationLifecycle.fence()
        await activeInstance.close()
      },
    }
  }

  // watch a zero instance's connection and own the generic recovery: stale-poke
  // and transport reconnects, needs-auth token refresh, reconnect-status events,
  // and optional dataset mirroring. plain subscription, no react — a headless
  // host needs exactly this and previously got none of it, so its token expiry
  // ended as a permanent 401 with no way back.
  function watchZeroConnection(args: {
    zeroInstance: ZeroInstance
    auth?: string | null
    refreshAuth?: () => Promise<string | undefined>
    exposeDataset?: boolean
    datasetCacheUrl?: string
  }): () => void {
    const { zeroInstance, auth, refreshAuth, exposeDataset, datasetCacheUrl } = args
    let prevState = zeroInstance.connection.state.current.name
    let hasConnected = false
    let active = true
    let activeAuth = auth
    const currentReconnect =
      zeroEvents.value?.type === 'reconnect' && zeroEvents.value.status !== 'connected'
        ? zeroEvents.value
        : null
    let reconnect: { reasonKey: ZeroReconnectReasonKey; reason: string } | null =
      currentReconnect
        ? { reasonKey: currentReconnect.reasonKey, reason: currentReconnect.reason }
        : null
    // one reconnect per distinct recoverable error / one refresh per needs-auth
    // transition, so a stuck state doesn't retry-storm.
    let recoverableError: string | null = null
    let needsAuth = false
    let authRefreshAttempts = 0
    const installedTransport = installedTransports.get(zeroInstance)
    const unsubscribeTransport = installedTransport
      ? installedTransport.subscribeLifecycle((event: HttpPullLifecycleEvent) => {
          if (event.type === 'pull' && event.clientID === zeroInstance.clientID) {
            authRefreshAttempts = 0
          }
        })
      : undefined

    const handle = () => {
      const state = zeroInstance.connection.state.current
      const name = state.name
      const reason =
        'reason' in state && typeof state.reason === 'string' ? state.reason : ''

      // mirror connection state onto the body dataset for e2e/diagnostics
      // (enabled on one instance so instances don't clobber each other).
      if (exposeDataset && typeof document !== 'undefined' && document.body) {
        document.body.dataset.zeroState = name
        if (datasetCacheUrl) document.body.dataset.zeroCacheUrl = datasetCacheUrl
        if (reason) document.body.dataset.zeroReason = reason.slice(0, 200)
        else delete document.body.dataset.zeroReason
        if (name === 'connected') document.body.dataset.zeroConnected = 'true'
        else delete document.body.dataset.zeroConnected
      }

      if (name === 'connected') {
        hasConnected = true
        recoverableError = null
        if (!installedTransport) authRefreshAttempts = 0
        if (reconnect) {
          reconnect = null
          emitReconnectStatus({ type: 'reconnect', status: 'connected' })
        }
      }

      const reconnectReasonKey: ZeroReconnectReasonKey | undefined = reason.includes(
        'ServerOverloaded'
      )
        ? 'server-overloaded'
        : reason.includes('Failed to fetch') ||
            reason.includes('fetch failed') ||
            reason.includes('NetworkError when attempting to fetch resource') ||
            reason.includes('Network request failed') ||
            reason.includes('Load failed')
          ? 'transport'
          : undefined

      // stale-poke and paused transport errors both resume the existing client.
      // ServerOverloaded remains in Zero's own retry/backoff loop.
      if (
        name === 'error' &&
        (isRecoverableZeroStalePokeMessage(reason) || reconnectReasonKey)
      ) {
        if (recoverableError !== reason) {
          recoverableError = reason
          reconnect = { reasonKey: reconnectReasonKey ?? 'transport', reason }
          emitReconnectStatus({ type: 'reconnect', status: 'trying', ...reconnect })
          void Promise.resolve(zeroInstance.connection?.connect?.()).catch(() => {})
        }
        return
      }
      if (name !== 'error') recoverableError = null

      if (name === 'connecting' && (reconnect || hasConnected || Boolean(reason))) {
        reconnect = {
          reasonKey: reconnectReasonKey ?? reconnect?.reasonKey ?? 'transport',
          reason: reason || reconnect?.reason || 'connection interrupted',
        }
        emitReconnectStatus({
          type: 'reconnect',
          status: reason ? 'waiting' : 'trying',
          ...reconnect,
        })
      } else if (name === 'disconnected' && (reconnect || hasConnected)) {
        reconnect = {
          reasonKey: reconnect?.reasonKey ?? 'transport',
          reason: reason || reconnect?.reason || 'connection interrupted',
        }
        emitReconnectStatus({ type: 'reconnect', status: 'waiting', ...reconnect })
      }

      // needs-auth: zero is already parked, so resume only for a changed token.
      // repeated distinct rejected tokens get a finite budget; a successful
      // pull proves the replacement and resets it.
      if (name === 'needs-auth') {
        if (
          refreshAuth &&
          !needsAuth &&
          authRefreshAttempts < MAX_AUTH_REFRESH_ATTEMPTS
        ) {
          needsAuth = true
          void refreshAuth()
            .then((token) => {
              if (!active || !token || token === activeAuth) return
              activeAuth = token
              authRefreshAttempts++
              return zeroInstance.connection?.connect?.({ auth: token })
            })
            .catch(() => {})
        }
      } else {
        needsAuth = false
      }

      if (name !== prevState) {
        prevState = name
        if (name === 'error' || name === 'needs-auth') {
          zeroEvents.emit({
            type: 'error',
            reasonKey:
              name === 'needs-auth' ? 'connection-needs-auth' : 'connection-error',
            message: reason || name,
          })
        }
      }
    }

    const unsubscribe = zeroInstance.connection.state.subscribe(handle)
    handle()
    return () => {
      active = false
      unsubscribeTransport?.()
      unsubscribe()
    }
  }

  // monitors connection state and emits events (replaces onError callback removed
  // in 0.25). also owns the generic-Zero connection recovery that used to live in
  // each consumer: stale-poke reconnect, needs-auth token refresh, and optional
  // e2e dataset bookkeeping.
  const ConnectionMonitor = memo(
    ({
      zeroEvents: _zeroEvents,
      auth,
      refreshAuth,
      exposeDataset,
      datasetCacheUrl,
    }: {
      zeroEvents: ZeroEventsEmitter
      auth?: string | null
      refreshAuth?: () => Promise<string | undefined>
      exposeDataset?: boolean
      datasetCacheUrl?: string
    }) => {
      const zeroInstance = useZero<Schema, ZeroMutators>()
      // the recovery logic itself is not react's — it is a subscription to the
      // instance's connection state, shared with connectHeadless so a non-react
      // host recovers identically instead of going dark on token expiry.
      useEffect(
        () =>
          watchZeroConnection({
            zeroInstance,
            auth,
            refreshAuth,
            exposeDataset,
            datasetCacheUrl,
          }),
        [zeroInstance, auth, refreshAuth, exposeDataset, datasetCacheUrl]
      )
      return null
    }
  )

  // preload data for a query into cache without materializing
  // uses same function signature as useQuery
  function preload<TArg, TTable extends keyof Schema['tables'] & string, TReturn>(
    fn: PlainQueryFn<TArg, Query<TTable, Schema, TReturn>>,
    params: TArg,
    options?: PreloadOptions
  ): { cleanup: () => void; complete: Promise<void> }
  function preload<TTable extends keyof Schema['tables'] & string, TReturn>(
    fn: PlainQueryFn<void, Query<TTable, Schema, TReturn>>,
    options?: PreloadOptions
  ): { cleanup: () => void; complete: Promise<void> }
  function preload(
    fnArg: any,
    paramsOrOptions?: any,
    optionsArg?: PreloadOptions
  ): { cleanup: () => void; complete: Promise<void> } {
    const hasParams =
      optionsArg !== undefined || (paramsOrOptions && !('ttl' in paramsOrOptions))
    const params = hasParams ? paramsOrOptions : undefined
    const options = hasParams ? optionsArg : paramsOrOptions

    const queryRequest = resolveQuery({ customQueries, fn: fnArg, params })
    return zero.preload(queryRequest as any, options)
  }

  function getQuery<TArg, TTable extends keyof Schema['tables'] & string, TReturn>(
    fn: PlainQueryFn<TArg, Query<TTable, Schema, TReturn>>,
    params: TArg
  ): ReturnType<typeof resolveQuery<Schema>>
  function getQuery<TTable extends keyof Schema['tables'] & string, TReturn>(
    fn: PlainQueryFn<void, Query<TTable, Schema, TReturn>>
  ): ReturnType<typeof resolveQuery<Schema>>
  function getQuery(fn: any, params?: any) {
    return resolveQuery({ customQueries, fn, params })
  }

  function ControlQueries({
    children,
    action = 'disable',
    whenDisabled = 'empty',
  }: {
    children: ReactNode
    action?: 'enable' | 'disable'
    whenDisabled?: 'empty' | 'last-value'
  }) {
    const mode: QueryControlMode = action === 'enable' ? false : whenDisabled
    return <DisabledContext.Provider value={mode}>{children}</DisabledContext.Provider>
  }

  return {
    instanceName,
    zeroEvents,
    reloadPage,
    ProvideZero,
    connectHeadless,
    ControlQueries,
    useQuery,
    useQueryDirect,
    usePermission,
    usePermissionDirect,
    useZeroProviderGeneration,
    zero,
    preload,
    getQuery,
    waitForZero,
    remint,
    retire,
    // combineZeroClients dispatches acknowledgement through this
    mutationLifecycle,
    drainBackgroundMutations: mutationLifecycle.drainBackgroundMutations,
    enqueueBackgroundMutation: mutationLifecycle.enqueueBackgroundMutation,
    awaitMutationClient: mutationLifecycle.awaitMutationClient,
    awaitMutationServer: mutationLifecycle.awaitMutationServer,
  }
}