leptos_fetch/

query_client.rs

#![allow(ungated_async_fn_track_caller)] // Want it to auto-turn on when stable

use std::{
    borrow::Borrow,
    collections::HashMap,
    fmt::Debug,
    hash::Hash,
    sync::{Arc, LazyLock, atomic::AtomicBool},
};

use chrono::{DateTime, Utc};
use codee::{Decoder, Encoder};
use futures::{FutureExt, pin_mut};
use leptos::{
    prelude::{
        ArcRwSignal, ArcSignal, Effect, Get, GetUntracked, LocalStorage, Owner, Read, ReadUntracked, Set, Signal,
        Track, provide_context,
    },
    server::{ArcLocalResource, ArcResource, FromEncodedStr, IntoEncodedString, LocalResource, Resource},
};
use send_wrapper::SendWrapper;

use crate::{
    ArcLocalSignal, QueryOptions,
    cache::{CachedOrFetchCbInput, CachedOrFetchCbInputVariant, CachedOrFetchCbOutput, OnScopeMissing},
    cache_scope::{QueryAbortReason, QueryOrPending},
    debug_if_devtools_enabled::DebugIfDevtoolsEnabled,
    maybe_local::MaybeLocal,
    query::Query,
    query_maybe_key::QueryMaybeKey,
    query_scope::{QueryScopeInfo, QueryScopeLocalTrait, QueryScopeQueryInfo, QueryScopeTrait, ScopeCacheKey},
    resource_drop_guard::ResourceDropGuard,
    utils::{
        KeyHash, OwnerChain, ResetInvalidated, client_only_yield_now, new_buster_id, new_resource_id,
        run_external_callbacks,
    },
};

use super::cache::ScopeLookup;

#[cfg(not(feature = "rkyv"))]
pub(crate) type DefaultCodec = codee::string::JsonSerdeCodec;

#[cfg(feature = "rkyv")]
pub(crate) type DefaultCodec = codee::binary::RkyvCodec;

task_local::task_local! {
    pub(crate) static ASYNC_TRACK_UPDATE_MARKER: Arc<AtomicBool>;
}

std::thread_local! {
    pub(crate) static SYNC_TRACK_UPDATE_MARKER: AtomicBool = const { AtomicBool::new(true) };
}

/// The [`QueryClient`] stores all query data, and is used to manage queries.
///
/// Should be provided via leptos context at the top of the app.
///
/// # Example
///
/// ```
/// use leptos::prelude::*;
/// use leptos_fetch::QueryClient;
///
/// #[component]
/// pub fn App() -> impl IntoView {
///    QueryClient::new().provide();
///     // ...
/// }
///
/// #[component]
/// pub fn MyComponent() -> impl IntoView {
///     let client: QueryClient = expect_context();
///      // ...
/// }
/// ```
pub struct QueryClient<Codec = DefaultCodec> {
    pub(crate) untyped_client: UntypedQueryClient,
    _ser: std::marker::PhantomData<SendWrapper<Codec>>,
}

/// The internal untyped part of the [`QueryClient`],
/// containing the methods that don't rely on the Codec generic.
#[derive(Clone, Copy)]
pub(crate) struct UntypedQueryClient {
    pub(crate) scope_lookup: ScopeLookup,
    options: QueryOptions,
    created_at: DateTime<Utc>,
}

impl<Codec> Debug for QueryClient<Codec> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("QueryClient")
            .field("scope_lookup", &self.untyped_client.scope_lookup)
            .field("options", &self.untyped_client.options)
            .field("codec", &std::any::type_name::<Codec>())
            .finish()
    }
}

impl<Codec: 'static> Clone for QueryClient<Codec> {
    fn clone(&self) -> Self {
        *self
    }
}

impl<Codec: 'static> Copy for QueryClient<Codec> {}

impl Default for QueryClient<DefaultCodec> {
    fn default() -> Self {
        Self::new()
    }
}

impl QueryClient<DefaultCodec> {
    /// Creates a new [`QueryClient`] with the default codec: [`codee::string::JsonSerdeCodec`].
    ///
    /// Call [`QueryClient::set_codec()`] to set a different codec.
    ///
    /// Call [`QueryClient::with_options()`] to set non-default options.
    #[track_caller]
    pub fn new() -> Self {
        Self {
            untyped_client: UntypedQueryClient {
                scope_lookup: ScopeLookup::new(),
                options: QueryOptions::default(),
                created_at: Utc::now(),
            },
            _ser: std::marker::PhantomData,
        }
    }
}

