wp_knowledge/

facade.rs

use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use std::{
    collections::hash_map::DefaultHasher,
    collections::{HashMap, HashSet},
    hash::{Hash, Hasher},
};

use crate::error::KnowledgeResult;
use async_trait::async_trait;
use rusqlite::ToSql;
use rusqlite::{Connection, OpenFlags};
use wp_log::{info_ctrl, warn_kdb};
use wp_model_core::model::DataField;

use crate::cache::CacheAble;
use crate::loader::{
    DEFAULT_SQLDB_NAME, ProviderKind, SqlProviderKind, SqlProviderSpec, parse_knowdb_conf,
};
use crate::mem::RowData;
use crate::mem::memdb::MemDB;
use crate::mem::thread_clone::ThreadClonedMDB;
use crate::mysql::{MySqlProvider, MySqlProviderConfig};
use crate::param::named_params_to_fields;
use crate::postgres::{PostgresProvider, PostgresProviderConfig};
use crate::runtime::{
    CachePolicy, CachedRedisValue, DatasourceId, Generation, MetadataCacheScope, ProviderExecutor,
    QueryRequest, QueryResponse, RedisCacheKey, RedisCmdTag, RuntimeSnapshot, runtime,
};
use crate::telemetry::{
    CacheLayer, CacheOutcome, CacheTelemetryEvent, KnowledgeTelemetry, install_telemetry,
    telemetry, telemetry_enabled,
};

struct MemProvider {
    db: MemDB,
    metadata_scope: MetadataCacheScope,
}

#[async_trait]
impl ProviderExecutor for ThreadClonedMDB {
    fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
        ThreadClonedMDB::query_with_scope(self, sql)
    }

    fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
        ThreadClonedMDB::query_fields_with_scope(self, sql, params)
    }

    fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
        ThreadClonedMDB::query_row_with_scope(self, sql)
    }

    fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
        ThreadClonedMDB::query_named_fields_with_scope(self, sql, params)
    }
}

#[async_trait]
impl ProviderExecutor for MemProvider {
    fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
        self.db.query_with_scope(&self.metadata_scope, sql)
    }

    fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
        self.db
            .query_fields_with_scope(&self.metadata_scope, sql, params)
    }

    fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
        self.db.query_row_with_scope(&self.metadata_scope, sql)
    }

    fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
        self.db
            .query_named_fields_with_scope(&self.metadata_scope, sql, params)
    }
}

#[async_trait]
impl ProviderExecutor for PostgresProvider {
    fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
        PostgresProvider::query(self, sql)
    }

    fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
        PostgresProvider::query_fields(self, sql, params)
    }

    fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
        PostgresProvider::query_row(self, sql)
    }

    fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
        PostgresProvider::query_named_fields(self, sql, params)
    }

    async fn query_async(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
        PostgresProvider::query_async(self, sql).await
    }

    async fn query_fields_async(
        &self,
        sql: &str,
        params: &[DataField],
    ) -> KnowledgeResult<Vec<RowData>> {
        PostgresProvider::query_fields_async(self, sql, params).await
    }

    async fn query_row_async(&self, sql: &str) -> KnowledgeResult<RowData> {
        PostgresProvider::query_row_async(self, sql).await
    }

    async fn query_named_fields_async(
        &self,
        sql: &str,
        params: &[DataField],
    ) -> KnowledgeResult<RowData> {
        PostgresProvider::query_named_fields_async(self, sql, params).await
    }
}

#[async_trait]
impl ProviderExecutor for MySqlProvider {
    fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
        MySqlProvider::query(self, sql)
    }

    fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
        MySqlProvider::query_fields(self, sql, params)
    }

    fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
        MySqlProvider::query_row(self, sql)
    }

    fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
        MySqlProvider::query_named_fields(self, sql, params)
    }

    async fn query_async(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
        MySqlProvider::query_async(self, sql).await
    }

    async fn query_fields_async(
        &self,
        sql: &str,
        params: &[DataField],
    ) -> KnowledgeResult<Vec<RowData>> {
        MySqlProvider::query_fields_async(self, sql, params).await
    }

    async fn query_row_async(&self, sql: &str) -> KnowledgeResult<RowData> {
        MySqlProvider::query_row_async(self, sql).await
    }

    async fn query_named_fields_async(
        &self,
        sql: &str,
        params: &[DataField],
    ) -> KnowledgeResult<RowData> {
        MySqlProvider::query_named_fields_async(self, sql, params).await
    }
}

fn install_provider<F>(
    kind: ProviderKind,
    datasource_id: DatasourceId,
    build: F,
) -> KnowledgeResult<()>
where
    F: FnOnce(Generation) -> KnowledgeResult<Arc<dyn ProviderExecutor>>,
{
    runtime().install_provider(kind, datasource_id, build)?;
    Ok(())
}

fn datasource_id_for(kind: ProviderKind, seed: &str) -> DatasourceId {
    DatasourceId::from_seed(kind, seed)
}

pub fn init_thread_cloned_from_authority(authority_uri: &str) -> KnowledgeResult<()> {
    let datasource_id = datasource_id_for(ProviderKind::SqliteAuthority, authority_uri);
    install_provider(ProviderKind::SqliteAuthority, datasource_id, |generation| {
        Ok(Arc::new(ThreadClonedMDB::from_authority_with_scope(
            authority_uri,
            datasource_id_for(ProviderKind::SqliteAuthority, authority_uri),
            generation.0,
        )))
    })
}

pub fn init_mem_provider(memdb: MemDB) -> KnowledgeResult<()> {
    install_provider(
        ProviderKind::SqliteAuthority,
        datasource_id_for(ProviderKind::SqliteAuthority, "memdb"),
        |generation| {
            Ok(Arc::new(MemProvider {
                db: memdb,
                metadata_scope: MetadataCacheScope {
                    datasource_id: datasource_id_for(ProviderKind::SqliteAuthority, "memdb"),
                    generation,
                },
            }))
        },
    )
}

pub fn init_postgres_provider(connection_uri: &str, pool_size: Option<u32>) -> KnowledgeResult<()> {
    init_postgres_provider_with_config(
        PostgresProviderConfig::new(connection_uri).with_pool_size(pool_size),
    )
}

pub fn init_postgres_provider_with_config(config: PostgresProviderConfig) -> KnowledgeResult<()> {
    let connection_uri = config.connection_uri().to_string();
    install_postgres_provider(DEFAULT_SQLDB_NAME, config, &connection_uri, true)
}

/// 以命名 provider 安装 PostgreSQL 连接（不替换默认 provider）。
pub fn init_postgres_provider_named(
    name: &str,
    config: PostgresProviderConfig,
) -> KnowledgeResult<()> {
    install_postgres_provider(name, config, name, false)
}

fn install_postgres_provider(
    name: &str,
    config: PostgresProviderConfig,
    datasource_seed: &str,
    is_default: bool,
) -> KnowledgeResult<()> {
    let datasource_id = datasource_id_for(ProviderKind::Postgres, datasource_seed);
    runtime().install_provider_named(
        name,
        ProviderKind::Postgres,
        datasource_id.clone(),
        |generation| {
            let provider = PostgresProvider::connect(
                name,
                &config,
                MetadataCacheScope {
                    datasource_id: datasource_id.clone(),
                    generation,
                },
            )?;
            Ok(Arc::new(provider))
        },
        is_default,
    )?;
    Ok(())
}

pub fn init_mysql_provider(connection_uri: &str, pool_size: Option<u32>) -> KnowledgeResult<()> {
    init_mysql_provider_with_config(
        MySqlProviderConfig::new(connection_uri).with_pool_size(pool_size),
    )
}

pub fn init_mysql_provider_with_config(config: MySqlProviderConfig) -> KnowledgeResult<()> {
    let connection_uri = config.connection_uri().to_string();
    install_mysql_provider(DEFAULT_SQLDB_NAME, config, &connection_uri, true)
}

/// 以命名 provider 安装 MySQL 连接（不替换默认 provider）。
pub fn init_mysql_provider_named(name: &str, config: MySqlProviderConfig) -> KnowledgeResult<()> {
    install_mysql_provider(name, config, name, false)
}

fn install_mysql_provider(
    name: &str,
    config: MySqlProviderConfig,
    datasource_seed: &str,
    is_default: bool,
) -> KnowledgeResult<()> {
    let datasource_id = datasource_id_for(ProviderKind::Mysql, datasource_seed);
    runtime().install_provider_named(
        name,
        ProviderKind::Mysql,
        datasource_id.clone(),
        |generation| {
            let provider = MySqlProvider::connect(
                &config,
                MetadataCacheScope {
                    datasource_id: datasource_id.clone(),
                    generation,
                },
            )?;
            Ok(Arc::new(provider))
        },
        is_default,
    )?;
    Ok(())
}

fn build_postgres_config(spec: &SqlProviderSpec) -> PostgresProviderConfig {
    PostgresProviderConfig::new(spec.connection_uri.clone())
        .with_pool_size(spec.pool_size)
        .with_min_connections(spec.min_connections)
        .with_acquire_timeout_ms(spec.acquire_timeout_ms)
        .with_idle_timeout_ms(spec.idle_timeout_ms)
        .with_max_lifetime_ms(spec.max_lifetime_ms)
        .with_session(spec.postgres_session.as_ref())
}

fn build_mysql_config(spec: &SqlProviderSpec) -> MySqlProviderConfig {
    MySqlProviderConfig::new(spec.connection_uri.clone())
        .with_pool_size(spec.pool_size)
        .with_min_connections(spec.min_connections)
        .with_acquire_timeout_ms(spec.acquire_timeout_ms)
        .with_idle_timeout_ms(spec.idle_timeout_ms)
        .with_max_lifetime_ms(spec.max_lifetime_ms)
}

// ---------------------------------------------------------------------------
// Redis provider API
// ---------------------------------------------------------------------------

/// Check whether an item exists in a Bloom filter.
///
/// Returns `true` if the item *may* exist, `false` if it definitely does not.
/// Requires the RedisBloom module.
pub fn redis_bf_exists(key: &str, item: &str) -> KnowledgeResult<bool> {
    if let Some(generation) = current_generation() {
        let ck = redis_cache_key(RedisCmdTag::BfExists, generation, key, &[item]);
        if let Some(CachedRedisValue::Bool(v)) = runtime().redis_cache_get(&ck) {
            return Ok(v);
        }
        let result = crate::redis::bf_exists("knowdb", key, item)?;
        runtime().redis_cache_put(ck, CachedRedisValue::Bool(result));
        return Ok(result);
    }
    crate::redis::bf_exists("knowdb", key, item)
}

/// Get the value of a hash field.
///
/// Returns `None` when the key or field does not exist.
pub fn redis_hget(key: &str, field: &str) -> KnowledgeResult<Option<String>> {
    if let Some(generation) = current_generation() {
        let ck = redis_cache_key(RedisCmdTag::HGet, generation, key, &[field]);
        if let Some(CachedRedisValue::OptString(ref v)) = runtime().redis_cache_get(&ck) {
            return Ok(v.clone());
        }
        let result = crate::redis::hget("knowdb", key, field)?;
        runtime().redis_cache_put(ck, CachedRedisValue::OptString(result.clone()));
        return Ok(result);
    }
    crate::redis::hget("knowdb", key, field)
}

/// Get the value of a key.
///
/// Returns `None` when the key does not exist.
pub fn redis_get(key: &str) -> KnowledgeResult<Option<String>> {
    if let Some(generation) = current_generation() {
        let ck = redis_cache_key(RedisCmdTag::Get, generation, key, &[]);
        if let Some(CachedRedisValue::OptString(ref v)) = runtime().redis_cache_get(&ck) {
            return Ok(v.clone());
        }
        let result = crate::redis::get("knowdb", key)?;
        runtime().redis_cache_put(ck, CachedRedisValue::OptString(result.clone()));
        return Ok(result);
    }
    crate::redis::get("knowdb", key)
}

/// Check whether a member exists in a set.
pub fn redis_set_exists(key: &str, member: &str) -> KnowledgeResult<bool> {
    if let Some(generation) = current_generation() {
        let ck = redis_cache_key(RedisCmdTag::SetExists, generation, key, &[member]);
        if let Some(CachedRedisValue::Bool(v)) = runtime().redis_cache_get(&ck) {
            return Ok(v);
        }
        let result = crate::redis::set_exists("knowdb", key, member)?;
        runtime().redis_cache_put(ck, CachedRedisValue::Bool(result));
        return Ok(result);
    }
    crate::redis::set_exists("knowdb", key, member)
}

/// Add items to a Bloom filter.
///
/// Returns a `Vec<bool>` indicating whether each item was *new* to the filter.
/// Requires the RedisBloom module.
pub fn redis_bf_add(key: &str, items: &[&str]) -> KnowledgeResult<Vec<bool>> {
    let owned: Vec<String> = items.iter().map(|s| s.to_string()).collect();
    crate::redis::bf_madd("knowdb", key, &owned)
}

/// Create a new Bloom filter.
///
/// `error_rate`: desired false positive rate (e.g. `0.01` for 1%).
/// `capacity`: expected number of items to add.
/// Requires the RedisBloom module.
pub fn redis_bf_create(key: &str, error_rate: f64, capacity: i64) -> KnowledgeResult<()> {
    crate::redis::bf_reserve("knowdb", key, error_rate, capacity)
}

// ---------------------------------------------------------------------------
// External call API — named queries from [fun.<name>]
// ---------------------------------------------------------------------------

/// Register [fun] definitions from knowdb.toml. Called once during init.
pub(crate) fn register_fun_map(map: HashMap<String, crate::loader::FunSpec>) {
    crate::fun::register_fun_map(map);
}

/// Execute a [fun.<name>] named query where returns = "bool".
pub fn external_exists(service: &str, arg: &str) -> KnowledgeResult<bool> {
    let spec = crate::fun::resolve_spec(service, true)?;
    let redis_key = spec.key.as_deref().unwrap_or(arg);
    if spec.enabled()
        && let Some(generation) = current_generation()
    {
        let ck = redis_cache_key(RedisCmdTag::BfExists, generation, redis_key, &[arg]);
        if let Some(CachedRedisValue::Bool(v)) = runtime().redis_cache_get(&ck) {
            return Ok(v);
        }
        let result = crate::fun::external_exists(service, arg)?;
        if let Some(ttl) = spec.ttl_ms {
            runtime().redis_cache_put_with_ttl(ck, CachedRedisValue::Bool(result), ttl);
        } else {
            runtime().redis_cache_put(ck, CachedRedisValue::Bool(result));
        }
        return Ok(result);
    }
    crate::fun::external_exists(service, arg)
}

/// Execute a [fun.<name>] named query where returns = "value".
pub fn external_value(service: &str, arg: &str) -> KnowledgeResult<Option<String>> {
    let spec = crate::fun::resolve_spec(service, false)?;
    let redis_key = spec.key.as_deref().unwrap_or(arg);
    if spec.enabled()
        && let Some(generation) = current_generation()
    {
        let ck = redis_cache_key(RedisCmdTag::Get, generation, redis_key, &[arg]);
        if let Some(CachedRedisValue::OptString(ref v)) = runtime().redis_cache_get(&ck) {
            return Ok(v.clone());
        }
        let result = crate::fun::external_value(service, arg)?;
        if let Some(ttl) = spec.ttl_ms {
            runtime().redis_cache_put_with_ttl(
                ck,
                CachedRedisValue::OptString(result.clone()),
                ttl,
            );
        } else {
            runtime().redis_cache_put(ck, CachedRedisValue::OptString(result.clone()));
        }
        return Ok(result);
    }
    crate::fun::external_value(service, arg)
}

#[allow(dead_code)]
pub(crate) fn redis_ping() -> KnowledgeResult<bool> {
    crate::redis::ping_blocking("knowdb")
}

#[allow(dead_code)]
pub(crate) fn redis_close() -> KnowledgeResult<()> {
    crate::redis::close(Some("knowdb"))
}

fn redis_cache_key(cmd: RedisCmdTag, generation: u64, key: &str, args: &[&str]) -> RedisCacheKey {
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};
    let mut hasher = DefaultHasher::new();
    key.hash(&mut hasher);
    let key_hash = hasher.finish();
    let mut hasher = DefaultHasher::new();
    for arg in args {
        arg.hash(&mut hasher);
    }
    let args_hash = hasher.finish();
    RedisCacheKey {
        generation,
        cmd_tag: cmd,
        key_hash,
        args_hash,
    }
}

// ---------------------------------------------------------------------------

pub fn current_generation() -> Option<u64> {
    runtime()
        .current_generation()
        .map(|generation| generation.0)
}

pub fn runtime_snapshot() -> RuntimeSnapshot {
    runtime().snapshot()
}

pub fn install_runtime_telemetry(
    telemetry_impl: Arc<dyn KnowledgeTelemetry>,
) -> Arc<dyn KnowledgeTelemetry> {
    install_telemetry(telemetry_impl)
}

pub use crate::sql_route::route_provider_sql;

/// 指定名称的 provider 是否已安装。
pub fn provider_exists(name: &str) -> bool {
    runtime().provider_exists(name)
}