impl<Codec: 'static> QueryClient<Codec> {
    /// Provide the client to leptos context.
    /// ```no_run
    /// use leptos_fetch::QueryClient;
    /// use leptos::prelude::*;
    ///
    /// QueryClient::new().provide();
    ///
    /// let client: QueryClient = expect_context();
    /// ```
    #[track_caller]
    pub fn provide(self) -> Self {
        provide_context(self);
        self
    }

    /// **Applies to `ssr` only**
    ///
    /// It's possible to use non-json codecs for streaming leptos resources from the backend.
    /// The default is [`codee::string::JsonSerdeCodec`].
    ///
    /// The current `codee` major version is `0.3` and will need to be
    /// imported in your project to customize the codec.
    ///
    /// E.g. to use [`codee::binary::MsgpackSerdeCodec`]:
    /// ```toml
    /// codee = { version = "0.3", features = ["msgpack_serde"] }
    /// ```
    ///
    /// [`codee::binary::MsgpackSerdeCodec`]: https://docs.rs/codee/latest/codee/binary/struct.MsgpackSerdeCodec.html
    ///
    /// This is a generic type on the [`QueryClient`], so when calling
    /// [`leptos::prelude::expect_context`],
    /// this type must be specified when not using the default.
    ///
    /// A useful pattern is to type alias the client with the custom codec for your whole app:
    ///
    /// ```no_run
    /// use codee::binary::MsgpackSerdeCodec;
    /// use leptos::prelude::*;
    /// use leptos_fetch::QueryClient;
    ///
    /// type MyQueryClient = QueryClient<MsgpackSerdeCodec>;
    ///
    /// // Create and provide to context to make accessible everywhere:
    /// QueryClient::new().set_codec::<MsgpackSerdeCodec>().provide();
    ///
    /// let client: MyQueryClient = expect_context();
    /// ```
    #[track_caller]
    pub fn set_codec<NewCodec>(self) -> QueryClient<NewCodec> {
        QueryClient {
            untyped_client: self.untyped_client,
            _ser: std::marker::PhantomData,
        }
    }

    /// Set non-default options to apply to all queries.
    ///
    /// These options will be combined with any options for a specific query scope.
    #[track_caller]
    pub fn with_options(mut self, options: QueryOptions) -> Self {
        self.untyped_client.options = options;
        self
    }

    /// Read the base [`QueryOptions`] for this [`QueryClient`].
    ///
    /// These will be combined with any options for a specific query scope.
    #[track_caller]
    pub fn options(&self) -> QueryOptions {
        self.untyped_client.options
    }

    /// Provide a signal to globally enable/disable auto refetching of queries
    /// that have active resources and have set [`QueryOptions::with_refetch_interval`].
    ///
    /// Whilst this signal returns `false`, refetches will be skipped.
    #[track_caller]
    pub fn with_refetch_enabled_toggle(self, refetch_enabled: impl Into<ArcSignal<bool>>) -> Self {
        self.untyped_client.scope_lookup.scopes_mut().refetch_enabled = Some(refetch_enabled.into());
        self
    }

    /// If set, access the signal that globally enables/disables auto refetching of queries.
    #[track_caller]
    pub fn refetch_enabled(&self) -> Option<ArcSignal<bool>> {
        self.untyped_client.scope_lookup.scopes().refetch_enabled.clone()
    }

    /// Query with [`LocalResource`]. Local resouces only load data on the client,
    /// so can be used with non-threadsafe/serializable data.
    ///
    /// If a cached value exists but is stale, the cached value will be initially used,
    /// then refreshed in the background, updating once the new value is ready.
    #[track_caller]
    pub fn local_resource<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M> + 'static,
        keyer: impl Fn() -> MaybeKey + 'static,
    ) -> LocalResource<MaybeKey::MappedValue>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: DebugIfDevtoolsEnabled + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        self.arc_local_resource(query_scope, keyer).into()
    }

    /// Query with [`ArcLocalResource`]. Local resouces only load data on the client,
    /// so can be used with non-threadsafe/serializable data.
    ///
    /// If a cached value exists but is stale, the cached value will be initially used,
    /// then refreshed in the background, updating once the new value is ready.
    #[track_caller]
    pub fn arc_local_resource<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M> + 'static,
        keyer: impl Fn() -> MaybeKey + 'static,
    ) -> ArcLocalResource<MaybeKey::MappedValue>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: DebugIfDevtoolsEnabled + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let client = *self;
        let client_options = self.options();
        let cache_key = query_scope.cache_key();
        let query_scope_info = QueryScopeInfo::new_local(&query_scope);
        let query_scope = Arc::new(query_scope);
        let query_options = query_scope.options();
        let resource_id = new_resource_id();

        // To call .mark_resource_dropped() when the resource is dropped:
        let drop_guard = ResourceDropGuard::<K, V>::new(self.untyped_client.scope_lookup, resource_id, cache_key);

        ArcLocalResource::new({
            move || {
                let query_scope = query_scope.clone();
                let query_scope_info = query_scope_info.clone();
                let maybe_key = keyer().into_maybe_key();
                let drop_guard = drop_guard.clone();
                if let Some(key) = maybe_key.as_ref() {
                    drop_guard.set_active_key(KeyHash::new(key));
                }
                // Note: cannot hoist outside of resource,
                // it prevents the resource itself dropping when
                // the owner is held by the resource itself, here
                // each instance only lasts as long as the query.
                let owner_chain = OwnerChain::new(client.untyped_client, query_scope_info.cache_key, Owner::current());
                async move {
                    if let Some(key) = maybe_key {
                        let query_scope_query_info = || QueryScopeQueryInfo::new_local(&query_scope, &key);

                        // Regression test for https://github.com/zakstucke/leptos-fetch/issues/64
                        //
                        // When data is prefetched, resources should still go through an async suspend-resolve
                        // cycle (yield) rather than resolving synchronously. This is critical for `Transition` to
                        // work correctly: without the yield, the Transition's internal `nth_run` counter doesn't
                        // advance past the threshold, causing the fallback to show on the first key change.
                        let was_cached = AtomicBool::new(false);

                        let value = client
                            .untyped_client
                            .cached_or_fetch(
                                client_options,
                                query_options,
                                None,
                                &query_scope_info,
                                query_scope_query_info,
                                &key,
                                {
                                    let query_scope = query_scope.clone();
                                    move |key| {
                                        let query_scope = query_scope.clone();

                                        async move { MaybeLocal::new_local(query_scope.query(key).await) }
                                    }
                                },
                                |info| {
                                    info.cached.buster.track();
                                    info.cached.mark_resource_active(resource_id);
                                    match info.variant {
                                        CachedOrFetchCbInputVariant::CachedUntouched => {
                                            was_cached.store(true, std::sync::atomic::Ordering::Relaxed);
                                            // If stale refetch in the background with the
                                            // prefetch() function, which'll recognise it's
                                            // stale, refetch it and invalidate busters:
                                            if cfg!(any(test, not(feature = "ssr")))
                                                && info.cached.stale_or_invalidated()
                                            {
                                                let key = key.clone();
                                                let query_scope = query_scope.clone();
                                                let owner_chain = owner_chain.clone();
                                                // Just adding the SendWrapper and using
                                                // spawn() rather than spawn_local()
                                                // to fix tests:
                                                leptos::task::spawn(SendWrapper::new(async move {
                                                    client
                                                        .prefetch_inner(
                                                            QueryScopeInfo::new_local(&query_scope),
                                                            || QueryScopeQueryInfo::new_local(&query_scope, &key),
                                                            async |key| {
                                                                MaybeLocal::new_local(query_scope.query(key).await)
                                                            },
                                                            key.borrow(),
                                                            || MaybeLocal::new_local(key.borrow().clone()),
                                                            &owner_chain,
                                                        )
                                                        .await;
                                                }));
                                            }
                                        }
                                        CachedOrFetchCbInputVariant::Fresh => {}
                                        CachedOrFetchCbInputVariant::CachedUpdated => {
                                            panic!("Didn't direct inner to refetch here. (bug)")
                                        }
                                    }
                                    CachedOrFetchCbOutput::Return(
                                        // WONTPANIC: cached_or_fetch will
                                        // only output values that are safe
                                        // on this thread:
                                        info.cached.value_maybe_stale().value_may_panic().clone(),
                                    )
                                },
                                None,
                                || MaybeLocal::new_local(key.clone()),
                                &owner_chain,
                            )
                            .await;

                        // Used to add a yield sleep when cached,
                        // to prevent Transitions registering the first load, leading to fallback being shown on second load.
                        // https://github.com/zakstucke/leptos-fetch/issues/64
                        if was_cached.load(std::sync::atomic::Ordering::Relaxed) {
                            client_only_yield_now().await;
                        }

                        MaybeKey::prepare_mapped_value(Some(value))
                    } else {
                        MaybeKey::prepare_mapped_value(None)
                    }
                }
            }
        })
    }

    /// Query with [`Resource`].
    ///
    /// Resources must be serializable to potentially load in `ssr` and stream to the client.
    ///
    /// Resources must be `Send` and `Sync` to be multithreaded in ssr.
    ///
    /// If a cached value exists but is stale, the cached value will be initially used,
    /// then refreshed in the background, updating once the new value is ready.
    #[track_caller]
    pub fn resource<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeTrait<K, V, M> + Send + Sync + 'static,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
    ) -> Resource<MaybeKey::MappedValue, Codec>
    where
        K: DebugIfDevtoolsEnabled + PartialEq + Hash + Clone + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: Clone + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
        Codec: Encoder<MaybeKey::MappedValue> + Decoder<MaybeKey::MappedValue>,
        <Codec as Encoder<MaybeKey::MappedValue>>::Error: Debug,
        <Codec as Decoder<MaybeKey::MappedValue>>::Error: Debug,
        <<Codec as Decoder<MaybeKey::MappedValue>>::Encoded as FromEncodedStr>::DecodingError: Debug,
        <Codec as Encoder<MaybeKey::MappedValue>>::Encoded: IntoEncodedString,
        <Codec as Decoder<MaybeKey::MappedValue>>::Encoded: FromEncodedStr,
    {
        self.arc_resource_with_options(query_scope, keyer, false).into()
    }

    /// Query with a blocking [`Resource`].
    ///
    /// Resources must be serializable to potentially load in `ssr` and stream to the client.
    ///
    /// Resources must be `Send` and `Sync` to be multithreaded in ssr.
    ///
    /// If a cached value exists but is stale, the cached value will be initially used,
    /// then refreshed in the background, updating once the new value is ready.
    #[track_caller]
    pub fn resource_blocking<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeTrait<K, V, M> + Send + Sync + 'static,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
    ) -> Resource<MaybeKey::MappedValue, Codec>
    where
        K: DebugIfDevtoolsEnabled + PartialEq + Hash + Clone + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: Clone + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
        Codec: Encoder<MaybeKey::MappedValue> + Decoder<MaybeKey::MappedValue>,
        <Codec as Encoder<MaybeKey::MappedValue>>::Error: Debug,
        <Codec as Decoder<MaybeKey::MappedValue>>::Error: Debug,
        <<Codec as Decoder<MaybeKey::MappedValue>>::Encoded as FromEncodedStr>::DecodingError: Debug,
        <Codec as Encoder<MaybeKey::MappedValue>>::Encoded: IntoEncodedString,
        <Codec as Decoder<MaybeKey::MappedValue>>::Encoded: FromEncodedStr,
    {
        self.arc_resource_with_options(query_scope, keyer, true).into()
    }

    /// Query with [`ArcResource`].
    ///
    /// Resources must be serializable to potentially load in `ssr` and stream to the client.
    ///
    /// Resources must be `Send` and `Sync` to be multithreaded in ssr.
    ///
    /// If a cached value exists but is stale, the cached value will be initially used,
    /// then refreshed in the background, updating once the new value is ready.
    #[track_caller]
    pub fn arc_resource<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeTrait<K, V, M> + Send + Sync + 'static,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
    ) -> ArcResource<MaybeKey::MappedValue, Codec>
    where
        K: DebugIfDevtoolsEnabled + PartialEq + Hash + Clone + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: Clone + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
        Codec: Encoder<MaybeKey::MappedValue> + Decoder<MaybeKey::MappedValue>,
        <Codec as Encoder<MaybeKey::MappedValue>>::Error: Debug,
        <Codec as Decoder<MaybeKey::MappedValue>>::Error: Debug,
        <<Codec as Decoder<MaybeKey::MappedValue>>::Encoded as FromEncodedStr>::DecodingError: Debug,
        <Codec as Encoder<MaybeKey::MappedValue>>::Encoded: IntoEncodedString,
        <Codec as Decoder<MaybeKey::MappedValue>>::Encoded: FromEncodedStr,
    {
        self.arc_resource_with_options(query_scope, keyer, false)
    }

    /// Query with a blocking [`ArcResource`].
    ///
    /// Resources must be serializable to potentially load in `ssr` and stream to the client.
    ///
    /// Resources must be `Send` and `Sync` to be multithreaded in ssr.
    ///
    /// If a cached value exists but is stale, the cached value will be initially used,
    /// then refreshed in the background, updating once the new value is ready.
    #[track_caller]
    pub fn arc_resource_blocking<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeTrait<K, V, M> + Send + Sync + 'static,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
    ) -> ArcResource<MaybeKey::MappedValue, Codec>
    where
        K: DebugIfDevtoolsEnabled + PartialEq + Hash + Clone + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: Clone + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
        Codec: Encoder<MaybeKey::MappedValue> + Decoder<MaybeKey::MappedValue>,
        <Codec as Encoder<MaybeKey::MappedValue>>::Error: Debug,
        <Codec as Decoder<MaybeKey::MappedValue>>::Error: Debug,
        <<Codec as Decoder<MaybeKey::MappedValue>>::Encoded as FromEncodedStr>::DecodingError: Debug,
        <Codec as Encoder<MaybeKey::MappedValue>>::Encoded: IntoEncodedString,
        <Codec as Decoder<MaybeKey::MappedValue>>::Encoded: FromEncodedStr,
    {
        self.arc_resource_with_options(query_scope, keyer, true)
    }

    #[track_caller]
    fn arc_resource_with_options<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeTrait<K, V, M> + Send + Sync + 'static,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
        blocking: bool,
    ) -> ArcResource<MaybeKey::MappedValue, Codec>
    where
        K: DebugIfDevtoolsEnabled + PartialEq + Hash + Clone + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: Clone + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
        Codec: Encoder<MaybeKey::MappedValue> + Decoder<MaybeKey::MappedValue>,
        <Codec as Encoder<MaybeKey::MappedValue>>::Error: Debug,
        <Codec as Decoder<MaybeKey::MappedValue>>::Error: Debug,
        <<Codec as Decoder<MaybeKey::MappedValue>>::Encoded as FromEncodedStr>::DecodingError: Debug,
        <Codec as Encoder<MaybeKey::MappedValue>>::Encoded: IntoEncodedString,
        <Codec as Decoder<MaybeKey::MappedValue>>::Encoded: FromEncodedStr,
    {
        let client = *self;
        let client_options = self.options();
        let cache_key = query_scope.cache_key();
        let query_scope_info = QueryScopeInfo::new(&query_scope);
        let query_scope = Arc::new(query_scope);
        let scope_lookup = self.untyped_client.scope_lookup;
        let query_options = query_scope.options();

        let buster_if_uncached = ArcRwSignal::new(new_buster_id());
        let resource_id = new_resource_id();

        // To call .mark_resource_dropped() when the resource is dropped:
        let drop_guard = ResourceDropGuard::<K, V>::new(self.untyped_client.scope_lookup, resource_id, cache_key);

        let keyer = Arc::new(keyer);
        let resource = ArcResource::new_with_options(
            {
                let buster_if_uncached = buster_if_uncached.clone();
                let drop_guard = drop_guard.clone();
                let keyer = keyer.clone();
                move || {
                    if let Some(key) = keyer().into_maybe_key() {
                        let key_hash = KeyHash::new(&key);
                        drop_guard.set_active_key(key_hash);
                        scope_lookup.with_cached_query::<K, V, _>(&key_hash, &cache_key, |maybe_cached| {
                            if let Some(cached) = maybe_cached {
                                // Buster must be returned for it to be tracked.
                                (Some(key.clone()), cached.buster.get())
                            } else {
                                // Buster must be returned for it to be tracked.
                                (Some(key.clone()), buster_if_uncached.get())
                            }
                        })
                    } else {
                        (None, buster_if_uncached.get())
                    }
                }
            },
            {
                let buster_if_uncached = buster_if_uncached.clone();
                let query_scope_info = query_scope_info.clone();
                let query_scope = query_scope.clone();
                move |(maybe_key, last_used_buster)| {
                    let query_scope = query_scope.clone();
                    let query_scope_info = query_scope_info.clone();
                    let buster_if_uncached = buster_if_uncached.clone();
                    // Want the guard around everywhere
                    // until the resource is dropped.
                    let _drop_guard = drop_guard.clone();
                    // Note: cannot hoist outside of resource,
                    // it prevents the resource itself dropping
                    // when the owner is held by the resource
                    // itself, here each instance only lasts
                    // as long as the query.
                    let owner_chain =
                        OwnerChain::new(client.untyped_client, query_scope_info.cache_key, Owner::current());
                    async move {
                        if let Some(key) = maybe_key {
                            let query_scope_query_info = || QueryScopeQueryInfo::new(&query_scope, &key);

                            // Regression test for https://github.com/zakstucke/leptos-fetch/issues/64
                            //
                            // When data is prefetched, resources should still go through an async suspend-resolve
                            // cycle (yield) rather than resolving synchronously. This is critical for `Transition` to
                            // work correctly: without the yield, the Transition's internal `nth_run` counter doesn't
                            // advance past the threshold, causing the fallback to show on the first key change.
                            let was_cached = AtomicBool::new(false);

                            let value = client
                                .untyped_client
                                .cached_or_fetch(
                                    client_options,
                                    query_options,
                                    Some(buster_if_uncached.clone()),
                                    &query_scope_info,
                                    query_scope_query_info,
                                    &key,
                                    {
                                        let query_scope = query_scope.clone();
                                        move |key| {
                                            let query_scope = query_scope.clone();
                                            async move { MaybeLocal::new(query_scope.query(key).await) }
                                        }
                                    },
                                    |info| {
                                        info.cached.mark_resource_active(resource_id);
                                        match info.variant {
                                            CachedOrFetchCbInputVariant::CachedUntouched => {
                                                was_cached.store(true, std::sync::atomic::Ordering::Relaxed);

                                                // If stale, refetch in the background
                                                // with prefetch(), which'll recognise
                                                // it's stale and invalidate busters:
                                                if cfg!(any(test, not(feature = "ssr")))
                                                    && info.cached.stale_or_invalidated()
                                                {
                                                    let key = key.clone();
                                                    let query_scope = query_scope.clone();
                                                    let owner_chain = owner_chain.clone();

                                                    leptos::task::spawn(async move {
                                                        client
                                                            .prefetch_inner(
                                                                QueryScopeInfo::new(&query_scope),
                                                                || QueryScopeQueryInfo::new(&query_scope, &key),
                                                                {
                                                                    let query_scope = query_scope.clone();
                                                                    move |key| {
                                                                        let query_scope = query_scope.clone();

                                                                        async move {
                                                                            MaybeLocal::new(
                                                                                query_scope.query(key).await,
                                                                            )
                                                                        }
                                                                    }
                                                                },
                                                                key.borrow(),
                                                                || MaybeLocal::new(key.borrow().clone()),
                                                                &owner_chain,
                                                            )
                                                            .await;
                                                    });
                                                }

                                                // Handle edge case where the key
                                                // function saw it was uncached, so
                                                // entered here, but when
                                                // cached_or_fetch was acquiring the
                                                // fetch mutex, someone else cached
                                                // it, meaning our key function won't
                                                // be reactive correctly unless we
                                                // invalidate it now:
                                                if last_used_buster != info.cached.buster.get_untracked() {
                                                    buster_if_uncached.set(info.cached.buster.get_untracked());
                                                }
                                            }
                                            CachedOrFetchCbInputVariant::Fresh => {}
                                            CachedOrFetchCbInputVariant::CachedUpdated => {
                                                panic!("Didn't direct inner to refetch here. (bug)")
                                            }
                                        }
                                        CachedOrFetchCbOutput::Return(
                                            // WONTPANIC: cached_or_fetch will
                                            // only output values that are safe
                                            // on this thread:
                                            info.cached.value_maybe_stale().value_may_panic().clone(),
                                        )
                                    },
                                    None,
                                    || MaybeLocal::new(key.clone()),
                                    &owner_chain,
                                )
                                .await;

                            // Used to add a yield sleep when cached,
                            // to prevent Transitions registering the first load, leading to fallback being shown on second load.
                            // https://github.com/zakstucke/leptos-fetch/issues/64
                            if was_cached.load(std::sync::atomic::Ordering::Relaxed) {
                                client_only_yield_now().await;
                            }

                            MaybeKey::prepare_mapped_value(Some(value))
                        } else {
                            MaybeKey::prepare_mapped_value(None)
                        }
                    }
                }
            },
            blocking,
        );

        // On the client, want to repopulate the frontend cache,
        // so should write resources to the cache here if they
        // don't exist.
        // It would be better if in here we could check if the
        // resource was started on the backend/streamed, saves
        // doing most of this if already a frontend resource.
        let effect = {
            let resource = resource.clone();
            let buster_if_uncached = buster_if_uncached.clone();
            let client_created_at = self.untyped_client.created_at;
            // Converting to Arc because the tests like the client
            // get dropped even though this persists:
            move |complete: Option<Option<()>>| {
                if let Some(Some(())) = complete {
                    return Some(());
                }
                if let Some(val) = resource.read().as_ref() {
                    if MaybeKey::mapped_value_is_some(val) {
                        scope_lookup.with_cached_scope_mut::<K, V, _, _>(
                            &mut scope_lookup.scopes_mut(),
                            query_scope_info.cache_key,
                            OnScopeMissing::Create(&query_scope_info),
                            |_| {},
                            |maybe_scope, _| {
                                let scope = maybe_scope.expect("provided a default");
                                if let Some(key) = keyer().into_maybe_key() {
                                    let query_scope_query_info = || QueryScopeQueryInfo::new(&query_scope, &key);
                                    let key_hash = KeyHash::new(&key);

                                    // Had a bug in tests where:
                                    // calling client.resource() multiple
                                    // times, previous impl would only run
                                    // the effect once per
                                    // client.resource(), but would still
                                    // run again on subsequent
                                    // client.resource() with the same key
                                    // value
                                    // - client.resource()
                                    // - invalidate, starts loading in bg
                                    // - client.resource()
                                    // - effect runs again, sees resource
                                    //   has value, goes to set it, fires
                                    //   SsrStreamedValueOverride which
                                    //   cancels the actual new query
                                    //   loading to replace invalidated one
                                    // So have to have a client wide
                                    // protector, to make sure the effect
                                    // really only runs once on the client,
                                    // even if resource() called multiple
                                    // times.
                                    if (cfg!(test) || cfg!(not(feature = "ssr"))) && 
                                        // Preventing memory leak, issue is only for first render:
                                        (Utc::now() - client_created_at).num_seconds() < 10
                                    {
                                        use parking_lot::Mutex;

                                        type AlreadySeenMap = HashMap<(u64, ScopeCacheKey, KeyHash), ()>;
                                        static ALREADY_SEEN: LazyLock<Mutex<AlreadySeenMap>> =
                                            LazyLock::new(|| Mutex::new(HashMap::new()));

                                        let key = (scope_lookup.scope_id, query_scope_info.cache_key, key_hash);
                                        let mut guard = ALREADY_SEEN.lock();
                                        if guard.contains_key(&key) {
                                            return;
                                        }
                                        guard.insert(key, ());
                                    }

                                    let mut was_pending = false;
                                    // Protect against race condition:
                                    // cancel any frontend query that
                                    // started already if there was a
                                    // client side local resource or
                                    // prefetch/fetch_query etc that
                                    // started on initial client-side ticks:
                                    if let Some(QueryOrPending::Pending { query_abort_tx, .. }) =
                                        scope.get_mut_include_pending(&key_hash)
                                        && let Some(tx) = query_abort_tx.take()
                                    {
                                        tx.send(QueryAbortReason::SsrStreamedValueOverride).unwrap();
                                        was_pending = true;
                                    }

                                    if was_pending || !scope.contains_key(&key_hash) {
                                        let query = Query::new(
                                            client_options,
                                            client.untyped_client,
                                            &query_scope_info,
                                            query_scope_query_info(),
                                            key_hash,
                                            MaybeLocal::new(key),
                                            MaybeLocal::new(MaybeKey::mapped_to_maybe_value(val.clone()).expect(
                                                "Just checked \
                                                    MaybeKey::mapped_value_is_some() \
                                                    is true",
                                            )),
                                            buster_if_uncached.clone(),
                                            query_options,
                                            None,
                                            #[cfg(any(
                                                all(debug_assertions, feature = "devtools"),
                                                feature = "devtools-always"
                                            ))]
                                            crate::events::Event::new(crate::events::EventVariant::StreamedFromServer),
                                        );
                                        scope.insert(key_hash, query);
                                    }
                                    scope
                                        .get(&key_hash)
                                        .expect("Just inserted")
                                        .mark_resource_active(resource_id)
                                }
                            },
                        );
                    }
                    Some(())
                } else {
                    None
                }
            }
        };
        // Won't run in tests if not isomorphic, but in prod Effect is wanted to not run on server:
        if cfg!(test) {
            // Wants Send + Sync on the Codec despite using SendWrapper in the PhantomData.
            // Can put a SendWrapper around it as this is test only and they're single threaded:
            let effect = SendWrapper::new(effect);
            #[allow(clippy::redundant_closure)]
            Effect::new_isomorphic(move |v| effect(v));
        } else {
            Effect::new(effect);
        }

        resource
    }

    /// Prevent reactive updates being triggered from the current updater callback fn.
    ///
    /// Call [`QueryClient::untrack_update_query`] in the callback of [`QueryClient::update_query`]
    /// to prevent resources being updated after the callback completes.
    ///
    /// This is really useful when e.g:
    /// - you know nothing changes after reading the existing value in the callback
    /// - you don't want resources/subs to react and rerender, given nothings changed.
    ///
    /// This function is a noop outside the callbacks of:
    /// - [`QueryClient::update_query`]
    /// - [`QueryClient::update_query_async`]
    /// - [`QueryClient::update_query_async_local`]
    pub fn untrack_update_query(&self) {
        // Set markers to false, these will all be reset to true automatically where needed.
        SYNC_TRACK_UPDATE_MARKER.with(|marker| {
            marker.store(false, std::sync::atomic::Ordering::Relaxed);
        });
        // Ignore returned AccessError if didn't exist (not in async ctx):
        let _ = ASYNC_TRACK_UPDATE_MARKER.try_with(|marker| {
            marker.store(false, std::sync::atomic::Ordering::Relaxed);
        });
    }

    /// Prefetch a query and store it in the cache.
    ///
    /// - Entry doesn't exist: fetched and stored in the cache.
    /// - Entry exists but stale: fetched and updated in the cache.
    /// - Entry exists but **not** stale: not refreshed, existing cache item remains.
    ///
    /// If the cached query changes, active resources using the query will be updated.
    #[track_caller]
    pub async fn prefetch_query<K, V, M>(&self, query_scope: impl QueryScopeTrait<K, V, M>, key: impl Borrow<K>)
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
    {
        let query_scope_info = QueryScopeInfo::new(&query_scope);
        let owner_chain = OwnerChain::new(self.untyped_client, query_scope_info.cache_key, Owner::current());
        self.prefetch_inner(
            query_scope_info,
            || QueryScopeQueryInfo::new(&query_scope, key.borrow()),
            async |key| MaybeLocal::new(query_scope.query(key).await),
            key.borrow(),
            || MaybeLocal::new(key.borrow().clone()),
            &owner_chain,
        )
        .await
    }

    /// Prefetch a non-threadsafe query and store it in the cache for this thread only.
    ///
    /// - Entry doesn't exist: fetched and stored in the cache.
    /// - Entry exists but stale: fetched and updated in the cache.
    /// - Entry exists but **not** stale: not refreshed, existing cache item remains.
    ///
    /// If the cached query changes, active resources using the query will be updated.
    #[track_caller]
    pub async fn prefetch_query_local<K, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
    ) where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let query_scope_info = QueryScopeInfo::new_local(&query_scope);
        let owner_chain = OwnerChain::new(self.untyped_client, query_scope_info.cache_key, Owner::current());
        self.prefetch_inner(
            query_scope_info,
            || QueryScopeQueryInfo::new_local(&query_scope, key.borrow()),
            async |key| MaybeLocal::new_local(query_scope.query(key).await),
            key.borrow(),
            || MaybeLocal::new_local(key.borrow().clone()),
            &owner_chain,
        )
        .await
    }

    #[track_caller]
    async fn prefetch_inner<K, V, FetcherFut>(
        &self,
        query_scope_info: QueryScopeInfo,
        query_scope_info_for_new_query: impl Fn() -> QueryScopeQueryInfo<K>,
        fetcher: impl Fn(K) -> FetcherFut,
        key: &K,
        lazy_maybe_local_key: impl FnOnce() -> MaybeLocal<K>,
        owner_chain: &OwnerChain,
    ) where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
        FetcherFut: Future<Output = MaybeLocal<V>>,
    {
        self.untyped_client
            .cached_or_fetch(
                self.options(),
                query_scope_info.options,
                None,
                &query_scope_info,
                query_scope_info_for_new_query,
                key,
                fetcher,
                |info| {
                    match info.variant {
                        CachedOrFetchCbInputVariant::CachedUntouched => {
                            if info.cached.stale_or_invalidated() {
                                return CachedOrFetchCbOutput::Refetch;
                            }
                        }
                        CachedOrFetchCbInputVariant::CachedUpdated => {
                            // Update anything using it:
                            info.cached.buster.set(new_buster_id());
                        }
                        CachedOrFetchCbInputVariant::Fresh => {}
                    }
                    CachedOrFetchCbOutput::Return(())
                },
                None,
                lazy_maybe_local_key,
                owner_chain,
            )
            .await;
    }

    /// Fetch a query, store it in the cache and return it.
    ///
    /// - Entry doesn't exist: fetched and stored in the cache.
    /// - Entry exists but stale: fetched and updated in the cache.
    /// - Entry exists but **not** stale: not refreshed, existing cache item remains.
    ///
    /// If the cached query changes, active resources using the query will be updated.
    ///
    /// Returns the up-to-date cached query.
    #[track_caller]
    pub async fn fetch_query<K, V, M>(&self, query_scope: impl QueryScopeTrait<K, V, M>, key: impl Borrow<K>) -> V
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
    {
        self.untyped_client.fetch_query(query_scope, key).await
    }

    /// Fetch a non-threadsafe query, store it in the cache for this thread only and return it.
    ///
    /// - Entry doesn't exist: fetched and stored in the cache.
    /// - Entry exists but stale: fetched and updated in the cache.
    /// - Entry exists but **not** stale: not refreshed, existing cache item remains.
    ///
    /// If the cached query changes, active resources using the query will be updated.
    ///
    /// Returns the up-to-date cached query.
    #[track_caller]
    pub async fn fetch_query_local<K, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
    ) -> V
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        self.untyped_client.fetch_query_local(query_scope, key).await
    }

    /// Set the value of a query in the cache.
    /// This cached value will be available from all threads and take priority
    /// over any locally cached value for this query.
    ///
    /// Active resources using the query will be updated.
    #[track_caller]
    pub fn set_query<K, V, M>(&self, query_scope: impl QueryScopeTrait<K, V, M>, key: impl Borrow<K>, new_value: V)
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
    {
        self.set_inner(
            QueryScopeInfo::new(&query_scope),
            || QueryScopeQueryInfo::new(&query_scope, key.borrow()),
            key.borrow(),
            MaybeLocal::new(new_value),
            || MaybeLocal::new(key.borrow().clone()),
            true,
            // like update methods, this does not come from the query function itself,
            // so should not reset the invalidated status:
            ResetInvalidated::NoReset,
        )
    }

    /// Set the value of a non-threadsafe query in the cache for this thread only.
    /// This cached value will only be available from this thread, the cache will
    /// return empty, unless a nonlocal value is set, from any other thread.
    ///
    /// Active resources using the query will be updated.
    #[track_caller]
    pub fn set_query_local<K, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
        new_value: V,
    ) where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        self.set_inner::<K, V>(
            QueryScopeInfo::new_local(&query_scope),
            || QueryScopeQueryInfo::new_local(&query_scope, key.borrow()),
            key.borrow(),
            MaybeLocal::new_local(new_value),
            || MaybeLocal::new_local(key.borrow().clone()),
            true,
            // like update methods, this does not come from the query function itself,
            // so should not reset the invalidated status:
            ResetInvalidated::NoReset,
        )
    }

    #[track_caller]
    fn set_inner<K, V>(
        &self,
        query_scope_info: QueryScopeInfo,
        query_scope_info_for_new_query: impl Fn() -> QueryScopeQueryInfo<K>,
        key: &K,
        new_value: MaybeLocal<V>,
        lazy_maybe_local_key: impl FnOnce() -> MaybeLocal<K>,
        track: bool,
        reset_invalidated: ResetInvalidated,
    ) where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let key_hash = KeyHash::new(key.borrow());
        self.untyped_client.scope_lookup.with_cached_scope_mut::<K, V, _, _>(
            &mut self.untyped_client.scope_lookup.scopes_mut(),
            query_scope_info.cache_key,
            OnScopeMissing::Create(&query_scope_info),
            |_| {},
            |maybe_scope, _| {
                let scope = maybe_scope.expect("provided a default");

                // Make sure to look both caches if threadsafe, and prefer threadsafe:
                let maybe_cached = if !new_value.is_local() {
                    if let Some(threadsafe_existing) = scope.get_mut_threadsafe_only(&key_hash) {
                        Some(threadsafe_existing)
                    } else {
                        scope.get_mut_local_only(&key_hash)
                    }
                } else {
                    scope.get_mut_local_only(&key_hash)
                };

                if let Some(cached) = maybe_cached {
                    cached.set_value(
                        new_value,
                        track,
                        #[cfg(any(all(debug_assertions, feature = "devtools"), feature = "devtools-always"))]
                        crate::events::Event::new(crate::events::EventVariant::DeclarativeSet),
                        reset_invalidated,
                    );
                } else {
                    let query = Query::new(
                        self.options(),
                        self.untyped_client,
                        &query_scope_info,
                        query_scope_info_for_new_query(),
                        key_hash,
                        lazy_maybe_local_key(),
                        new_value,
                        ArcRwSignal::new(new_buster_id()),
                        query_scope_info.options,
                        None,
                        #[cfg(any(all(debug_assertions, feature = "devtools"), feature = "devtools-always"))]
                        crate::events::Event::new(crate::events::EventVariant::DeclarativeSet),
                    );
                    scope.insert(key_hash, query);
                }
            },
        );
    }

    /// Synchronously update the value of a query in the cache with a callback.
    ///
    /// Active resources using the query will be updated.
    ///
    /// The callback takes `Option<&mut V>`, will be None if the value is not available in the cache.
    ///
    /// If you want async and/or always get the value, use
    /// [`QueryClient::update_query_async`]/[`QueryClient::update_query_async_local`].
    ///
    /// Returns the output of the callback.
    ///
    /// If you decide you don't want to trigger resources and subscribers, e.g. if you know nothing changed,
    /// call [`QueryClient::untrack_update_query`] in the callback to prevent reactive updates.
    #[track_caller]
    pub fn update_query<K, V, T, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
        modifier: impl FnOnce(Option<&mut V>) -> T,
    ) -> T
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        self.update_query_inner::<K, V, T>(
            &QueryScopeInfo::new_local(&query_scope),
            key.borrow(),
            ResetInvalidated::NoReset,
            modifier,
        )
    }

    #[track_caller]
    fn update_query_inner<K, V, T>(
        &self,
        query_scope_info: &QueryScopeInfo,
        key: &K,
        reset_invalidated: ResetInvalidated,
        modifier: impl FnOnce(Option<&mut V>) -> T,
    ) -> T
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let key_hash = KeyHash::new(key);
        let mut modifier_holder = Some(modifier);

        let maybe_return_value = self.untyped_client.scope_lookup.with_cached_scope_mut::<K, V, _, _>(
            &mut self.untyped_client.scope_lookup.scopes_mut(),
            query_scope_info.cache_key,
            OnScopeMissing::Skip,
            |_| {},
            |maybe_scope, _| {
                if let Some(scope) = maybe_scope
                    && let Some(cached) = scope.get_mut(&key_hash)
                {
                    let modifier = modifier_holder
                        .take()
                        .expect("Should never be used more than once. (bug)");
                    let return_value = cached.update_value(
                        // WONTPANIC: the internals will only supply the value if available from this thread:
                        |value| modifier(Some(value.value_mut_may_panic())),
                        #[cfg(any(all(debug_assertions, feature = "devtools"), feature = "devtools-always"))]
                        crate::events::Event::new(crate::events::EventVariant::DeclarativeUpdate),
                        reset_invalidated,
                    );
                    return Some(return_value);
                }
                None
            },
        );
        if let Some(return_value) = maybe_return_value {
            return_value
        } else {
            // Didn't exist:
            modifier_holder
                .take()
                .expect("Should never be used more than once. (bug)")(None)
        }
    }

    /// Asynchronously map a threadsafe query in the cache from one value to another.
    ///
    /// Unlike [`QueryClient::update_query`], this will fetch the query first, if it doesn't exist.
    ///
    /// Active resources using the query will be updated.
    ///
    /// Returns the output of the callback.
    ///
    /// If you decide you don't want to trigger resources and subscribers, e.g. if you know nothing changed,
    /// call [`QueryClient::untrack_update_query`] in the callback to prevent reactive updates.
    #[track_caller]
    pub async fn update_query_async<'a, K, V, T, M>(
        &'a self,
        query_scope: impl QueryScopeTrait<K, V, M>,
        key: impl Borrow<K>,
        mapper: impl AsyncFnOnce(&mut V) -> T,
    ) -> T
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
    {
        let query_scope_info = QueryScopeInfo::new(&query_scope);
        let owner_chain = OwnerChain::new(self.untyped_client, query_scope_info.cache_key, Owner::current());
        self.update_query_async_inner(
            query_scope_info,
            || QueryScopeQueryInfo::new(&query_scope, key.borrow()),
            async |key| MaybeLocal::new(query_scope.query(key).await),
            key.borrow(),
            mapper,
            MaybeLocal::new,
            || MaybeLocal::new(key.borrow().clone()),
            &owner_chain,
        )
        .await
    }

    /// Asynchronously map a non-threadsafe query in the cache from one value to another.
    ///
    /// Unlike [`QueryClient::update_query`], this will fetch the query first, if it doesn't exist.
    ///
    /// Active resources using the query will be updated.
    ///
    /// Returns the output of the callback.
    ///
    /// If you decide you don't want to trigger resources and subscribers, e.g. if you know nothing changed,
    /// call [`QueryClient::untrack_update_query`] in the callback to prevent reactive updates.
    #[track_caller]
    pub async fn update_query_async_local<'a, K, V, T, M>(
        &'a self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
        mapper: impl AsyncFnOnce(&mut V) -> T,
    ) -> T
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let query_scope_info = QueryScopeInfo::new_local(&query_scope);
        let owner_chain = OwnerChain::new(self.untyped_client, query_scope_info.cache_key, Owner::current());
        self.update_query_async_inner(
            query_scope_info,
            || QueryScopeQueryInfo::new_local(&query_scope, key.borrow()),
            async |key| MaybeLocal::new_local(query_scope.query(key).await),
            key.borrow(),
            mapper,
            MaybeLocal::new_local,
            || MaybeLocal::new_local(key.borrow().clone()),
            &owner_chain,
        )
        .await
    }

    #[track_caller]
    async fn update_query_async_inner<'a, K, V, T, FetcherFut>(
        &'a self,
        query_scope_info: QueryScopeInfo,
        query_scope_info_for_new_query: impl Fn() -> QueryScopeQueryInfo<K>,
        fetcher: impl Fn(K) -> FetcherFut,
        key: &K,
        mapper: impl AsyncFnOnce(&mut V) -> T,
        into_maybe_local: impl FnOnce(V) -> MaybeLocal<V>,
        lazy_maybe_local_key: impl Fn() -> MaybeLocal<K>,
        owner_chain: &OwnerChain,
    ) -> T
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
        FetcherFut: Future<Output = MaybeLocal<V>>,
    {
        let key_hash = KeyHash::new(key.borrow());

        // By holding the fetcher mutex from start to finish, prevent the chance of the value
        // being fetched between the fetch async call and the external user mapper async call
        // and the final set().
        let fetcher_mutex = self
            .untyped_client
            .scope_lookup
            .fetcher_mutex::<K, V>(key_hash, &query_scope_info);
        let fetcher_guard = fetcher_mutex.lock().await;

        let mut new_value = self
            .untyped_client
            .fetch_inner(
                query_scope_info.clone(),
                &query_scope_info_for_new_query,
                fetcher,
                key.borrow(),
                Some(&fetcher_guard),
                &lazy_maybe_local_key,
                owner_chain,
            )
            .await;

        // The fetch will "turn on" is_fetching during it's lifetime, but we also want it during the mapper function:
        self.untyped_client
            .scope_lookup
            .with_notify_fetching(query_scope_info.cache_key, key_hash, false, async {
                let track = Arc::new(AtomicBool::new(true));

                // Will monitor for invalidations during the user async fn, which could take a long time:
                let query_abort_rx = self.untyped_client.scope_lookup.prepare_invalidation_channel::<K, V>(
                    &query_scope_info,
                    key_hash,
                    &lazy_maybe_local_key(),
                );

                let result_fut = ASYNC_TRACK_UPDATE_MARKER.scope(track.clone(), async { mapper(&mut new_value).await });

                // Call the user async function, but also monitor for invalidations:
                // Specifically, we care about if .clear() is called.
                // If so, we shouldn't set the value, as it will have been generated with
                // the cleared cached query value.
                // If invalidated should still set, because the value will be deemed
                // "less stale", but won't de-invalidate it.
                // Don't care about QueryAbortReason::SsrStreamedValueOverride
                // as it shouldn't be possible to get to this point before all ssr streaming is done.
                let mut cleared_during_user_fn = false;
                let result = {
                    pin_mut!(result_fut);
                    pin_mut!(query_abort_rx);
                    match futures::future::select(result_fut, query_abort_rx).await {
                        futures::future::Either::Left((result, _query_abort_rx)) => result,
                        futures::future::Either::Right((query_abort_reason, result_fut)) => {
                            if let Ok(QueryAbortReason::Clear) = query_abort_reason {
                                // If the query was cleared, don't set the new value.
                                cleared_during_user_fn = true;
                            }
                            result_fut.await
                        }
                    }
                };

                // If the query was cleared, don't set the new value.
                if !cleared_during_user_fn {
                    // Cover a small edge case:
                    // - value in threadsafe
                    // - .update_query_async_local called
                    // - replacement value .set() called, stored in local cache because
                    //   no type safety on threadsafe cache
                    // - now have 2 versions in both caches
                    // by trying to update first, means it'll stick with existing cache if it exists,
                    // otherwise doesn't matter and can just be set normally:
                    let mut new_value = Some(new_value);
                    let updated = self.update_query_inner::<K, V, _>(
                        &query_scope_info,
                        key,
                        // fetch_inner would've reset it if needed, then the update itself shouldn't change the value:
                        ResetInvalidated::NoReset,
                        |value| {
                            if let Some(value) = value {
                                // This sync update call itself needs untracking if !track:
                                if !track.load(std::sync::atomic::Ordering::Relaxed) {
                                    self.untrack_update_query();
                                }

                                *value = new_value.take().expect("Should be Some");
                                true
                            } else {
                                false
                            }
                        },
                    );
                    if !updated {
                        self.set_inner::<K, V>(
                            query_scope_info,
                            query_scope_info_for_new_query,
                            key.borrow(),
                            into_maybe_local(
                                new_value
                                    .take()
                                    .expect("Should be Some, should only be here if not updated"),
                            ),
                            lazy_maybe_local_key,
                            track.load(std::sync::atomic::Ordering::Relaxed),
                            // fetch_inner would've reset it if needed, then the update
                            // itself shouldn't change the value:
                            ResetInvalidated::NoReset,
                        );
                    }
                }

                result
            })
            .await
    }

    /// Synchronously get a query from the cache, if it exists.
    #[track_caller]
    pub fn get_cached_query<K, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
    ) -> Option<V>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        self.untyped_client.scope_lookup.with_cached_query::<K, V, _>(
            &KeyHash::new(key.borrow()),
            &query_scope.cache_key(),
            |maybe_cached| {
                // WONTPANIC: with_cached_query will only output values that are safe on this thread:
                maybe_cached.map(|cached| cached.value_maybe_stale().value_may_panic().clone())
            },
        )
    }

    /// Synchronously check if a query exists in the cache.
    ///
    /// Returns `true` if the query exists.
    #[track_caller]
    pub fn query_exists<K, V, M>(&self, query_scope: impl QueryScopeLocalTrait<K, V, M>, key: impl Borrow<K>) -> bool
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + 'static,
    {
        let key_hash = KeyHash::new(key.borrow());
        self.untyped_client.scope_lookup.with_cached_query::<K, V, _>(
            &key_hash,
            &query_scope.cache_key(),
            |maybe_cached| maybe_cached.is_some(),
        )
    }

    /// Subscribe to the `is_loading` status of a query.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This is `true` when the query is in the process of fetching data for the first time.
    /// This is in contrast to `is_fetching`, that is `true` whenever the query
    /// is fetching data, including when it's refetching.
    ///
    /// From a resource perspective:
    /// - `is_loading=true`, the resource will be in a pending state until ready and implies `is_fetching=true`
    /// - `is_fetching=true` + `is_loading=false` means the resource is showing
    ///   previous data, and will update once new data finishes refetching
    /// - `is_fetching=false` means the resource is showing the latest data and implies `is_loading=false`
    #[track_caller]
    pub fn subscribe_is_loading<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
    ) -> Signal<bool>
    where
        K: Hash + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: 'static,
        V: 'static,
    {
        self.subscribe_is_loading_arc(query_scope, keyer).into()
    }

    /// Subscribe to the `is_loading` status of a query with a non-threadsafe key.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This is `true` when the query is in the process of fetching data for the first time.
    /// This is in contrast to `is_fetching`, that is `true` whenever the query
    /// is fetching data, including when it's refetching.
    ///
    /// From a resource perspective:
    /// - `is_loading=true`, the resource will be in a pending state until ready and implies `is_fetching=true`
    /// - `is_fetching=true` + `is_loading=false` means the resource is showing
    ///   previous data, and will update once new data finishes refetching
    /// - `is_fetching=false` means the resource is showing the latest data and implies `is_loading=false`
    #[track_caller]
    pub fn subscribe_is_loading_local<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + 'static,
    ) -> Signal<bool>
    where
        K: Hash + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: 'static,
        V: 'static,
    {
        self.subscribe_is_loading_arc_local(query_scope, keyer).into()
    }

    /// Subscribe to the `is_loading` status of a query with a non-threadsafe key.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This is `true` when the query is in the process of fetching data for the first time.
    /// This is in contrast to `is_fetching`, that is `true` whenever the query
    /// is fetching data, including when it's refetching.
    ///
    /// From a resource perspective:
    /// - `is_loading=true`, the resource will be in a pending state until ready and implies `is_fetching=true`
    /// - `is_fetching=true` + `is_loading=false` means the resource is showing
    ///   previous data, and will update once new data finishes refetching
    /// - `is_fetching=false` means the resource is showing the latest data and implies `is_loading=false`
    #[track_caller]
    pub fn subscribe_is_loading_arc_local<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + 'static,
    ) -> ArcSignal<bool>
    where
        K: Hash + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: 'static,
        V: 'static,
    {
        let keyer = SendWrapper::new(keyer);
        self.untyped_client
            .scope_lookup
            .scope_subscriptions_mut()
            .add_is_loading_subscription(
                query_scope.cache_key(),
                MaybeLocal::new_local(ArcSignal::derive(move || {
                    keyer().into_maybe_key().map(|k| KeyHash::new(&k))
                })),
            )
    }

    /// Subscribe to the `is_loading` status of a query.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This is `true` when the query is in the process of fetching data for the first time.
    /// This is in contrast to `is_fetching`, that is `true` whenever the query
    /// is fetching data, including when it's refetching.
    ///
    /// From a resource perspective:
    /// - `is_loading=true`, the resource will be in a pending state until ready and implies `is_fetching=true`
    /// - `is_fetching=true` + `is_loading=false` means the resource is showing
    ///   previous data, and will update once new data finishes refetching
    /// - `is_fetching=false` means the resource is showing the latest data and implies `is_loading=false`
    #[track_caller]
    pub fn subscribe_is_loading_arc<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
    ) -> ArcSignal<bool>
    where
        K: Hash + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: 'static,
        V: 'static,
    {
        self.untyped_client
            .scope_lookup
            .scope_subscriptions_mut()
            .add_is_loading_subscription(
                query_scope.cache_key(),
                MaybeLocal::new(ArcSignal::derive(move || {
                    keyer().into_maybe_key().map(|k| KeyHash::new(&k))
                })),
            )
    }

    /// Subscribe to the `is_fetching` status of a query.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This is `true` is `true` whenever the query is fetching data, including when it's refetching.
    /// This is in contrast to `is_loading`, that is `true` when the query is in
    /// the process of fetching data for the first time only.
    ///
    /// From a resource perspective:
    /// - `is_loading=true`, the resource will be in a pending state until ready and implies `is_fetching=true`
    /// - `is_fetching=true` + `is_loading=false` means the resource is showing
    ///   previous data, and will update once new data finishes refetching
    /// - `is_fetching=false` means the resource is showing the latest data and implies `is_loading=false`
    #[track_caller]
    pub fn subscribe_is_fetching<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
    ) -> Signal<bool>
    where
        K: Hash + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: 'static,
        V: 'static,
    {
        self.subscribe_is_fetching_arc(query_scope, keyer).into()
    }

    /// Subscribe to the `is_fetching` status of a query with a non-threadsafe key.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This is `true` is `true` whenever the query is fetching data, including when it's refetching.
    /// This is in contrast to `is_loading`, that is `true` when the query is in
    /// the process of fetching data for the first time only.
    ///
    /// From a resource perspective:
    /// - `is_loading=true`, the resource will be in a pending state until ready and implies `is_fetching=true`
    /// - `is_fetching=true` + `is_loading=false` means the resource is showing
    ///   previous data, and will update once new data finishes refetching
    /// - `is_fetching=false` means the resource is showing the latest data and implies `is_loading=false`
    #[track_caller]
    pub fn subscribe_is_fetching_local<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + 'static,
    ) -> Signal<bool>
    where
        K: Hash + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: 'static,
        V: 'static,
    {
        self.subscribe_is_fetching_arc_local(query_scope, keyer).into()
    }

    /// Subscribe to the `is_fetching` status of a query with a non-threadsafe key.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This is `true` is `true` whenever the query is fetching data, including when it's refetching.
    /// This is in contrast to `is_loading`, that is `true` when the query is in
    /// the process of fetching data for the first time only.
    ///
    /// From a resource perspective:
    /// - `is_loading=true`, the resource will be in a pending state until ready and implies `is_fetching=true`
    /// - `is_fetching=true` + `is_loading=false` means the resource is showing
    ///   previous data, and will update once new data finishes refetching
    /// - `is_fetching=false` means the resource is showing the latest data and implies `is_loading=false`
    #[track_caller]
    pub fn subscribe_is_fetching_arc_local<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + 'static,
    ) -> ArcSignal<bool>
    where
        K: Hash + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: 'static,
        V: 'static,
    {
        let keyer = SendWrapper::new(keyer);
        self.untyped_client
            .scope_lookup
            .scope_subscriptions_mut()
            .add_is_fetching_subscription(
                query_scope.cache_key(),
                MaybeLocal::new_local(ArcSignal::derive(move || {
                    keyer().into_maybe_key().map(|k| KeyHash::new(&k))
                })),
            )
    }

    /// Subscribe to the `is_fetching` status of a query.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This is `true` is `true` whenever the query is fetching data, including when it's refetching.
    /// This is in contrast to `is_loading`, that is `true` when the query is in
    /// the process of fetching data for the first time only.
    ///
    /// From a resource perspective:
    /// - `is_loading=true`, the resource will be in a pending state until ready and implies `is_fetching=true`
    /// - `is_fetching=true` + `is_loading=false` means the resource is showing
    ///   previous data, and will update once new data finishes refetching
    /// - `is_fetching=false` means the resource is showing the latest data and implies `is_loading=false`
    #[track_caller]
    pub fn subscribe_is_fetching_arc<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
    ) -> ArcSignal<bool>
    where
        K: Hash + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: 'static,
        V: 'static,
    {
        self.untyped_client
            .scope_lookup
            .scope_subscriptions_mut()
            .add_is_fetching_subscription(
                query_scope.cache_key(),
                MaybeLocal::new(ArcSignal::derive(move || {
                    keyer().into_maybe_key().map(|k| KeyHash::new(&k))
                })),
            )
    }

    /// Subscribe to the value of a query.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This will update whenever the query is created, removed, updated, refetched or set.
    ///
    /// Compared to a resource:
    /// - This will not trigger a fetch of a query, if it's not in the cache, this will be `None`.       
    #[track_caller]
    pub fn subscribe_value<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
    ) -> Signal<Option<V>>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
    {
        self.subscribe_value_arc(query_scope, keyer).into()
    }

    /// Subscribe to the value of a non-threadsafe query.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This will update whenever the query is created, removed, updated, refetched or set.
    ///
    /// Compared to a resource:
    /// - This will not trigger a fetch of a query, if it's not in the cache, this will be `None`.
    #[track_caller]
    pub fn subscribe_value_local<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + 'static,
    ) -> Signal<Option<V>, LocalStorage>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: DebugIfDevtoolsEnabled + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        self.subscribe_value_arc_local(query_scope, keyer).into()
    }

    /// Subscribe to the value of a non-threadsafe query.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This will update whenever the query is created, removed, updated, refetched or set.
    ///
    /// Compared to a resource:
    /// - This will not trigger a fetch of a query, if it's not in the cache, this will be `None`.
    #[track_caller]
    pub fn subscribe_value_arc_local<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + 'static,
    ) -> ArcLocalSignal<Option<V>>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: DebugIfDevtoolsEnabled + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let cache_key = query_scope.cache_key();
        let keyer = ArcLocalSignal::derive_local(move || keyer().into_maybe_key().map(|k| KeyHash::new(&k)));
        let dyn_signal = self
            .untyped_client
            .scope_lookup
            .scope_subscriptions_mut()
            .add_value_set_updated_or_removed_subscription(
                cache_key,
                MaybeLocal::new_local({
                    let keyer = keyer.clone();
                    move || keyer.get()
                }),
            );

        let scope_lookup = self.untyped_client.scope_lookup;
        ArcLocalSignal::derive_local(move || {
            dyn_signal.track();
            if let Some(key) = keyer.read_untracked().as_ref() {
                scope_lookup.with_cached_query::<K, V, _>(key, &cache_key, |maybe_cached| {
                    // WONTPANIC: with_cached_query will only output values that are safe on this thread:
                    maybe_cached.map(|cached| cached.value_maybe_stale().value_may_panic().clone())
                })
            } else {
                None
            }
        })
    }

    /// Subscribe to the value of a query.
    /// The keyer function is reactive to changes in `K`.
    ///
    /// This will update whenever the query is created, removed, updated, refetched or set.
    ///
    /// Compared to a resource:
    /// - This will not trigger a fetch of a query, if it's not in the cache, this will be `None`.
    #[track_caller]
    pub fn subscribe_value_arc<K, MaybeKey, V, M>(
        &self,
        query_scope: impl QueryScopeTrait<K, V, M>,
        keyer: impl Fn() -> MaybeKey + Send + Sync + 'static,
    ) -> ArcSignal<Option<V>>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + Send + Sync + 'static,
        MaybeKey: QueryMaybeKey<K, V>,
        MaybeKey::MappedValue: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
    {
        let cache_key = query_scope.cache_key();
        let keyer = ArcSignal::derive(move || keyer().into_maybe_key().map(|k| KeyHash::new(&k)));

        let dyn_signal = self
            .untyped_client
            .scope_lookup
            .scope_subscriptions_mut()
            .add_value_set_updated_or_removed_subscription(
                cache_key,
                MaybeLocal::new({
                    let keyer = keyer.clone();
                    move || keyer.get()
                }),
            );

        let scope_lookup = self.untyped_client.scope_lookup;
        ArcSignal::derive(move || {
            dyn_signal.track();
            if let Some(key) = keyer.read_untracked().as_ref() {
                scope_lookup.with_cached_query::<K, V, _>(key, &cache_key, |maybe_cached| {
                    // WONTPANIC: with_cached_query will only output values that are safe on this thread:
                    maybe_cached.map(|cached| cached.value_maybe_stale().value_may_panic().clone())
                })
            } else {
                None
            }
        })
    }

    /// Mark a query as stale.
    ///
    /// Any active resources will refetch in the background, replacing them when ready.
    #[track_caller]
    pub fn invalidate_query<K, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
    ) -> bool
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        self.untyped_client.invalidate_query(query_scope, key)
    }

    /// Mark multiple queries of a specific type as stale.
    ///
    /// Any active resources will refetch in the background, replacing them when ready.
    #[track_caller]
    pub fn invalidate_queries<K, V, KRef, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keys: impl IntoIterator<Item = KRef>,
    ) -> Vec<KRef>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
        KRef: Borrow<K>,
    {
        self.untyped_client.invalidate_queries(query_scope, keys)
    }

    /// Mark one or more queries of a specific type as stale with a callback.
    ///
    /// When the callback returns `true`, the specific query will be invalidated.
    ///
    /// Any active resources subscribing to that query will refetch in the background,
    /// replacing them when ready.
    #[track_caller]
    pub fn invalidate_queries_with_predicate<K, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        should_invalidate: impl Fn(&K) -> bool,
    ) where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let mut scopes = self.untyped_client.scope_lookup.scopes_mut();
        let mut cbs_scopes = vec![];
        let query_scope_info = QueryScopeInfo::new_local(&query_scope);
        self.untyped_client.scope_lookup.with_cached_scope_mut::<K, V, _, _>(
            &mut scopes,
            query_scope_info.cache_key,
            OnScopeMissing::Skip,
            |_| {},
            |maybe_scope, _| {
                if let Some(scope) = maybe_scope {
                    for query in scope.all_queries_mut_include_pending() {
                        if let Some(key) = query.key().value_if_safe()
                            && should_invalidate(key)
                        {
                            let cb_scopes = query.invalidate(QueryAbortReason::Invalidate);
                            cbs_scopes.push(cb_scopes);
                        }
                    }
                }
            },
        );
        let mut cbs_external = vec![];
        for cb in cbs_scopes {
            if let Some(cb_external) = cb(&mut scopes) {
                cbs_external.push(cb_external);
            }
        }
        drop(scopes);
        run_external_callbacks(self.untyped_client, query_scope_info.cache_key, cbs_external);
    }

    /// Mark all queries of a specific type as stale.
    ///
    /// Any active resources will refetch in the background, replacing them when ready.
    #[track_caller]
    pub fn invalidate_query_scope<K, V, M>(&self, query_scope: impl QueryScopeLocalTrait<K, V, M>)
    where
        K: Hash + 'static,
        V: Clone + 'static,
    {
        self.invalidate_query_scope_inner(query_scope.cache_key())
    }

    pub(crate) fn invalidate_query_scope_inner(&self, scope_cache_key: ScopeCacheKey) {
        let mut scopes = self.untyped_client.scope_lookup.scopes_mut();
        let mut cbs_scopes = vec![];
        if let Some(scope) = scopes.get_mut(&scope_cache_key) {
            let cb_scopes = scope.invalidate_scope(QueryAbortReason::Invalidate);
            cbs_scopes.push(cb_scopes);
            for buster in scope.busters() {
                buster.try_set(new_buster_id());
            }
        }
        let mut cbs_external = vec![];
        for cb in cbs_scopes {
            if let Some(cb_external) = cb(&mut scopes) {
                cbs_external.push(cb_external);
            }
        }
        drop(scopes);
        run_external_callbacks(self.untyped_client, scope_cache_key, cbs_external);
    }

    /// Mark all queries as stale.
    ///
    /// Any active resources will refetch in the background, replacing them when ready.
    ///
    /// To have the cache instantly cleared and all listeners reset to pending, e.g. for user logout,
    /// see [`QueryClient::clear`].
    #[track_caller]
    pub fn invalidate_all_queries(&self) {
        let mut scopes = self.untyped_client.scope_lookup.scopes_mut();
        let mut cbs_scopes = vec![];
        for scope in scopes.values_mut() {
            let busters = scope.busters();
            let cb_scopes = scope.invalidate_scope(QueryAbortReason::Invalidate);
            cbs_scopes.push((cb_scopes, scope.cache_key()));
            for buster in busters {
                buster.try_set(new_buster_id());
            }
        }
        let mut cbs_external = vec![];
        for (cb, cache_key) in cbs_scopes {
            if let Some(cb_external) = cb(&mut scopes) {
                cbs_external.push((cb_external, cache_key));
            }
        }
        drop(scopes);
        for (cb_external, cache_key) in cbs_external {
            run_external_callbacks(self.untyped_client, cache_key, vec![cb_external]);
        }
    }

    /// Empty the cache, like [`QueryClient::invalidate_all_queries`] except:
    /// - the cache is instantly cleared of all queries
    /// - All active resources etc are reset to their pending state instantly
    ///   until the new query finishes refetching.
    ///
    /// Useful for e.g. user logout.
    ///
    /// [`QueryClient::invalidate_all_queries`] on the other hand, will only refetch
    /// active queries in the background, replacing them when ready.
    #[track_caller]
    pub fn clear(&self) {
        let mut scopes = self.untyped_client.scope_lookup.scopes_mut();
        let mut cbs_scopes = vec![];
        for scope in scopes.values_mut() {
            let busters = scope.busters();
            let cb_scopes = scope.invalidate_scope(QueryAbortReason::Clear);
            cbs_scopes.push((cb_scopes, scope.cache_key()));
            scope.clear();
            for buster in busters {
                buster.try_set(new_buster_id());
            }
        }
        let mut cbs_external = vec![];
        for (cb, cache_key) in cbs_scopes {
            if let Some(cb_external) = cb(&mut scopes) {
                cbs_external.push((cb_external, cache_key));
            }
        }
        drop(scopes);
        for (cb_external, cache_key) in cbs_external {
            run_external_callbacks(self.untyped_client, cache_key, vec![cb_external]);
        }
    }

    #[cfg(test)]
    pub(crate) fn total_cached_queries(&self) -> usize {
        self.untyped_client
            .scope_lookup
            .scopes()
            .values()
            .map(|scope| scope.total_cached_queries())
            .sum()
    }

    #[cfg(test)]
    pub(crate) fn is_key_invalid<K, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
    ) -> bool
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let query_scope_info = QueryScopeInfo::new_local(&query_scope);
        self.untyped_client.scope_lookup.with_cached_scope_mut::<K, V, _, _>(
            &mut self.untyped_client.scope_lookup.scopes_mut(),
            query_scope_info.cache_key,
            OnScopeMissing::Skip,
            |_| {},
            |maybe_scope, _| {
                if let Some(scope) = maybe_scope {
                    scope
                        .get(&KeyHash::new(key.borrow()))
                        .map(|query| query.is_invalidated())
                        .unwrap_or(false)
                } else {
                    false
                }
            },
        )
    }

    #[cfg(test)]
    #[allow(dead_code)]
    pub(crate) fn mark_key_valid<K, V, M>(&self, query_scope: impl QueryScopeLocalTrait<K, V, M>, key: impl Borrow<K>)
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        self.untyped_client.scope_lookup.with_cached_scope_mut::<K, V, _, _>(
            &mut self.untyped_client.scope_lookup.scopes_mut(),
            QueryScopeInfo::new_local(&query_scope).cache_key,
            OnScopeMissing::Skip,
            |_| {},
            |maybe_scope, _| {
                if let Some(scope) = maybe_scope
                    && let Some(query) = scope.get_mut(&KeyHash::new(key.borrow()))
                {
                    query.mark_valid();
                }
            },
        );
    }

    #[cfg(test)]
    pub(crate) fn subscriber_count(&self) -> usize {
        self.untyped_client.scope_lookup.scope_subscriptions_mut().count()
    }
}