/// 当前已安装的命名 provider 名称列表。
pub fn provider_names() -> Vec<String> {
    runtime().provider_names()
}

/// 对指定命名 provider 执行查询（全部行）。
pub fn query_for(name: &str, sql: &str) -> KnowledgeResult<Vec<RowData>> {
    runtime()
        .execute_for(
            name,
            &QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass),
        )
        .map(QueryResponse::into_rows)
}

/// 对指定命名 provider 异步执行查询（全部行）。
pub async fn query_async_for(name: &str, sql: &str) -> KnowledgeResult<Vec<RowData>> {
    runtime()
        .execute_for_async(
            name,
            &QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass),
        )
        .await
        .map(QueryResponse::into_rows)
}

/// 对指定命名 provider 执行首行查询。
pub fn query_fields_for(name: &str, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
    runtime().execute_first_row_fields_for(name, sql, params, CachePolicy::Bypass)
}

/// 对指定命名 provider 异步执行首行查询。
pub async fn query_fields_for_async(
    name: &str,
    sql: &str,
    params: &[DataField],
) -> KnowledgeResult<RowData> {
    runtime()
        .execute_first_row_fields_for_async(name, sql, params, CachePolicy::Bypass)
        .await
}

pub fn query(sql: &str) -> KnowledgeResult<Vec<RowData>> {
    runtime()
        .execute(&QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass))
        .map(QueryResponse::into_rows)
}

pub async fn query_async(sql: &str) -> KnowledgeResult<Vec<RowData>> {
    runtime()
        .execute_async(&QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass))
        .await
        .map(QueryResponse::into_rows)
}

pub fn query_row(sql: &str) -> KnowledgeResult<RowData> {
    runtime()
        .execute(&QueryRequest::first_row(
            sql,
            Vec::new(),
            CachePolicy::Bypass,
        ))
        .map(QueryResponse::into_row)
}

pub async fn query_row_async(sql: &str) -> KnowledgeResult<RowData> {
    runtime()
        .execute_async(&QueryRequest::first_row(
            sql,
            Vec::new(),
            CachePolicy::Bypass,
        ))
        .await
        .map(QueryResponse::into_row)
}

pub fn query_fields(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
    runtime().execute_first_row_fields(sql, params, CachePolicy::Bypass)
}

pub async fn query_fields_async(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
    runtime()
        .execute_first_row_fields_async(sql, params, CachePolicy::Bypass)
        .await
}

pub fn query_named_fields(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
    query_fields(sql, params)
}

pub async fn query_named_fields_async(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
    query_fields_async(sql, params).await
}

pub fn query_named<'a>(
    sql: &str,
    params: &'a [(&'a str, &'a dyn ToSql)],
) -> KnowledgeResult<RowData> {
    let fields = named_params_to_fields(params)?;
    query_fields(sql, &fields)
}

pub fn cache_query_fields<const N: usize>(
    sql: &str,
    c_params: &[DataField; N],
    query_params: &[DataField],
    cache: &mut impl CacheAble<DataField, RowData, N>,
) -> RowData {
    cache_query_fields_with_scope(sql, stable_hash(sql), c_params, query_params, cache)
}

pub fn cache_query_fields_with_scope<const N: usize>(
    sql: &str,
    local_cache_scope: u64,
    c_params: &[DataField; N],
    query_params: &[DataField],
    cache: &mut impl CacheAble<DataField, RowData, N>,
) -> RowData {
    let generation = current_generation();
    let provider_kind = runtime().current_provider_kind();
    if let Some(hit) = local_cache_hit(
        local_cache_scope,
        c_params,
        cache,
        generation,
        provider_kind,
    ) {
        return hit;
    }
    match runtime().execute_first_row_fields(sql, query_params, CachePolicy::UseGlobal) {
        Ok(row) => {
            cache.save_scoped(local_cache_scope, c_params, row.clone());
            row
        }
        Err(err) => {
            warn_kdb!("[kdb] query error: {}", err);
            Vec::new()
        }
    }
}

/// 本地缓存路由派发：`provider` 为 `Some(name)` 时走命名 provider，`None` 时走默认 provider。
pub fn cache_query_fields_route<const N: usize>(
    provider: Option<&str>,
    sql: &str,
    local_cache_scope: u64,
    c_params: &[DataField; N],
    query_params: &[DataField],
    cache: &mut impl CacheAble<DataField, RowData, N>,
) -> RowData {
    let Some(name) = provider else {
        return cache_query_fields_with_scope(
            sql,
            local_cache_scope,
            c_params,
            query_params,
            cache,
        );
    };
    let handle = match runtime().provider_by_name(name) {
        Ok(handle) => handle,
        Err(err) => {
            warn_kdb!("[kdb] named provider '{name}' unavailable: {}", err);
            return Vec::new();
        }
    };
    let generation = Some(handle.generation.0);
    let provider_kind = Some(handle.kind.clone());
    // provider 粒度 scope：与默认 provider 区分，使本地缓存代际按 provider 独立跟踪，
    // 混用命名库与默认库时不互相重置。
    let scope = local_cache_scope ^ stable_hash(name);
    if let Some(hit) = local_cache_hit(scope, c_params, cache, generation, provider_kind) {
        return hit;
    }
    match runtime().execute_first_row_fields_for(name, sql, query_params, CachePolicy::UseGlobal) {
        Ok(row) => {
            cache.save_scoped(scope, c_params, row.clone());
            row
        }
        Err(err) => {
            warn_kdb!("[kdb] query error: {}", err);
            Vec::new()
        }
    }
}

/// 本地缓存命中检查（含 generation 准备）与 miss 记录。命中时返回 `Some(hit)`。
fn local_cache_hit<const N: usize>(
    local_cache_scope: u64,
    c_params: &[DataField; N],
    cache: &mut impl CacheAble<DataField, RowData, N>,
    generation: Option<u64>,
    provider_kind: Option<ProviderKind>,
) -> Option<RowData> {
    if let Some(generation) = generation {
        cache.prepare_generation(local_cache_scope, generation);
    }
    if let Some(hit) = cache.fetch_scoped(local_cache_scope, c_params) {
        runtime().record_local_cache_hit();
        if telemetry_enabled() {
            telemetry().on_cache(&CacheTelemetryEvent {
                layer: CacheLayer::Local,
                outcome: CacheOutcome::Hit,
                provider_kind,
            });
        }
        return Some(hit.clone());
    }
    runtime().record_local_cache_miss();
    if telemetry_enabled() {
        telemetry().on_cache(&CacheTelemetryEvent {
            layer: CacheLayer::Local,
            outcome: CacheOutcome::Miss,
            provider_kind,
        });
    }
    None
}

pub async fn cache_query_fields_async<const N: usize>(
    sql: &str,
    c_params: &[DataField; N],
    query_params: &[DataField],
    cache: &mut impl CacheAble<DataField, RowData, N>,
) -> RowData {
    cache_query_fields_async_with_scope(
        sql,
        stable_hash(sql),
        c_params,
        || query_params.to_vec(),
        cache,
    )
    .await
}

pub async fn cache_query_fields_async_with<const N: usize, F>(
    sql: &str,
    c_params: &[DataField; N],
    build_query_params: F,
    cache: &mut impl CacheAble<DataField, RowData, N>,
) -> RowData
where
    F: FnOnce() -> Vec<DataField>,
{
    cache_query_fields_async_with_scope(sql, stable_hash(sql), c_params, build_query_params, cache)
        .await
}

pub async fn cache_query_fields_async_with_scope<const N: usize, F>(
    sql: &str,
    local_cache_scope: u64,
    c_params: &[DataField; N],
    build_query_params: F,
    cache: &mut impl CacheAble<DataField, RowData, N>,
) -> RowData
where
    F: FnOnce() -> Vec<DataField>,
{
    let generation = current_generation();
    let provider_kind = runtime().current_provider_kind();
    if let Some(hit) = local_cache_hit(
        local_cache_scope,
        c_params,
        cache,
        generation,
        provider_kind,
    ) {
        return hit;
    }
    let query_params = build_query_params();
    match runtime()
        .execute_first_row_fields_async(sql, &query_params, CachePolicy::UseGlobal)
        .await
    {
        Ok(row) => {
            cache.save_scoped(local_cache_scope, c_params, row.clone());
            row
        }
        Err(err) => {
            warn_kdb!("[kdb] query error: {}", err);
            Vec::new()
        }
    }
}