impl UntypedQueryClient {
    #[track_caller]
    pub(crate) async fn fetch_query<K, V, M>(
        &self,
        query_scope: impl QueryScopeTrait<K, V, M>,
        key: impl Borrow<K>,
    ) -> V
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + Send + Sync + 'static,
        V: DebugIfDevtoolsEnabled + Clone + Send + Sync + 'static,
    {
        let query_scope_info = QueryScopeInfo::new(&query_scope);
        let owner_chain = OwnerChain::new(*self, query_scope_info.cache_key, Owner::current());
        self.fetch_inner(
            query_scope_info,
            || QueryScopeQueryInfo::new(&query_scope, key.borrow()),
            async |key| MaybeLocal::new(query_scope.query(key).await),
            key.borrow(),
            None,
            || MaybeLocal::new(key.borrow().clone()),
            &owner_chain,
        )
        .await
    }

    #[track_caller]
    pub(crate) async fn fetch_query_local<K, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
    ) -> V
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let query_scope_info = QueryScopeInfo::new_local(&query_scope);
        let owner_chain = OwnerChain::new(*self, query_scope_info.cache_key, Owner::current());
        self.fetch_inner(
            query_scope_info,
            || QueryScopeQueryInfo::new_local(&query_scope, key.borrow()),
            async |key| MaybeLocal::new_local(query_scope.query(key).await),
            key.borrow(),
            None,
            || MaybeLocal::new_local(key.borrow().clone()),
            &owner_chain,
        )
        .await
    }

    #[track_caller]
    async fn fetch_inner<K, V, FetcherFut>(
        &self,
        query_scope_info: QueryScopeInfo,
        query_scope_info_for_new_query: impl Fn() -> QueryScopeQueryInfo<K>,
        fetcher: impl Fn(K) -> FetcherFut,
        key: &K,
        maybe_preheld_fetcher_mutex_guard: Option<&futures::lock::MutexGuard<'_, ()>>,
        lazy_maybe_local_key: impl FnOnce() -> MaybeLocal<K>,
        owner_chain: &OwnerChain,
    ) -> V
    where
        K: DebugIfDevtoolsEnabled + Clone + Hash + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
        FetcherFut: Future<Output = MaybeLocal<V>>,
    {
        self.cached_or_fetch(
            self.options,
            query_scope_info.options,
            None,
            &query_scope_info,
            query_scope_info_for_new_query,
            key,
            fetcher,
            |info| {
                match info.variant {
                    CachedOrFetchCbInputVariant::CachedUntouched => {
                        if info.cached.stale_or_invalidated() {
                            return CachedOrFetchCbOutput::Refetch;
                        }
                    }
                    CachedOrFetchCbInputVariant::CachedUpdated => {
                        // Update anything using it:
                        info.cached.buster.set(new_buster_id());
                    }
                    CachedOrFetchCbInputVariant::Fresh => {}
                }
                CachedOrFetchCbOutput::Return(
                    // WONTPANIC: cached_or_fetch will only output values that are safe on this thread:
                    info.cached.value_maybe_stale().value_may_panic().clone(),
                )
            },
            maybe_preheld_fetcher_mutex_guard,
            lazy_maybe_local_key,
            owner_chain,
        )
        .await
    }

    #[track_caller]
    pub(crate) fn invalidate_query<K, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
    ) -> bool
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let cleared = self.invalidate_queries(query_scope, std::iter::once(key));
        !cleared.is_empty()
    }

    #[track_caller]
    pub(crate) fn invalidate_queries<K, V, KRef, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        keys: impl IntoIterator<Item = KRef>,
    ) -> Vec<KRef>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
        KRef: Borrow<K>,
    {
        self.invalidate_queries_inner::<K, V, _>(&QueryScopeInfo::new_local(&query_scope), keys)
    }

    #[track_caller]
    pub(crate) fn invalidate_queries_inner<K, V, KRef>(
        &self,
        query_scope_info: &QueryScopeInfo,
        keys: impl IntoIterator<Item = KRef>,
    ) -> Vec<KRef>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
        KRef: Borrow<K>,
    {
        let keys = keys.into_iter().collect::<Vec<_>>();
        let key_hashes = keys.iter().map(|key| KeyHash::new(key.borrow())).collect::<Vec<_>>();

        if key_hashes.is_empty() {
            return vec![];
        }

        let mut scopes = self.scope_lookup.scopes_mut();
        let mut cbs_scopes = vec![];
        let results = self.scope_lookup.with_cached_scope_mut::<K, V, _, _>(
            &mut scopes,
            query_scope_info.cache_key,
            OnScopeMissing::Skip,
            |_| {},
            |maybe_scope, _| {
                let mut invalidated = vec![];
                if let Some(scope) = maybe_scope {
                    for (key, key_hash) in keys.into_iter().zip(key_hashes.iter()) {
                        if let Some(cached) = scope.get_mut_include_pending(key_hash) {
                            let cb_scopes = cached.invalidate(QueryAbortReason::Invalidate);
                            cbs_scopes.push(cb_scopes);
                            invalidated.push(key);
                        }
                    }
                }
                invalidated
            },
        );
        let mut cbs_external = vec![];
        for cb in cbs_scopes {
            if let Some(cb_external) = cb(&mut scopes) {
                cbs_external.push(cb_external);
            }
        }
        drop(scopes);
        run_external_callbacks(*self, query_scope_info.cache_key, cbs_external);
        results
    }

    #[cfg(test)]
    /// Clear a specific query scope of all its queries.
    #[track_caller]
    pub(crate) fn clear_query_scope<K, V, M>(&self, query_scope: impl QueryScopeLocalTrait<K, V, M>)
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let mut scopes = self.scope_lookup.scopes_mut();
        let query_scope_info = QueryScopeInfo::new_local(&query_scope);
        let mut cbs_scopes = vec![];
        self.scope_lookup.with_cached_scope_mut::<K, V, _, _>(
            &mut scopes,
            query_scope_info.cache_key,
            OnScopeMissing::Skip,
            |_| {},
            |maybe_scope, _| {
                if let Some(scope) = maybe_scope {
                    for cached in scope.all_queries_mut_include_pending() {
                        let cb_scopes = cached.invalidate(QueryAbortReason::Clear);
                        cbs_scopes.push(cb_scopes);
                    }
                }
            },
        );
        let mut cbs_external = vec![];
        for cb in cbs_scopes {
            if let Some(cb_external) = cb(&mut scopes) {
                cbs_external.push(cb_external);
            }
        }
        drop(scopes);
        run_external_callbacks(*self, query_scope_info.cache_key, cbs_external);
    }

    /// Clear a specific query key.
    #[track_caller]
    pub(crate) fn clear_query<K, V, M>(
        &self,
        query_scope: impl QueryScopeLocalTrait<K, V, M>,
        key: impl Borrow<K>,
    ) -> bool
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        let mut scopes = self.scope_lookup.scopes_mut();
        let query_scope_info = QueryScopeInfo::new_local(&query_scope);
        let mut cbs_scopes = vec![];
        let result = self.scope_lookup.with_cached_scope_mut::<K, V, _, _>(
            &mut scopes,
            query_scope_info.cache_key,
            OnScopeMissing::Skip,
            |_| {},
            |maybe_scope, _| {
                if let Some(scope) = maybe_scope {
                    let key_hash = KeyHash::new(key.borrow());
                    if let Some(cached) = scope.get_mut_include_pending(&key_hash) {
                        let cb_scopes = cached.invalidate(QueryAbortReason::Clear);
                        cbs_scopes.push(cb_scopes);
                    }
                    let removed = scope.remove_entry(&key_hash);
                    // Calling it again just in case because in tests might be in sync cache and non sync cache:
                    scope.remove_entry(&key_hash);
                    return removed.is_some();
                }
                false
            },
        );
        let mut cbs_external = vec![];
        for cb in cbs_scopes {
            if let Some(cb_external) = cb(&mut scopes) {
                cbs_external.push(cb_external);
            }
        }
        drop(scopes);
        run_external_callbacks(*self, query_scope_info.cache_key, cbs_external);
        result
    }

    pub(crate) fn query_metadata<K, V>(
        &self,
        scope_cache_key: ScopeCacheKey,
        key: impl Borrow<K>,
    ) -> Option<QueryMetadata>
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
    {
        self.scope_lookup.with_cached_scope_mut::<K, V, _, _>(
            &mut self.scope_lookup.scopes_mut(),
            scope_cache_key,
            OnScopeMissing::Skip,
            |_| {},
            |maybe_scope, _| {
                if let Some(scope) = maybe_scope
                    && let Some(cached) = scope.get(&KeyHash::new(key.borrow()))
                {
                    return Some(QueryMetadata {
                        updated_at: cached.updated_at,
                        stale_or_invalidated: cached.stale_or_invalidated(),
                    });
                }
                None
            },
        )
    }

    pub async fn cached_or_fetch<K, V, T, FetcherFut>(
        &self,
        client_options: QueryOptions,
        scope_options: Option<QueryOptions>,
        maybe_buster_if_uncached: Option<ArcRwSignal<u64>>,
        query_scope_info: &QueryScopeInfo,
        query_scope_info_for_new_query: impl Fn() -> QueryScopeQueryInfo<K>,
        key: &K,
        fetcher: impl Fn(K) -> FetcherFut,
        return_cb: impl Fn(CachedOrFetchCbInput<K, V>) -> CachedOrFetchCbOutput<T>,
        maybe_preheld_fetcher_mutex_guard: Option<&futures::lock::MutexGuard<'_, ()>>,
        lazy_maybe_local_key: impl FnOnce() -> MaybeLocal<K>,
        owner_chain: &OwnerChain,
    ) -> T
    where
        K: DebugIfDevtoolsEnabled + Hash + Clone + 'static,
        V: DebugIfDevtoolsEnabled + Clone + 'static,
        FetcherFut: Future<Output = MaybeLocal<V>>,
    {
        let scope_lookup = &self.scope_lookup;
        let key_hash = KeyHash::new(key);
        let mut cached_buster = None;
        let next_directive =
            scope_lookup.with_cached_query::<K, V, _>(&key_hash, &query_scope_info.cache_key, |maybe_cached| {
                if let Some(cached) = maybe_cached {
                    cached_buster = Some(cached.buster.clone());
                    return_cb(CachedOrFetchCbInput {
                        cached,
                        variant: CachedOrFetchCbInputVariant::CachedUntouched,
                    })
                } else {
                    CachedOrFetchCbOutput::Refetch
                }
            });

        match next_directive {
            CachedOrFetchCbOutput::Return(value) => value,
            CachedOrFetchCbOutput::Refetch => {
                // Will probably need to fetch, unless someone fetches whilst trying to get hold of the fetch mutex:
                let fetcher_mutex = scope_lookup.fetcher_mutex::<K, V>(key_hash, query_scope_info);
                let _maybe_fetcher_mutex_guard_local = if maybe_preheld_fetcher_mutex_guard.is_none() {
                    let _fetcher_guard = match fetcher_mutex.try_lock() {
                        Some(fetcher_guard) => fetcher_guard,
                        None => {
                            // If have to wait, should check cache again in case it was fetched while waiting.
                            let fetcher_guard = fetcher_mutex.lock().await;
                            let next_directive = scope_lookup.with_cached_query::<K, V, _>(
                                &key_hash,
                                &query_scope_info.cache_key,
                                |maybe_cached| {
                                    if let Some(cached) = maybe_cached {
                                        cached_buster = Some(cached.buster.clone());
                                        return_cb(CachedOrFetchCbInput {
                                            cached,
                                            variant: CachedOrFetchCbInputVariant::CachedUntouched,
                                        })
                                    } else {
                                        CachedOrFetchCbOutput::Refetch
                                    }
                                },
                            );
                            match next_directive {
                                CachedOrFetchCbOutput::Return(value) => return value,
                                CachedOrFetchCbOutput::Refetch => fetcher_guard,
                            }
                        }
                    };
                    Some(_fetcher_guard)
                } else {
                    // Owned externally so not an issue.
                    None
                };

                #[cfg(any(all(debug_assertions, feature = "devtools"), feature = "devtools-always"))]
                let before_time = chrono::Utc::now();

                let loading_first_time = cached_buster.is_none();

                #[cfg(any(all(debug_assertions, feature = "devtools"), feature = "devtools-always"))]
                {
                    // Running this before fetching so it'll show up in devtools straight away:
                    if loading_first_time {
                        scope_lookup.client_subscriptions_mut().notify_query_created(
                            crate::subs_client::QueryCreatedInfo {
                                cache_key: query_scope_info.cache_key,
                                scope_title: query_scope_info.title.clone(),
                                key_hash,
                                debug_key: crate::utils::DebugValue::new(key),
                                combined_options: crate::options_combine(client_options, scope_options),
                            },
                        );
                    }
                }

                let maybe_local_key = lazy_maybe_local_key();

                enum MaybeNewValue<V> {
                    NewValue(V),
                    SsrStreamedValueOverride,
                }

                let maybe_new_value = scope_lookup
                    .with_notify_fetching(
                        query_scope_info.cache_key,
                        key_hash,
                        loading_first_time,
                        // Call the fetcher, but reset and repeat if an invalidation occurs whilst in-flight:
                        async {
                            loop {
                                let query_abort_rx = scope_lookup.prepare_invalidation_channel::<K, V>(
                                    query_scope_info,
                                    key_hash,
                                    &maybe_local_key,
                                );

                                let fut = owner_chain.with(|| fetcher(key.clone()));

                                futures::select_biased! {
                                    rx_result = query_abort_rx.fuse() => {
                                        if let Ok(reason) = rx_result {
                                            match reason {
                                                QueryAbortReason::Invalidate | QueryAbortReason::Clear => {},
                                                QueryAbortReason::SsrStreamedValueOverride => {
                                                    break MaybeNewValue::SsrStreamedValueOverride;
                                                },
                                            }
                                        }
                                    },
                                    new_value = fut.fuse() => {
                                        break MaybeNewValue::NewValue(new_value);
                                    },
                                }
                            }
                        },
                    )
                    .await;

                #[cfg(any(all(debug_assertions, feature = "devtools"), feature = "devtools-always"))]
                let elapsed_ms = chrono::Utc::now().signed_duration_since(before_time).num_milliseconds();

                let buster_if_uncached = if loading_first_time {
                    Some(maybe_buster_if_uncached.unwrap_or_else(|| ArcRwSignal::new(new_buster_id())))
                } else {
                    None
                };

                let next_directive = scope_lookup.with_cached_scope_mut::<_, _, _, _>(
                    &mut scope_lookup.scopes_mut(),
                    query_scope_info.cache_key,
                    OnScopeMissing::Create(query_scope_info),
                    |_| {},
                    |scope, _| {
                        let scope = scope.expect("provided a default");
                        match maybe_new_value {
                            MaybeNewValue::NewValue(new_value) => {
                                if let Some(cached) = scope.get_mut(&key_hash) {
                                    cached.set_value(
                                        new_value,
                                        true,
                                        #[cfg(any(
                                            all(debug_assertions, feature = "devtools"),
                                            feature = "devtools-always"
                                        ))]
                                        crate::events::Event::new(crate::events::EventVariant::Fetched { elapsed_ms }),
                                        ResetInvalidated::Reset,
                                    );
                                    return_cb(CachedOrFetchCbInput {
                                        cached,
                                        variant: CachedOrFetchCbInputVariant::CachedUpdated,
                                    })
                                } else {
                                    // We already notified before the async fetch, so it would show up sooner.
                                    scope.insert_without_query_created_notif(
                                        key_hash,
                                        Query::new(
                                            client_options,
                                            *self,
                                            query_scope_info,
                                            query_scope_info_for_new_query(),
                                            key_hash,
                                            maybe_local_key,
                                            new_value,
                                            buster_if_uncached.expect("loading_first_time means this is Some(). (bug)"),
                                            scope_options,
                                            None,
                                            #[cfg(any(
                                                all(debug_assertions, feature = "devtools"),
                                                feature = "devtools-always"
                                            ))]
                                            crate::events::Event::new(crate::events::EventVariant::Fetched {
                                                elapsed_ms,
                                            }),
                                        ),
                                    );
                                    return_cb(CachedOrFetchCbInput {
                                        cached: scope.get(&key_hash).expect("Just set. (bug)"),
                                        variant: CachedOrFetchCbInputVariant::Fresh,
                                    })
                                }
                            }
                            MaybeNewValue::SsrStreamedValueOverride => return_cb(CachedOrFetchCbInput {
                                cached: scope
                                    .get(&key_hash)
                                    .expect("Should contain value streamed from server. (bug)"),
                                variant: CachedOrFetchCbInputVariant::Fresh,
                            }),
                        }
                    },
                );

                match next_directive {
                    CachedOrFetchCbOutput::Refetch => {
                        panic!("Unexpected refetch directive after providing fresh value. (bug)")
                    }
                    CachedOrFetchCbOutput::Return(return_value) => return_value,
                }
            }
        }
    }
}

pub(crate) struct QueryMetadata {
    pub updated_at: DateTime<Utc>,
    pub stale_or_invalidated: bool,
}