/// 异步本地缓存路由派发：`provider` 为 `Some(name)` 时走命名 provider，`None` 时走默认 provider。
pub async fn cache_query_fields_route_async<const N: usize, F>(
    provider: Option<&str>,
    sql: &str,
    local_cache_scope: u64,
    c_params: &[DataField; N],
    build_query_params: F,
    cache: &mut impl CacheAble<DataField, RowData, N>,
) -> RowData
where
    F: FnOnce() -> Vec<DataField>,
{
    let Some(name) = provider else {
        return cache_query_fields_async_with_scope(
            sql,
            local_cache_scope,
            c_params,
            build_query_params,
            cache,
        )
        .await;
    };
    let handle = match runtime().provider_by_name(name) {
        Ok(handle) => handle,
        Err(err) => {
            warn_kdb!("[kdb] named provider '{name}' unavailable: {}", err);
            return Vec::new();
        }
    };
    let generation = Some(handle.generation.0);
    let provider_kind = Some(handle.kind.clone());
    // provider 粒度 scope：与默认 provider 区分，本地缓存代际按 provider 独立跟踪。
    let scope = local_cache_scope ^ stable_hash(name);
    if let Some(hit) = local_cache_hit(scope, c_params, cache, generation, provider_kind) {
        return hit;
    }
    let query_params = build_query_params();
    match runtime()
        .execute_first_row_fields_for_async(name, sql, &query_params, CachePolicy::UseGlobal)
        .await
    {
        Ok(row) => {
            cache.save_scoped(scope, c_params, row.clone());
            row
        }
        Err(err) => {
            warn_kdb!("[kdb] query error: {}", err);
            Vec::new()
        }
    }
}

pub fn cache_query<const N: usize>(
    sql: &str,
    c_params: &[DataField; N],
    named_params: &[(&str, &dyn ToSql)],
    cache: &mut impl CacheAble<DataField, RowData, N>,
) -> RowData {
    let query_fields = match named_params_to_fields(named_params) {
        Ok(fields) => fields,
        Err(err) => {
            warn_kdb!("[kdb] query param conversion error: {}", err);
            return Vec::new();
        }
    };

    cache_query_fields(sql, c_params, &query_fields, cache)
}

pub async fn cache_query_async<const N: usize>(
    sql: &str,
    c_params: &[DataField; N],
    named_params: &[(&str, &dyn ToSql)],
    cache: &mut impl CacheAble<DataField, RowData, N>,
) -> RowData {
    let query_fields = match named_params_to_fields(named_params) {
        Ok(fields) => fields,
        Err(err) => {
            warn_kdb!("[kdb] query param conversion error: {}", err);
            return Vec::new();
        }
    };

    cache_query_fields_async(sql, c_params, &query_fields, cache).await
}

fn ensure_wal(authority_uri: &str) -> KnowledgeResult<()> {
    if let Ok(conn) = Connection::open_with_flags(
        authority_uri,
        OpenFlags::SQLITE_OPEN_READ_WRITE
            | OpenFlags::SQLITE_OPEN_CREATE
            | OpenFlags::SQLITE_OPEN_URI,
    ) {
        let _ = conn.execute_batch(
            "PRAGMA journal_mode=WAL;\nPRAGMA synchronous=NORMAL;\nPRAGMA temp_store=MEMORY;",
        );
    }
    Ok(())
}

pub fn init_wal_pool_from_authority(authority_uri: &str, pool_size: u32) -> KnowledgeResult<()> {
    ensure_wal(authority_uri)?;
    let flags = OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI;
    let mem = MemDB::new_file(authority_uri, pool_size, flags)?;
    init_mem_provider(mem)
}

pub fn init_thread_cloned_from_knowdb(
    root: &Path,
    knowdb_conf: &Path,
    authority_uri: &str,
    dict: &orion_variate::EnvDict,
) -> KnowledgeResult<()> {
    let (conf, conf_abs, _) = parse_knowdb_conf(root, knowdb_conf, dict)?;
    if let Some(provider_cfg) = conf.provider() {
        // New-style [provider.sqldb] / [[provider.sqldb]]
        if let Some(specs) = provider_cfg.sqldb
            && !specs.is_empty()
        {
            let expected = SqlProviderSpec::validate_specs(&specs)?;
            // 默认 provider：生效名为 `default`（历史单库）者优先，否则取第一个。
            let default_name = if expected.contains(DEFAULT_SQLDB_NAME) {
                DEFAULT_SQLDB_NAME.to_string()
            } else {
                specs[0].effective_name().to_string()
            };
            // 外部 SQL provider 只读（仅 SELECT），初始化中途失败时允许部分安装：
            // 已连通的库照常可用、连不上的库不可查询，错误由调用方记录后继续。
            // 因此不做"先全部建池再统一提交"的事务化处理。
            for spec in &specs {
                let name = spec.effective_name().to_string();
                let is_default = name == default_name;
                // 无显式 name 的历史单库保留 connection_uri 作为 datasource 种子
                let seed = match &spec.name {
                    Some(_) => name.clone(),
                    None => spec.connection_uri.clone(),
                };
                match spec.kind {
                    SqlProviderKind::Postgres => {
                        info_ctrl!(
                            "init postgres knowdb provider({name}:{}) ",
                            conf_abs.display(),
                        );
                        install_postgres_provider(
                            &name,
                            build_postgres_config(spec),
                            &seed,
                            is_default,
                        )?;
                    }
                    SqlProviderKind::Mysql => {
                        info_ctrl!("init mysql knowdb provider({name}:{}) ", conf_abs.display(),);
                        install_mysql_provider(&name, build_mysql_config(spec), &seed, is_default)?;
                    }
                }
            }
            runtime().configure_result_cache(
                conf.cache.enabled,
                conf.cache.capacity,
                Duration::from_millis(conf.cache.ttl_ms.max(1)),
            );
            runtime().prune_named_providers_except(&expected);
            return Ok(());
        }
        // New-style [provider.redis]
        if let Some(redis_cfg) = provider_cfg.redis {
            info_ctrl!("init redis knowdb provider({}) ", conf_abs.display(),);
            crate::redis::init_with_opts(
                "knowdb",
                &redis_cfg.connection_uri,
                redis_cfg.pool_size,
                redis_cfg.command_timeout_ms,
            )?;
            runtime().configure_redis_cache(conf.cache.enabled, conf.cache.capacity);
            register_fun_map(conf.fun.clone());
            // redis-only 配置不注册任何 sqldb 命名 provider，清空过期命名注册
            runtime().prune_named_providers_except(&HashSet::new());
            return Ok(());
        }
    }

    crate::loader::build_authority_from_knowdb(root, knowdb_conf, authority_uri, dict)?;
    let ro_uri = if let Some(rest) = authority_uri.strip_prefix("file:") {
        let path_part = rest.split('?').next().unwrap_or(rest);
        format!("file:{}?mode=ro&uri=true", path_part)
    } else {
        authority_uri.to_string()
    };

    info_ctrl!("init authority knowdb success({}) ", knowdb_conf.display(),);
    init_thread_cloned_from_authority(&ro_uri)?;
    runtime().configure_result_cache(
        conf.cache.enabled,
        conf.cache.capacity,
        Duration::from_millis(conf.cache.ttl_ms.max(1)),
    );
    // 本地 authority 模式：仅保留默认 provider，清理残留的命名 provider
    runtime().prune_named_providers_except(&HashSet::from([DEFAULT_SQLDB_NAME.to_string()]));
    Ok(())
}

fn stable_hash(value: &str) -> u64 {
    let mut hasher = DefaultHasher::new();
    value.hash(&mut hasher);
    hasher.finish()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cache::FieldQueryCache;
    use crate::error::KnowReason;
    use crate::mem::memdb::MemDB;
    use crate::mem::query_util::{COLNAME_CACHE, metadata_cache_key_for_scope};
    use crate::runtime::fields_to_params;
    use crate::telemetry::{
        CacheLayer, CacheTelemetryEvent, KnowledgeTelemetry, QueryTelemetryEvent,
        ReloadTelemetryEvent, reset_telemetry,
    };
    use orion_error::conversion::ToStructError;
    use orion_variate::EnvDict;
    use std::fs;
    use std::hint::black_box;
    use std::path::PathBuf;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::{Duration, Instant};

    #[derive(Default)]
    struct TestTelemetry {
        reload_success: AtomicU64,
        reload_failure: AtomicU64,
        local_hits: AtomicU64,
        local_misses: AtomicU64,
        result_hits: AtomicU64,
        result_misses: AtomicU64,
        metadata_hits: AtomicU64,
        metadata_misses: AtomicU64,
        query_success: AtomicU64,
        query_failure: AtomicU64,
    }

    impl KnowledgeTelemetry for TestTelemetry {
        fn on_cache(&self, event: &CacheTelemetryEvent) {
            let counter = match (event.layer, event.outcome) {
                (CacheLayer::Local, CacheOutcome::Hit) => &self.local_hits,
                (CacheLayer::Local, CacheOutcome::Miss) => &self.local_misses,
                (CacheLayer::Result, CacheOutcome::Hit) => &self.result_hits,
                (CacheLayer::Result, CacheOutcome::Miss) => &self.result_misses,
                (CacheLayer::Metadata, CacheOutcome::Hit) => &self.metadata_hits,
                (CacheLayer::Metadata, CacheOutcome::Miss) => &self.metadata_misses,
            };
            counter.fetch_add(1, Ordering::Relaxed);
        }

        fn on_reload(&self, event: &ReloadTelemetryEvent) {
            let counter = match event.outcome {
                crate::telemetry::ReloadOutcome::Success => &self.reload_success,
                crate::telemetry::ReloadOutcome::Failure => &self.reload_failure,
            };
            counter.fetch_add(1, Ordering::Relaxed);
        }

        fn on_query(&self, event: &QueryTelemetryEvent) {
            let counter = if event.success {
                &self.query_success
            } else {
                &self.query_failure
            };
            counter.fetch_add(1, Ordering::Relaxed);
        }
    }

    fn perf_env_usize(key: &str, default: usize) -> usize {
        std::env::var(key)
            .ok()
            .and_then(|value| value.parse::<usize>().ok())
            .unwrap_or(default)
    }

    fn seed_perf_provider(rows: usize) {
        let db = MemDB::instance();
        db.execute("CREATE TABLE perf_kv (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create perf_kv");
        db.execute("BEGIN IMMEDIATE").expect("begin perf_kv load");
        for id in 1..=rows {
            let sql = format!("INSERT INTO perf_kv (id, value) VALUES ({id}, 'value_{id}')");
            db.execute(sql.as_str()).expect("insert perf_kv row");
        }
        db.execute("COMMIT").expect("commit perf_kv load");
        init_mem_provider(db).expect("init perf provider");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn query_async_works_with_mem_provider() {
        let _guard = crate::runtime::runtime_test_guard().lock_async().await;
        let db = MemDB::instance();
        db.execute("CREATE TABLE async_kv (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create async_kv");
        db.execute("INSERT INTO async_kv (id, value) VALUES (1, 'hello')")
            .expect("insert async_kv row");
        init_mem_provider(db).expect("init mem provider");

        let row = query_fields_async(
            "SELECT value FROM async_kv WHERE id=:id",
            &[DataField::from_digit(":id", 1)],
        )
        .await
        .expect("query async row");
        assert_eq!(row.len(), 1);
        assert_eq!(row[0].to_string(), "chars(hello)");
    }

    #[derive(Clone)]
    struct PerfQuery {
        cache_key: [DataField; 1],
        query_params: [DataField; 1],
        bypass_req: QueryRequest,
        global_req: QueryRequest,
    }

    fn build_perf_workload(ops: usize, hotset: usize) -> Vec<PerfQuery> {
        (0..ops)
            .map(|idx| {
                let id = ((idx * 17) % hotset + 1) as i64;
                let cache_key = [DataField::from_digit("id", id)];
                let query_params = [DataField::from_digit(":id", id)];
                let bypass_req = QueryRequest::first_row(
                    "SELECT value FROM perf_kv WHERE id=:id",
                    fields_to_params(&query_params),
                    CachePolicy::Bypass,
                );
                let global_req = QueryRequest::first_row(
                    "SELECT value FROM perf_kv WHERE id=:id",
                    fields_to_params(&query_params),
                    CachePolicy::UseGlobal,
                );
                PerfQuery {
                    cache_key,
                    query_params,
                    bypass_req,
                    global_req,
                }
            })
            .collect()
    }

    #[derive(Debug, Clone, Copy)]
    struct PerfCounters {
        result_hits: u64,
        result_misses: u64,
        local_hits: u64,
        local_misses: u64,
        metadata_hits: u64,
        metadata_misses: u64,
    }

    #[derive(Debug, Clone)]
    struct PerfResult {
        name: &'static str,
        elapsed: Duration,
        ops: usize,
        counters: PerfCounters,
    }

    impl PerfResult {
        fn qps(&self) -> f64 {
            let secs = self.elapsed.as_secs_f64();
            if secs == 0.0 {
                self.ops as f64
            } else {
                self.ops as f64 / secs
            }
        }
    }

    fn snapshot_delta(before: &RuntimeSnapshot, after: &RuntimeSnapshot) -> PerfCounters {
        PerfCounters {
            result_hits: after
                .result_cache_hits
                .saturating_sub(before.result_cache_hits),
            result_misses: after
                .result_cache_misses
                .saturating_sub(before.result_cache_misses),
            local_hits: after
                .local_cache_hits
                .saturating_sub(before.local_cache_hits),
            local_misses: after
                .local_cache_misses
                .saturating_sub(before.local_cache_misses),
            metadata_hits: after
                .metadata_cache_hits
                .saturating_sub(before.metadata_cache_hits),
            metadata_misses: after
                .metadata_cache_misses
                .saturating_sub(before.metadata_cache_misses),
        }
    }

    fn run_bypass_perf(workload: &[PerfQuery], rows: usize) -> PerfResult {
        seed_perf_provider(rows);
        let before = runtime_snapshot();
        let started = Instant::now();
        for item in workload {
            let row = runtime()
                .execute(&item.bypass_req)
                .expect("execute bypass request")
                .into_row();
            black_box(row);
        }
        let elapsed = started.elapsed();
        let after = runtime_snapshot();
        PerfResult {
            name: "bypass",
            elapsed,
            ops: workload.len(),
            counters: snapshot_delta(&before, &after),
        }
    }

    fn run_global_cache_perf(workload: &[PerfQuery], rows: usize) -> PerfResult {
        seed_perf_provider(rows);
        let before = runtime_snapshot();
        let started = Instant::now();
        for item in workload {
            let row = runtime()
                .execute(&item.global_req)
                .expect("execute global-cache request")
                .into_row();
            black_box(row);
        }
        let elapsed = started.elapsed();
        let after = runtime_snapshot();
        PerfResult {
            name: "global_cache",
            elapsed,
            ops: workload.len(),
            counters: snapshot_delta(&before, &after),
        }
    }

    fn run_local_cache_perf(workload: &[PerfQuery], rows: usize) -> PerfResult {
        seed_perf_provider(rows);
        let mut cache = FieldQueryCache::with_capacity(workload.len().max(1));
        let before = runtime_snapshot();
        let started = Instant::now();
        for item in workload {
            let row = cache_query_fields(
                "SELECT value FROM perf_kv WHERE id=:id",
                &item.cache_key,
                &item.query_params,
                &mut cache,
            );
            black_box(row);
        }
        let elapsed = started.elapsed();
        let after = runtime_snapshot();
        PerfResult {
            name: "local_cache",
            elapsed,
            ops: workload.len(),
            counters: snapshot_delta(&before, &after),
        }
    }

    fn print_perf_result(result: &PerfResult) {
        eprintln!(
            "[wp-knowledge][cache-perf] scenario={} elapsed_ms={} qps={:.0} result_hit={} result_miss={} local_hit={} local_miss={} metadata_hit={} metadata_miss={}",
            result.name,
            result.elapsed.as_millis(),
            result.qps(),
            result.counters.result_hits,
            result.counters.result_misses,
            result.counters.local_hits,
            result.counters.local_misses,
            result.counters.metadata_hits,
            result.counters.metadata_misses,
        );
    }

    fn uniq_cache_cfg_tmp_dir() -> PathBuf {
        use rand::{Rng, rng};
        let rnd: u64 = rng().next_u64();
        std::env::temp_dir().join(format!("wpk_cache_cfg_{}", rnd))
    }

    fn write_minimal_knowdb_with_cache(
        root: &Path,
        enabled: bool,
        capacity: usize,
        ttl_ms: u64,
    ) -> std::path::PathBuf {
        let models = root.join("models").join("knowledge");
        let example_dir = models.join("example");
        fs::create_dir_all(&example_dir).expect("create knowdb models/example");
        fs::write(
            models.join("knowdb.toml"),
            format!(
                r#"
version = 2
base_dir = "."

[cache]
enabled = {enabled}
capacity = {capacity}
ttl_ms = {ttl_ms}

[csv]
has_header = false

[[tables]]
name = "example"
columns.by_index = [0,1]
"#
            ),
        )
        .expect("write knowdb.toml");
        fs::write(
            example_dir.join("create.sql"),
            r#"
CREATE TABLE IF NOT EXISTS {table} (
  id INTEGER PRIMARY KEY,
  value TEXT NOT NULL
);
"#,
        )
        .expect("write create.sql");
        fs::write(
            example_dir.join("insert.sql"),
            "INSERT INTO {table} (id, value) VALUES (?1, ?2);\n",
        )
        .expect("write insert.sql");
        fs::write(example_dir.join("data.csv"), "1,alpha\n").expect("write data.csv");
        models.join("knowdb.toml")
    }

    fn write_provider_only_knowdb_with_cache(
        root: &Path,
        provider_kind: &str,
        connection_uri: &str,
        enabled: bool,
        capacity: usize,
        ttl_ms: u64,
    ) -> std::path::PathBuf {
        let models = root.join("models").join("knowledge");
        fs::create_dir_all(&models).expect("create knowdb models");
        fs::write(
            models.join("knowdb.toml"),
            format!(
                r#"
version = 2
base_dir = "."

[cache]
enabled = {enabled}
capacity = {capacity}
ttl_ms = {ttl_ms}

[provider.sqldb]
kind = "{provider_kind}"
connection_uri = "{connection_uri}"
"#
            ),
        )
        .expect("write provider knowdb.toml");
        models.join("knowdb.toml")
    }

    fn restore_default_result_cache_config() {
        runtime().configure_result_cache(true, 1024, Duration::from_millis(30_000));
    }

    #[test]
    fn provider_can_be_replaced() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let db1 = MemDB::instance();
        db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table in db1");
        db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
            .expect("seed db1");
        init_mem_provider(db1).expect("init provider db1");
        let row = query_row("SELECT value FROM t WHERE id = 1").expect("query db1");
        assert_eq!(row[0].to_string(), "chars(first)");

        let db2 = MemDB::instance();
        db2.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table in db2");
        db2.execute("INSERT INTO t (id, value) VALUES (1, 'second')")
            .expect("seed db2");
        init_mem_provider(db2).expect("replace provider with db2");
        let row = query_row("SELECT value FROM t WHERE id = 1").expect("query db2");
        assert_eq!(row[0].to_string(), "chars(second)");
    }

    #[test]
    fn sqlite_metadata_cache_uses_provider_scope_after_reload() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        COLNAME_CACHE.write().expect("metadata cache lock").clear();

        let db = MemDB::instance();
        db.execute("CREATE TABLE cache_scope_t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table");
        db.execute("INSERT INTO cache_scope_t (id, value) VALUES (1, 'scope-old')")
            .expect("seed table");

        let old_scope = MetadataCacheScope {
            datasource_id: DatasourceId("sqlite:old".to_string()),
            generation: Generation(1),
        };
        let new_scope = MetadataCacheScope {
            datasource_id: DatasourceId("sqlite:new".to_string()),
            generation: Generation(2),
        };
        let old_provider = MemProvider {
            db: db.clone(),
            metadata_scope: old_scope.clone(),
        };

        install_provider(
            ProviderKind::SqliteAuthority,
            new_scope.datasource_id.clone(),
            |_generation| {
                Ok(Arc::new(MemProvider {
                    db: db.clone(),
                    metadata_scope: new_scope.clone(),
                }))
            },
        )
        .expect("install new provider");

        let row = old_provider
            .query_row("SELECT value FROM cache_scope_t WHERE id = 1")
            .expect("old provider query");
        assert_eq!(row[0].to_string(), "chars(scope-old)");

        let cache = COLNAME_CACHE.read().expect("metadata cache lock");
        assert!(cache.contains(&metadata_cache_key_for_scope(
            &old_scope,
            "SELECT value FROM cache_scope_t WHERE id = 1",
        )));
        assert!(!cache.contains(&metadata_cache_key_for_scope(
            &new_scope,
            "SELECT value FROM cache_scope_t WHERE id = 1",
        )));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn async_query_uses_runtime_bridge() {
        let _guard = crate::runtime::runtime_test_guard().lock_async().await;
        let db = MemDB::instance();
        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table");
        db.execute("INSERT INTO t (id, value) VALUES (1, 'async-first')")
            .expect("seed table");
        init_mem_provider(db).expect("init provider");

        let row = query_row_async("SELECT value FROM t WHERE id = 1")
            .await
            .expect("async query row");
        assert_eq!(row[0].to_string(), "chars(async-first)");
    }

    #[tokio::test(flavor = "current_thread")]
    async fn async_cache_query_fields_hits_local_cache() {
        let _guard = crate::runtime::runtime_test_guard().lock_async().await;
        let db = MemDB::instance();
        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table");
        db.execute("INSERT INTO t (id, value) VALUES (1, 'async-cache')")
            .expect("seed table");
        init_mem_provider(db).expect("init provider");

        let key = [DataField::from_digit("id", 1)];
        let params = [DataField::from_digit(":id", 1)];
        let mut cache = FieldQueryCache::default();

        let first = cache_query_fields_async(
            "SELECT value FROM t WHERE id=:id",
            &key,
            &params,
            &mut cache,
        )
        .await;
        let second = cache_query_fields_async(
            "SELECT value FROM t WHERE id=:id",
            &key,
            &params,
            &mut cache,
        )
        .await;

        assert_eq!(first[0].to_string(), "chars(async-cache)");
        assert_eq!(second[0].to_string(), "chars(async-cache)");
    }

    #[test]
    fn local_cache_is_cleared_when_generation_changes() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let db1 = MemDB::instance();
        db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table in db1");
        db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
            .expect("seed db1");
        init_mem_provider(db1).expect("init provider db1");

        let key = [DataField::from_digit("id", 1)];
        let params = [DataField::from_digit(":id", 1)];
        let mut cache = FieldQueryCache::default();
        let row = cache_query_fields(
            "SELECT value FROM t WHERE id=:id",
            &key,
            &params,
            &mut cache,
        );
        assert_eq!(row[0].to_string(), "chars(first)");

        let db2 = MemDB::instance();
        db2.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table in db2");
        db2.execute("INSERT INTO t (id, value) VALUES (1, 'second')")
            .expect("seed db2");
        init_mem_provider(db2).expect("replace provider with db2");

        let row = cache_query_fields(
            "SELECT value FROM t WHERE id=:id",
            &key,
            &params,
            &mut cache,
        );
        assert_eq!(row[0].to_string(), "chars(second)");
    }

    #[test]
    fn local_cache_is_scoped_by_sql_text() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let db = MemDB::instance();
        db.execute("CREATE TABLE t1 (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create t1");
        db.execute("CREATE TABLE t2 (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create t2");
        db.execute("INSERT INTO t1 (id, value) VALUES (1, 'first')")
            .expect("seed t1");
        db.execute("INSERT INTO t2 (id, value) VALUES (1, 'second')")
            .expect("seed t2");
        init_mem_provider(db).expect("init provider");

        let key = [DataField::from_digit("id", 1)];
        let params = [DataField::from_digit(":id", 1)];
        let mut cache = FieldQueryCache::default();

        let row = cache_query_fields(
            "SELECT value FROM t1 WHERE id=:id",
            &key,
            &params,
            &mut cache,
        );
        assert_eq!(row[0].to_string(), "chars(first)");

        let row = cache_query_fields(
            "SELECT value FROM t2 WHERE id=:id",
            &key,
            &params,
            &mut cache,
        );
        assert_eq!(row[0].to_string(), "chars(second)");
    }

    #[test]
    fn runtime_snapshot_tracks_generation_and_cache_size() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let db = MemDB::instance();
        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table");
        db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
            .expect("seed table");
        init_mem_provider(db).expect("init provider");

        let mut cache = FieldQueryCache::default();
        let key = [DataField::from_digit("id", 1)];
        let params = [DataField::from_digit(":id", 1)];
        let row = cache_query_fields(
            "SELECT value FROM t WHERE id=:id",
            &key,
            &params,
            &mut cache,
        );
        assert_eq!(row[0].to_string(), "chars(first)");

        let snapshot = runtime_snapshot();
        assert!(matches!(
            snapshot.provider_kind,
            Some(ProviderKind::SqliteAuthority)
        ));
        assert!(snapshot.generation.is_some());
        assert!(snapshot.result_cache_len >= 1);
        assert!(snapshot.result_cache_capacity >= snapshot.result_cache_len);
        assert!(snapshot.metadata_cache_capacity >= snapshot.metadata_cache_len);
        assert!(snapshot.reload_successes >= 1);
    }

    #[test]
    fn metadata_cache_is_scoped_by_generation() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let sql = "SELECT value FROM t WHERE id = 1";
        let before = runtime_snapshot().metadata_cache_len;

        let db1 = MemDB::instance();
        db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table in db1");
        db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
            .expect("seed db1");
        init_mem_provider(db1).expect("init provider db1");
        let row = query_row(sql).expect("query db1");
        assert_eq!(row[0].to_string(), "chars(first)");
        let after_first = runtime_snapshot().metadata_cache_len;
        assert!(
            after_first > before,
            "metadata cache did not record first generation entry: before={before} after_first={after_first}"
        );

        let db2 = MemDB::instance();
        db2.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table in db2");
        db2.execute("INSERT INTO t (id, value) VALUES (1, 'second')")
            .expect("seed db2");
        init_mem_provider(db2).expect("replace provider with db2");
        let row = query_row(sql).expect("query db2");
        assert_eq!(row[0].to_string(), "chars(second)");
        let after_second = runtime_snapshot().metadata_cache_len;
        assert!(
            after_second > after_first,
            "metadata cache did not keep a distinct generation entry: after_first={after_first} after_second={after_second}"
        );
    }

    #[test]
    fn failed_provider_reload_keeps_previous_provider() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let db1 = MemDB::instance();
        db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table in db1");
        db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
            .expect("seed db1");
        init_mem_provider(db1).expect("init provider db1");
        let before_generation = current_generation();

        let reload_err = install_provider(
            ProviderKind::SqliteAuthority,
            datasource_id_for(ProviderKind::SqliteAuthority, "reload-failure"),
            |_generation| {
                Err(KnowReason::from_logic()
                    .to_err()
                    .with_detail("expected reload failure"))
            },
        );
        assert!(reload_err.is_err());

        let row = query_row("SELECT value FROM t WHERE id = 1").expect("query previous provider");
        assert_eq!(row[0].to_string(), "chars(first)");
        assert_eq!(current_generation(), before_generation);
    }

    #[test]
    fn runtime_snapshot_records_cache_counters() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let db = MemDB::instance();
        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table");
        db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
            .expect("seed table");
        init_mem_provider(db).expect("init provider");

        let before = runtime_snapshot();
        let mut cache = FieldQueryCache::default();
        let key = [DataField::from_digit("id", 1)];
        let params = [DataField::from_digit(":id", 1)];
        let row = cache_query_fields(
            "SELECT value FROM t WHERE id=:id",
            &key,
            &params,
            &mut cache,
        );
        assert_eq!(row[0].to_string(), "chars(first)");
        let row = cache_query_fields(
            "SELECT value FROM t WHERE id=:id",
            &key,
            &params,
            &mut cache,
        );
        assert_eq!(row[0].to_string(), "chars(first)");

        let after = runtime_snapshot();
        assert!(after.local_cache_hits > before.local_cache_hits);
        assert!(after.local_cache_misses > before.local_cache_misses);
        assert!(after.result_cache_misses > before.result_cache_misses);
        assert!(after.metadata_cache_misses > before.metadata_cache_misses);
    }

    #[test]
    fn telemetry_receives_reload_cache_and_query_events() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let telemetry_impl = Arc::new(TestTelemetry::default());
        let previous = install_runtime_telemetry(telemetry_impl.clone());

        let db = MemDB::instance();
        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table");
        db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
            .expect("seed table");
        init_mem_provider(db).expect("init provider");

        let mut cache = FieldQueryCache::default();
        let key = [DataField::from_digit("id", 1)];
        let params = [DataField::from_digit(":id", 1)];
        let row = cache_query_fields(
            "SELECT value FROM t WHERE id=:id",
            &key,
            &params,
            &mut cache,
        );
        assert_eq!(row[0].to_string(), "chars(first)");
        let row = cache_query_fields(
            "SELECT value FROM t WHERE id=:id",
            &key,
            &params,
            &mut cache,
        );
        assert_eq!(row[0].to_string(), "chars(first)");

        let reload_err = install_provider(
            ProviderKind::SqliteAuthority,
            datasource_id_for(ProviderKind::SqliteAuthority, "telemetry-failure"),
            |_generation| {
                Err(KnowReason::from_logic()
                    .to_err()
                    .with_detail("expected telemetry reload failure"))
            },
        );
        assert!(reload_err.is_err());

        install_runtime_telemetry(previous);
        reset_telemetry();

        assert!(telemetry_impl.reload_success.load(Ordering::Relaxed) >= 1);
        assert!(telemetry_impl.reload_failure.load(Ordering::Relaxed) >= 1);
        assert!(telemetry_impl.local_hits.load(Ordering::Relaxed) >= 1);
        assert!(telemetry_impl.local_misses.load(Ordering::Relaxed) >= 1);
        assert!(telemetry_impl.result_misses.load(Ordering::Relaxed) >= 1);
        assert!(telemetry_impl.metadata_misses.load(Ordering::Relaxed) >= 1);
        assert!(telemetry_impl.query_success.load(Ordering::Relaxed) >= 1);
    }

    #[test]
    #[ignore = "manual perf comparison; run with cargo test cache_perf_reports_cache_vs_no_cache -- --ignored --nocapture"]
    fn cache_perf_reports_cache_vs_no_cache() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let rows = perf_env_usize("WP_KDB_PERF_ROWS", 10_000).max(1);
        let ops = perf_env_usize("WP_KDB_PERF_OPS", 120_000).max(1);
        let hotset = perf_env_usize("WP_KDB_PERF_HOTSET", 128).clamp(1, rows);
        let workload = build_perf_workload(ops, hotset);

        eprintln!(
            "[wp-knowledge][cache-perf] rows={} ops={} hotset={} sql=SELECT value FROM perf_kv WHERE id=:id",
            rows, ops, hotset
        );

        let bypass = run_bypass_perf(&workload, rows);
        let global = run_global_cache_perf(&workload, rows);
        let local = run_local_cache_perf(&workload, rows);

        print_perf_result(&bypass);
        print_perf_result(&global);
        print_perf_result(&local);

        eprintln!(
            "[wp-knowledge][cache-perf] speedup global_vs_bypass={:.2}x local_vs_bypass={:.2}x",
            bypass.elapsed.as_secs_f64() / global.elapsed.as_secs_f64(),
            bypass.elapsed.as_secs_f64() / local.elapsed.as_secs_f64(),
        );

        assert_eq!(bypass.counters.result_hits, 0);
        assert_eq!(bypass.counters.result_misses, 0);
        assert_eq!(bypass.counters.local_hits, 0);
        assert_eq!(bypass.counters.local_misses, 0);
        assert!(global.counters.result_hits > 0);
        assert!(global.counters.result_misses > 0);
        assert_eq!(global.counters.local_hits, 0);
        assert_eq!(global.counters.local_misses, 0);
        assert!(local.counters.local_hits > 0);
        assert!(local.counters.local_misses > 0);
        assert!(local.counters.result_misses > 0);
    }

    #[test]
    fn init_thread_cloned_from_knowdb_applies_result_cache_config() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let root = uniq_cache_cfg_tmp_dir();
        let conf_path = write_minimal_knowdb_with_cache(&root, true, 7, 5);
        let auth_file = root.join(".run").join("authority.sqlite");
        fs::create_dir_all(auth_file.parent().expect("authority parent"))
            .expect("create authority parent");
        let authority_uri = format!("file:{}?mode=rwc&uri=true", auth_file.display());

        init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default())
            .expect("init knowdb with cache config");

        let snapshot = runtime_snapshot();
        assert!(snapshot.result_cache_enabled);
        assert_eq!(snapshot.result_cache_capacity, 7);
        assert_eq!(snapshot.result_cache_ttl_ms, 5);

        let req = QueryRequest::first_row(
            "SELECT value FROM example WHERE id=:id",
            fields_to_params(&[DataField::from_digit(":id", 1)]),
            CachePolicy::UseGlobal,
        );
        let before = runtime_snapshot();
        let row = runtime()
            .execute(&req)
            .expect("first result-cache query")
            .into_row();
        assert_eq!(row[0].to_string(), "chars(alpha)");
        let row = runtime()
            .execute(&req)
            .expect("second result-cache query")
            .into_row();
        assert_eq!(row[0].to_string(), "chars(alpha)");
        std::thread::sleep(Duration::from_millis(12));
        let row = runtime()
            .execute(&req)
            .expect("expired result-cache query")
            .into_row();
        assert_eq!(row[0].to_string(), "chars(alpha)");
        let after = runtime_snapshot();

        assert!(after.result_cache_hits > before.result_cache_hits);
        assert!(after.result_cache_misses >= before.result_cache_misses + 2);

        restore_default_result_cache_config();
        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn disabled_result_cache_from_knowdb_config_forces_bypass() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let root = uniq_cache_cfg_tmp_dir();
        let conf_path = write_minimal_knowdb_with_cache(&root, false, 3, 30_000);
        let auth_file = root.join(".run").join("authority.sqlite");
        fs::create_dir_all(auth_file.parent().expect("authority parent"))
            .expect("create authority parent");
        let authority_uri = format!("file:{}?mode=rwc&uri=true", auth_file.display());

        init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default())
            .expect("init knowdb with cache disabled");

        let snapshot = runtime_snapshot();
        assert!(!snapshot.result_cache_enabled);
        assert_eq!(snapshot.result_cache_capacity, 3);
        assert_eq!(snapshot.result_cache_ttl_ms, 30_000);

        let req = QueryRequest::first_row(
            "SELECT value FROM example WHERE id=:id",
            fields_to_params(&[DataField::from_digit(":id", 1)]),
            CachePolicy::UseGlobal,
        );
        let before = runtime_snapshot();
        let _ = runtime()
            .execute(&req)
            .expect("first bypassed result-cache query");
        let _ = runtime()
            .execute(&req)
            .expect("second bypassed result-cache query");
        let after = runtime_snapshot();

        assert_eq!(after.result_cache_hits, before.result_cache_hits);
        assert_eq!(after.result_cache_misses, before.result_cache_misses);

        restore_default_result_cache_config();
        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn failed_knowdb_provider_init_does_not_apply_cache_config() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        restore_default_result_cache_config();

        let db = MemDB::instance();
        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table");
        db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
            .expect("seed table");
        init_mem_provider(db).expect("init provider");

        let before = runtime_snapshot();
        let root = uniq_cache_cfg_tmp_dir();
        let conf_path = write_provider_only_knowdb_with_cache(
            &root,
            "mysql",
            "not-a-valid-mysql-url",
            false,
            3,
            5,
        );
        let authority_uri = format!(
            "file:{}?mode=rwc&uri=true",
            root.join("unused.sqlite").display()
        );

        let err =
            init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default());
        assert!(err.is_err());

        let after = runtime_snapshot();
        assert_eq!(after.result_cache_enabled, before.result_cache_enabled);
        assert_eq!(after.result_cache_capacity, before.result_cache_capacity);
        assert_eq!(after.result_cache_ttl_ms, before.result_cache_ttl_ms);

        let row = query_row("SELECT value FROM t WHERE id = 1").expect("query previous provider");
        assert_eq!(row[0].to_string(), "chars(first)");

        let _ = fs::remove_dir_all(&root);
    }

    // 安装一个命名内存 provider（不替换默认 provider）。
    fn install_named_mem_provider(name: &str, db: MemDB) {
        let datasource_id = DatasourceId::from_seed(ProviderKind::SqliteAuthority, name);
        let ds_for_scope = datasource_id.clone();
        runtime()
            .install_provider_named(
                name,
                ProviderKind::SqliteAuthority,
                datasource_id,
                |generation| {
                    Ok(Arc::new(MemProvider {
                        db,
                        metadata_scope: MetadataCacheScope {
                            datasource_id: ds_for_scope.clone(),
                            generation,
                        },
                    }))
                },
                false,
            )
            .expect("install named mem provider");
    }

    #[test]
    fn cache_query_fields_route_separates_named_and_default_providers() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");

        // 默认 provider：route_t.value = 'default'
        let db_default = MemDB::instance();
        db_default
            .execute("CREATE TABLE route_t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create default table");
        db_default
            .execute("INSERT INTO route_t (id, value) VALUES (1, 'default')")
            .expect("seed default");
        init_mem_provider(db_default).expect("init default provider");

        // 命名 provider "geo"：route_t.value = 'geo'
        let db_geo = MemDB::instance();
        db_geo
            .execute("CREATE TABLE route_t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create geo table");
        db_geo
            .execute("INSERT INTO route_t (id, value) VALUES (1, 'geo')")
            .expect("seed geo");
        install_named_mem_provider("geo", db_geo);

        let sql = "SELECT value FROM route_t WHERE id=:id";
        let c_params = [DataField::from_digit("id", 1)];
        let query_params = [DataField::from_digit(":id", 1)];
        let mut cache = FieldQueryCache::default();

        let named =
            cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
        assert_eq!(named[0].to_string(), "chars(geo)");

        // 同一 cache 实例切到默认 provider，本地缓存必须按 provider 隔离
        let default = cache_query_fields_route(None, sql, 0, &c_params, &query_params, &mut cache);
        assert_eq!(default[0].to_string(), "chars(default)");

        // 再查命名 provider，仍返回 geo
        let named_again =
            cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
        assert_eq!(named_again[0].to_string(), "chars(geo)");
    }

    #[test]
    fn mixed_provider_queries_keep_local_cache_per_provider() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");

        let db_default = MemDB::instance();
        db_default
            .execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create default table");
        db_default
            .execute("INSERT INTO t (id, value) VALUES (1, 'default')")
            .expect("seed default");
        init_mem_provider(db_default).expect("init default provider");

        let db_geo = MemDB::instance();
        db_geo
            .execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create geo table");
        db_geo
            .execute("INSERT INTO t (id, value) VALUES (1, 'geo')")
            .expect("seed geo");
        install_named_mem_provider("geo", db_geo);

        let sql = "SELECT value FROM t WHERE id=:id";
        let c_params = [DataField::from_digit("id", 1)];
        let query_params = [DataField::from_digit(":id", 1)];
        let mut cache = FieldQueryCache::default();

        let before = runtime_snapshot();
        // 1) 命名 provider：miss → 执行
        let row =
            cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
        assert_eq!(row[0].to_string(), "chars(geo)");
        // 2) 默认 provider：scope 不同，不应清空 geo 的本地缓存
        let row = cache_query_fields_route(None, sql, 0, &c_params, &query_params, &mut cache);
        assert_eq!(row[0].to_string(), "chars(default)");
        // 3) 再查命名 provider：应命中本地缓存（不重新执行）
        let row =
            cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
        assert_eq!(row[0].to_string(), "chars(geo)");
        let after = runtime_snapshot();
        assert!(
            after.local_cache_hits > before.local_cache_hits,
            "命名 provider 二次查询应命中本地缓存（按 provider 隔离代际）"
        );
    }

    #[test]
    fn local_authority_init_prunes_stale_named_providers() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");

        // 先安装命名 provider，模拟上一份配置的残留
        let db = MemDB::instance();
        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create table");
        db.execute("INSERT INTO t (id, value) VALUES (1, 'x')")
            .expect("seed table");
        install_named_mem_provider("geo", db);
        assert!(provider_exists("geo"));

        // 再用本地 authority 配置初始化 → 应清理 geo，仅保留 default
        let root = uniq_cache_cfg_tmp_dir();
        let conf_path = write_minimal_knowdb_with_cache(&root, true, 16, 30_000);
        let auth_file = root.join(".run").join("authority.sqlite");
        fs::create_dir_all(auth_file.parent().expect("authority parent"))
            .expect("create authority parent");
        let authority_uri = format!("file:{}?mode=rwc&uri=true", auth_file.display());

        init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default())
            .expect("init local authority");

        assert!(!provider_exists("geo"));
        assert!(provider_exists(DEFAULT_SQLDB_NAME));

        restore_default_result_cache_config();
        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn cache_query_fields_route_unknown_provider_returns_empty() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let c_params = [DataField::from_digit("id", 1)];
        let query_params = [DataField::from_digit(":id", 1)];
        let mut cache = FieldQueryCache::default();
        let out = cache_query_fields_route(
            Some("nope"),
            "SELECT value FROM t WHERE id=:id",
            0,
            &c_params,
            &query_params,
            &mut cache,
        );
        assert!(out.is_empty());
    }

    #[test]
    fn query_for_named_provider_returns_rows() {
        let _guard = crate::runtime::runtime_test_guard()
            .lock()
            .expect("provider test guard");
        let db = MemDB::instance();
        db.execute("CREATE TABLE named_t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create named table");
        db.execute("INSERT INTO named_t (id, value) VALUES (1, 'named')")
            .expect("seed named");
        install_named_mem_provider("geo", db);

        let rows = query_for("geo", "SELECT value FROM named_t WHERE id = 1").expect("query geo");
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0][0].to_string(), "chars(named)");
        assert!(provider_exists("geo"));
        assert!(!provider_exists("nope"));
        assert!(provider_names().iter().any(|name| name == "geo"));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn query_async_for_named_provider_works() {
        let _guard = crate::runtime::runtime_test_guard().lock_async().await;
        let db = MemDB::instance();
        db.execute("CREATE TABLE named_t (id INTEGER PRIMARY KEY, value TEXT)")
            .expect("create named table");
        db.execute("INSERT INTO named_t (id, value) VALUES (1, 'async-named')")
            .expect("seed named");
        install_named_mem_provider("geo", db);

        let rows = query_async_for("geo", "SELECT value FROM named_t WHERE id = 1")
            .await
            .expect("query geo");
        assert_eq!(rows[0][0].to_string(), "chars(async-named)");
    }
}