Skip to main content

wp_knowledge/
facade.rs

1use std::path::Path;
2use std::sync::Arc;
3use std::time::Duration;
4use std::{
5    collections::hash_map::DefaultHasher,
6    collections::{HashMap, HashSet},
7    hash::{Hash, Hasher},
8};
9
10use crate::error::KnowledgeResult;
11use async_trait::async_trait;
12use rusqlite::ToSql;
13use rusqlite::{Connection, OpenFlags};
14use wp_log::{info_ctrl, warn_kdb};
15use wp_model_core::model::DataField;
16
17use crate::cache::CacheAble;
18use crate::loader::{
19    DEFAULT_SQLDB_NAME, ProviderKind, SqlProviderKind, SqlProviderSpec, parse_knowdb_conf,
20};
21use crate::mem::RowData;
22use crate::mem::memdb::MemDB;
23use crate::mem::thread_clone::ThreadClonedMDB;
24use crate::mysql::{MySqlProvider, MySqlProviderConfig};
25use crate::param::named_params_to_fields;
26use crate::postgres::{PostgresProvider, PostgresProviderConfig};
27use crate::runtime::{
28    CachePolicy, CachedRedisValue, DatasourceId, Generation, MetadataCacheScope, ProviderExecutor,
29    QueryRequest, QueryResponse, RedisCacheKey, RedisCmdTag, RuntimeSnapshot, runtime,
30};
31use crate::telemetry::{
32    CacheLayer, CacheOutcome, CacheTelemetryEvent, KnowledgeTelemetry, install_telemetry,
33    telemetry, telemetry_enabled,
34};
35
36struct MemProvider {
37    db: MemDB,
38    metadata_scope: MetadataCacheScope,
39}
40
41#[async_trait]
42impl ProviderExecutor for ThreadClonedMDB {
43    fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
44        ThreadClonedMDB::query_with_scope(self, sql)
45    }
46
47    fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
48        ThreadClonedMDB::query_fields_with_scope(self, sql, params)
49    }
50
51    fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
52        ThreadClonedMDB::query_row_with_scope(self, sql)
53    }
54
55    fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
56        ThreadClonedMDB::query_named_fields_with_scope(self, sql, params)
57    }
58}
59
60#[async_trait]
61impl ProviderExecutor for MemProvider {
62    fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
63        self.db.query_with_scope(&self.metadata_scope, sql)
64    }
65
66    fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
67        self.db
68            .query_fields_with_scope(&self.metadata_scope, sql, params)
69    }
70
71    fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
72        self.db.query_row_with_scope(&self.metadata_scope, sql)
73    }
74
75    fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
76        self.db
77            .query_named_fields_with_scope(&self.metadata_scope, sql, params)
78    }
79}
80
81#[async_trait]
82impl ProviderExecutor for PostgresProvider {
83    fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
84        PostgresProvider::query(self, sql)
85    }
86
87    fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
88        PostgresProvider::query_fields(self, sql, params)
89    }
90
91    fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
92        PostgresProvider::query_row(self, sql)
93    }
94
95    fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
96        PostgresProvider::query_named_fields(self, sql, params)
97    }
98
99    async fn query_async(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
100        PostgresProvider::query_async(self, sql).await
101    }
102
103    async fn query_fields_async(
104        &self,
105        sql: &str,
106        params: &[DataField],
107    ) -> KnowledgeResult<Vec<RowData>> {
108        PostgresProvider::query_fields_async(self, sql, params).await
109    }
110
111    async fn query_row_async(&self, sql: &str) -> KnowledgeResult<RowData> {
112        PostgresProvider::query_row_async(self, sql).await
113    }
114
115    async fn query_named_fields_async(
116        &self,
117        sql: &str,
118        params: &[DataField],
119    ) -> KnowledgeResult<RowData> {
120        PostgresProvider::query_named_fields_async(self, sql, params).await
121    }
122}
123
124#[async_trait]
125impl ProviderExecutor for MySqlProvider {
126    fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
127        MySqlProvider::query(self, sql)
128    }
129
130    fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
131        MySqlProvider::query_fields(self, sql, params)
132    }
133
134    fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
135        MySqlProvider::query_row(self, sql)
136    }
137
138    fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
139        MySqlProvider::query_named_fields(self, sql, params)
140    }
141
142    async fn query_async(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
143        MySqlProvider::query_async(self, sql).await
144    }
145
146    async fn query_fields_async(
147        &self,
148        sql: &str,
149        params: &[DataField],
150    ) -> KnowledgeResult<Vec<RowData>> {
151        MySqlProvider::query_fields_async(self, sql, params).await
152    }
153
154    async fn query_row_async(&self, sql: &str) -> KnowledgeResult<RowData> {
155        MySqlProvider::query_row_async(self, sql).await
156    }
157
158    async fn query_named_fields_async(
159        &self,
160        sql: &str,
161        params: &[DataField],
162    ) -> KnowledgeResult<RowData> {
163        MySqlProvider::query_named_fields_async(self, sql, params).await
164    }
165}
166
167fn install_provider<F>(
168    kind: ProviderKind,
169    datasource_id: DatasourceId,
170    build: F,
171) -> KnowledgeResult<()>
172where
173    F: FnOnce(Generation) -> KnowledgeResult<Arc<dyn ProviderExecutor>>,
174{
175    runtime().install_provider(kind, datasource_id, build)?;
176    Ok(())
177}
178
179fn datasource_id_for(kind: ProviderKind, seed: &str) -> DatasourceId {
180    DatasourceId::from_seed(kind, seed)
181}
182
183pub fn init_thread_cloned_from_authority(authority_uri: &str) -> KnowledgeResult<()> {
184    let datasource_id = datasource_id_for(ProviderKind::SqliteAuthority, authority_uri);
185    install_provider(ProviderKind::SqliteAuthority, datasource_id, |generation| {
186        Ok(Arc::new(ThreadClonedMDB::from_authority_with_scope(
187            authority_uri,
188            datasource_id_for(ProviderKind::SqliteAuthority, authority_uri),
189            generation.0,
190        )))
191    })
192}
193
194pub fn init_mem_provider(memdb: MemDB) -> KnowledgeResult<()> {
195    install_provider(
196        ProviderKind::SqliteAuthority,
197        datasource_id_for(ProviderKind::SqliteAuthority, "memdb"),
198        |generation| {
199            Ok(Arc::new(MemProvider {
200                db: memdb,
201                metadata_scope: MetadataCacheScope {
202                    datasource_id: datasource_id_for(ProviderKind::SqliteAuthority, "memdb"),
203                    generation,
204                },
205            }))
206        },
207    )
208}
209
210pub fn init_postgres_provider(connection_uri: &str, pool_size: Option<u32>) -> KnowledgeResult<()> {
211    init_postgres_provider_with_config(
212        PostgresProviderConfig::new(connection_uri).with_pool_size(pool_size),
213    )
214}
215
216pub fn init_postgres_provider_with_config(config: PostgresProviderConfig) -> KnowledgeResult<()> {
217    let connection_uri = config.connection_uri().to_string();
218    install_postgres_provider(DEFAULT_SQLDB_NAME, config, &connection_uri, true)
219}
220
221/// 以命名 provider 安装 PostgreSQL 连接(不替换默认 provider)。
222pub fn init_postgres_provider_named(
223    name: &str,
224    config: PostgresProviderConfig,
225) -> KnowledgeResult<()> {
226    install_postgres_provider(name, config, name, false)
227}
228
229/// 以命名 provider 安装 PostgreSQL 连接(URI 便捷版,供引擎等外部调用方使用——
230/// `PostgresProviderConfig` 类型不经 pub 面导出)。
231pub fn init_postgres_provider_named_uri(
232    name: &str,
233    connection_uri: &str,
234    pool_size: Option<u32>,
235) -> KnowledgeResult<()> {
236    init_postgres_provider_named(
237        name,
238        PostgresProviderConfig::new(connection_uri).with_pool_size(pool_size),
239    )
240}
241
242fn install_postgres_provider(
243    name: &str,
244    config: PostgresProviderConfig,
245    datasource_seed: &str,
246    is_default: bool,
247) -> KnowledgeResult<()> {
248    let datasource_id = datasource_id_for(ProviderKind::Postgres, datasource_seed);
249    runtime().install_provider_named(
250        name,
251        ProviderKind::Postgres,
252        datasource_id.clone(),
253        |generation| {
254            let provider = PostgresProvider::connect(
255                name,
256                &config,
257                MetadataCacheScope {
258                    datasource_id: datasource_id.clone(),
259                    generation,
260                },
261            )?;
262            Ok(Arc::new(provider))
263        },
264        is_default,
265    )?;
266    Ok(())
267}
268
269pub fn init_mysql_provider(connection_uri: &str, pool_size: Option<u32>) -> KnowledgeResult<()> {
270    init_mysql_provider_with_config(
271        MySqlProviderConfig::new(connection_uri).with_pool_size(pool_size),
272    )
273}
274
275pub fn init_mysql_provider_with_config(config: MySqlProviderConfig) -> KnowledgeResult<()> {
276    let connection_uri = config.connection_uri().to_string();
277    install_mysql_provider(DEFAULT_SQLDB_NAME, config, &connection_uri, true)
278}
279
280/// 以命名 provider 安装 MySQL 连接(不替换默认 provider)。
281pub fn init_mysql_provider_named(name: &str, config: MySqlProviderConfig) -> KnowledgeResult<()> {
282    install_mysql_provider(name, config, name, false)
283}
284
285fn install_mysql_provider(
286    name: &str,
287    config: MySqlProviderConfig,
288    datasource_seed: &str,
289    is_default: bool,
290) -> KnowledgeResult<()> {
291    let datasource_id = datasource_id_for(ProviderKind::Mysql, datasource_seed);
292    runtime().install_provider_named(
293        name,
294        ProviderKind::Mysql,
295        datasource_id.clone(),
296        |generation| {
297            let provider = MySqlProvider::connect(
298                &config,
299                MetadataCacheScope {
300                    datasource_id: datasource_id.clone(),
301                    generation,
302                },
303            )?;
304            Ok(Arc::new(provider))
305        },
306        is_default,
307    )?;
308    Ok(())
309}
310
311fn build_postgres_config(spec: &SqlProviderSpec) -> PostgresProviderConfig {
312    PostgresProviderConfig::new(spec.connection_uri.clone())
313        .with_pool_size(spec.pool_size)
314        .with_min_connections(spec.min_connections)
315        .with_acquire_timeout_ms(spec.acquire_timeout_ms)
316        .with_idle_timeout_ms(spec.idle_timeout_ms)
317        .with_max_lifetime_ms(spec.max_lifetime_ms)
318        .with_session(spec.postgres_session.as_ref())
319}
320
321fn build_mysql_config(spec: &SqlProviderSpec) -> MySqlProviderConfig {
322    MySqlProviderConfig::new(spec.connection_uri.clone())
323        .with_pool_size(spec.pool_size)
324        .with_min_connections(spec.min_connections)
325        .with_acquire_timeout_ms(spec.acquire_timeout_ms)
326        .with_idle_timeout_ms(spec.idle_timeout_ms)
327        .with_max_lifetime_ms(spec.max_lifetime_ms)
328}
329
330// ---------------------------------------------------------------------------
331// Redis provider API
332// ---------------------------------------------------------------------------
333
334/// Check whether an item exists in a Bloom filter.
335///
336/// Returns `true` if the item *may* exist, `false` if it definitely does not.
337/// Requires the RedisBloom module.
338pub fn redis_bf_exists(key: &str, item: &str) -> KnowledgeResult<bool> {
339    if let Some(generation) = current_generation() {
340        let ck = redis_cache_key(RedisCmdTag::BfExists, generation, key, &[item]);
341        if let Some(CachedRedisValue::Bool(v)) = runtime().redis_cache_get(&ck) {
342            return Ok(v);
343        }
344        let result = crate::redis::bf_exists("knowdb", key, item)?;
345        runtime().redis_cache_put(ck, CachedRedisValue::Bool(result));
346        return Ok(result);
347    }
348    crate::redis::bf_exists("knowdb", key, item)
349}
350
351/// Get the value of a hash field.
352///
353/// Returns `None` when the key or field does not exist.
354pub fn redis_hget(key: &str, field: &str) -> KnowledgeResult<Option<String>> {
355    if let Some(generation) = current_generation() {
356        let ck = redis_cache_key(RedisCmdTag::HGet, generation, key, &[field]);
357        if let Some(CachedRedisValue::OptString(ref v)) = runtime().redis_cache_get(&ck) {
358            return Ok(v.clone());
359        }
360        let result = crate::redis::hget("knowdb", key, field)?;
361        runtime().redis_cache_put(ck, CachedRedisValue::OptString(result.clone()));
362        return Ok(result);
363    }
364    crate::redis::hget("knowdb", key, field)
365}
366
367/// Get the value of a key.
368///
369/// Returns `None` when the key does not exist.
370pub fn redis_get(key: &str) -> KnowledgeResult<Option<String>> {
371    if let Some(generation) = current_generation() {
372        let ck = redis_cache_key(RedisCmdTag::Get, generation, key, &[]);
373        if let Some(CachedRedisValue::OptString(ref v)) = runtime().redis_cache_get(&ck) {
374            return Ok(v.clone());
375        }
376        let result = crate::redis::get("knowdb", key)?;
377        runtime().redis_cache_put(ck, CachedRedisValue::OptString(result.clone()));
378        return Ok(result);
379    }
380    crate::redis::get("knowdb", key)
381}
382
383/// Check whether a member exists in a set.
384pub fn redis_set_exists(key: &str, member: &str) -> KnowledgeResult<bool> {
385    if let Some(generation) = current_generation() {
386        let ck = redis_cache_key(RedisCmdTag::SetExists, generation, key, &[member]);
387        if let Some(CachedRedisValue::Bool(v)) = runtime().redis_cache_get(&ck) {
388            return Ok(v);
389        }
390        let result = crate::redis::set_exists("knowdb", key, member)?;
391        runtime().redis_cache_put(ck, CachedRedisValue::Bool(result));
392        return Ok(result);
393    }
394    crate::redis::set_exists("knowdb", key, member)
395}
396
397/// Add items to a Bloom filter.
398///
399/// Returns a `Vec<bool>` indicating whether each item was *new* to the filter.
400/// Requires the RedisBloom module.
401pub fn redis_bf_add(key: &str, items: &[&str]) -> KnowledgeResult<Vec<bool>> {
402    let owned: Vec<String> = items.iter().map(|s| s.to_string()).collect();
403    crate::redis::bf_madd("knowdb", key, &owned)
404}
405
406/// Create a new Bloom filter.
407///
408/// `error_rate`: desired false positive rate (e.g. `0.01` for 1%).
409/// `capacity`: expected number of items to add.
410/// Requires the RedisBloom module.
411pub fn redis_bf_create(key: &str, error_rate: f64, capacity: i64) -> KnowledgeResult<()> {
412    crate::redis::bf_reserve("knowdb", key, error_rate, capacity)
413}
414
415// ---------------------------------------------------------------------------
416// External call API — named queries from [fun.<name>]
417// ---------------------------------------------------------------------------
418
419/// Register [fun] definitions from knowdb.toml. Called once during init.
420pub(crate) fn register_fun_map(map: HashMap<String, crate::loader::FunSpec>) {
421    crate::fun::register_fun_map(map);
422}
423
424/// Execute a [fun.<name>] named query where returns = "bool".
425pub fn external_exists(service: &str, arg: &str) -> KnowledgeResult<bool> {
426    let spec = crate::fun::resolve_spec(service, true)?;
427    let redis_key = spec.key.as_deref().unwrap_or(arg);
428    if spec.enabled()
429        && let Some(generation) = current_generation()
430    {
431        let ck = redis_cache_key(RedisCmdTag::BfExists, generation, redis_key, &[arg]);
432        if let Some(CachedRedisValue::Bool(v)) = runtime().redis_cache_get(&ck) {
433            return Ok(v);
434        }
435        let result = crate::fun::external_exists(service, arg)?;
436        if let Some(ttl) = spec.ttl_ms {
437            runtime().redis_cache_put_with_ttl(ck, CachedRedisValue::Bool(result), ttl);
438        } else {
439            runtime().redis_cache_put(ck, CachedRedisValue::Bool(result));
440        }
441        return Ok(result);
442    }
443    crate::fun::external_exists(service, arg)
444}
445
446/// Execute a [fun.<name>] named query where returns = "value".
447pub fn external_value(service: &str, arg: &str) -> KnowledgeResult<Option<String>> {
448    let spec = crate::fun::resolve_spec(service, false)?;
449    let redis_key = spec.key.as_deref().unwrap_or(arg);
450    if spec.enabled()
451        && let Some(generation) = current_generation()
452    {
453        let ck = redis_cache_key(RedisCmdTag::Get, generation, redis_key, &[arg]);
454        if let Some(CachedRedisValue::OptString(ref v)) = runtime().redis_cache_get(&ck) {
455            return Ok(v.clone());
456        }
457        let result = crate::fun::external_value(service, arg)?;
458        if let Some(ttl) = spec.ttl_ms {
459            runtime().redis_cache_put_with_ttl(
460                ck,
461                CachedRedisValue::OptString(result.clone()),
462                ttl,
463            );
464        } else {
465            runtime().redis_cache_put(ck, CachedRedisValue::OptString(result.clone()));
466        }
467        return Ok(result);
468    }
469    crate::fun::external_value(service, arg)
470}
471
472#[allow(dead_code)]
473pub(crate) fn redis_ping() -> KnowledgeResult<bool> {
474    crate::redis::ping_blocking("knowdb")
475}
476
477#[allow(dead_code)]
478pub(crate) fn redis_close() -> KnowledgeResult<()> {
479    crate::redis::close(Some("knowdb"))
480}
481
482fn redis_cache_key(cmd: RedisCmdTag, generation: u64, key: &str, args: &[&str]) -> RedisCacheKey {
483    use std::collections::hash_map::DefaultHasher;
484    use std::hash::{Hash, Hasher};
485    let mut hasher = DefaultHasher::new();
486    key.hash(&mut hasher);
487    let key_hash = hasher.finish();
488    let mut hasher = DefaultHasher::new();
489    for arg in args {
490        arg.hash(&mut hasher);
491    }
492    let args_hash = hasher.finish();
493    RedisCacheKey {
494        generation,
495        cmd_tag: cmd,
496        key_hash,
497        args_hash,
498    }
499}
500
501// ---------------------------------------------------------------------------
502
503pub fn current_generation() -> Option<u64> {
504    runtime()
505        .current_generation()
506        .map(|generation| generation.0)
507}
508
509pub fn runtime_snapshot() -> RuntimeSnapshot {
510    runtime().snapshot()
511}
512
513pub fn install_runtime_telemetry(
514    telemetry_impl: Arc<dyn KnowledgeTelemetry>,
515) -> Arc<dyn KnowledgeTelemetry> {
516    install_telemetry(telemetry_impl)
517}
518
519pub use crate::sql_route::route_provider_sql;
520
521/// 指定名称的 provider 是否已安装。
522pub fn provider_exists(name: &str) -> bool {
523    runtime().provider_exists(name)
524}
525
526/// 当前已安装的命名 provider 名称列表。
527pub fn provider_names() -> Vec<String> {
528    runtime().provider_names()
529}
530
531/// 对指定命名 provider 执行查询(全部行)。
532pub fn query_for(name: &str, sql: &str) -> KnowledgeResult<Vec<RowData>> {
533    runtime()
534        .execute_for(
535            name,
536            &QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass),
537        )
538        .map(QueryResponse::into_rows)
539}
540
541/// 对指定命名 provider 异步执行查询(全部行)。
542pub async fn query_async_for(name: &str, sql: &str) -> KnowledgeResult<Vec<RowData>> {
543    runtime()
544        .execute_for_async(
545            name,
546            &QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass),
547        )
548        .await
549        .map(QueryResponse::into_rows)
550}
551
552/// 对指定命名 provider 执行首行查询。
553pub fn query_fields_for(name: &str, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
554    runtime().execute_first_row_fields_for(name, sql, params, CachePolicy::Bypass)
555}
556
557/// 对指定命名 provider 异步执行首行查询。
558pub async fn query_fields_for_async(
559    name: &str,
560    sql: &str,
561    params: &[DataField],
562) -> KnowledgeResult<RowData> {
563    runtime()
564        .execute_first_row_fields_for_async(name, sql, params, CachePolicy::Bypass)
565        .await
566}
567
568pub fn query(sql: &str) -> KnowledgeResult<Vec<RowData>> {
569    runtime()
570        .execute(&QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass))
571        .map(QueryResponse::into_rows)
572}
573
574pub async fn query_async(sql: &str) -> KnowledgeResult<Vec<RowData>> {
575    runtime()
576        .execute_async(&QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass))
577        .await
578        .map(QueryResponse::into_rows)
579}
580
581pub fn query_row(sql: &str) -> KnowledgeResult<RowData> {
582    runtime()
583        .execute(&QueryRequest::first_row(
584            sql,
585            Vec::new(),
586            CachePolicy::Bypass,
587        ))
588        .map(QueryResponse::into_row)
589}
590
591pub async fn query_row_async(sql: &str) -> KnowledgeResult<RowData> {
592    runtime()
593        .execute_async(&QueryRequest::first_row(
594            sql,
595            Vec::new(),
596            CachePolicy::Bypass,
597        ))
598        .await
599        .map(QueryResponse::into_row)
600}
601
602pub fn query_fields(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
603    runtime().execute_first_row_fields(sql, params, CachePolicy::Bypass)
604}
605
606pub async fn query_fields_async(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
607    runtime()
608        .execute_first_row_fields_async(sql, params, CachePolicy::Bypass)
609        .await
610}
611
612pub fn query_named_fields(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
613    query_fields(sql, params)
614}
615
616pub async fn query_named_fields_async(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
617    query_fields_async(sql, params).await
618}
619
620pub fn query_named<'a>(
621    sql: &str,
622    params: &'a [(&'a str, &'a dyn ToSql)],
623) -> KnowledgeResult<RowData> {
624    let fields = named_params_to_fields(params)?;
625    query_fields(sql, &fields)
626}
627
628pub fn cache_query_fields<const N: usize>(
629    sql: &str,
630    c_params: &[DataField; N],
631    query_params: &[DataField],
632    cache: &mut impl CacheAble<DataField, RowData, N>,
633) -> RowData {
634    cache_query_fields_with_scope(sql, stable_hash(sql), c_params, query_params, cache)
635}
636
637pub fn cache_query_fields_with_scope<const N: usize>(
638    sql: &str,
639    local_cache_scope: u64,
640    c_params: &[DataField; N],
641    query_params: &[DataField],
642    cache: &mut impl CacheAble<DataField, RowData, N>,
643) -> RowData {
644    let generation = current_generation();
645    let provider_kind = runtime().current_provider_kind();
646    if let Some(hit) = local_cache_hit(
647        local_cache_scope,
648        c_params,
649        cache,
650        generation,
651        provider_kind,
652    ) {
653        return hit;
654    }
655    match runtime().execute_first_row_fields(sql, query_params, CachePolicy::UseGlobal) {
656        Ok(row) => {
657            cache.save_scoped(local_cache_scope, c_params, row.clone());
658            row
659        }
660        Err(err) => {
661            warn_kdb!("[kdb] query error: {}", err);
662            Vec::new()
663        }
664    }
665}
666
667/// 本地缓存路由派发:`provider` 为 `Some(name)` 时走命名 provider,`None` 时走默认 provider。
668pub fn cache_query_fields_route<const N: usize>(
669    provider: Option<&str>,
670    sql: &str,
671    local_cache_scope: u64,
672    c_params: &[DataField; N],
673    query_params: &[DataField],
674    cache: &mut impl CacheAble<DataField, RowData, N>,
675) -> RowData {
676    let Some(name) = provider else {
677        return cache_query_fields_with_scope(
678            sql,
679            local_cache_scope,
680            c_params,
681            query_params,
682            cache,
683        );
684    };
685    let handle = match runtime().provider_by_name(name) {
686        Ok(handle) => handle,
687        Err(err) => {
688            warn_kdb!("[kdb] named provider '{name}' unavailable: {}", err);
689            return Vec::new();
690        }
691    };
692    let generation = Some(handle.generation.0);
693    let provider_kind = Some(handle.kind.clone());
694    // provider 粒度 scope:与默认 provider 区分,使本地缓存代际按 provider 独立跟踪,
695    // 混用命名库与默认库时不互相重置。
696    let scope = local_cache_scope ^ stable_hash(name);
697    if let Some(hit) = local_cache_hit(scope, c_params, cache, generation, provider_kind) {
698        return hit;
699    }
700    match runtime().execute_first_row_fields_for(name, sql, query_params, CachePolicy::UseGlobal) {
701        Ok(row) => {
702            cache.save_scoped(scope, c_params, row.clone());
703            row
704        }
705        Err(err) => {
706            warn_kdb!("[kdb] query error: {}", err);
707            Vec::new()
708        }
709    }
710}
711
712/// 本地缓存命中检查(含 generation 准备)与 miss 记录。命中时返回 `Some(hit)`。
713fn local_cache_hit<const N: usize>(
714    local_cache_scope: u64,
715    c_params: &[DataField; N],
716    cache: &mut impl CacheAble<DataField, RowData, N>,
717    generation: Option<u64>,
718    provider_kind: Option<ProviderKind>,
719) -> Option<RowData> {
720    if let Some(generation) = generation {
721        cache.prepare_generation(local_cache_scope, generation);
722    }
723    if let Some(hit) = cache.fetch_scoped(local_cache_scope, c_params) {
724        runtime().record_local_cache_hit();
725        if telemetry_enabled() {
726            telemetry().on_cache(&CacheTelemetryEvent {
727                layer: CacheLayer::Local,
728                outcome: CacheOutcome::Hit,
729                provider_kind,
730            });
731        }
732        return Some(hit.clone());
733    }
734    runtime().record_local_cache_miss();
735    if telemetry_enabled() {
736        telemetry().on_cache(&CacheTelemetryEvent {
737            layer: CacheLayer::Local,
738            outcome: CacheOutcome::Miss,
739            provider_kind,
740        });
741    }
742    None
743}
744
745pub async fn cache_query_fields_async<const N: usize>(
746    sql: &str,
747    c_params: &[DataField; N],
748    query_params: &[DataField],
749    cache: &mut impl CacheAble<DataField, RowData, N>,
750) -> RowData {
751    cache_query_fields_async_with_scope(
752        sql,
753        stable_hash(sql),
754        c_params,
755        || query_params.to_vec(),
756        cache,
757    )
758    .await
759}
760
761pub async fn cache_query_fields_async_with<const N: usize, F>(
762    sql: &str,
763    c_params: &[DataField; N],
764    build_query_params: F,
765    cache: &mut impl CacheAble<DataField, RowData, N>,
766) -> RowData
767where
768    F: FnOnce() -> Vec<DataField>,
769{
770    cache_query_fields_async_with_scope(sql, stable_hash(sql), c_params, build_query_params, cache)
771        .await
772}
773
774pub async fn cache_query_fields_async_with_scope<const N: usize, F>(
775    sql: &str,
776    local_cache_scope: u64,
777    c_params: &[DataField; N],
778    build_query_params: F,
779    cache: &mut impl CacheAble<DataField, RowData, N>,
780) -> RowData
781where
782    F: FnOnce() -> Vec<DataField>,
783{
784    let generation = current_generation();
785    let provider_kind = runtime().current_provider_kind();
786    if let Some(hit) = local_cache_hit(
787        local_cache_scope,
788        c_params,
789        cache,
790        generation,
791        provider_kind,
792    ) {
793        return hit;
794    }
795    let query_params = build_query_params();
796    match runtime()
797        .execute_first_row_fields_async(sql, &query_params, CachePolicy::UseGlobal)
798        .await
799    {
800        Ok(row) => {
801            cache.save_scoped(local_cache_scope, c_params, row.clone());
802            row
803        }
804        Err(err) => {
805            warn_kdb!("[kdb] query error: {}", err);
806            Vec::new()
807        }
808    }
809}
810
811/// 异步本地缓存路由派发:`provider` 为 `Some(name)` 时走命名 provider,`None` 时走默认 provider。
812pub async fn cache_query_fields_route_async<const N: usize, F>(
813    provider: Option<&str>,
814    sql: &str,
815    local_cache_scope: u64,
816    c_params: &[DataField; N],
817    build_query_params: F,
818    cache: &mut impl CacheAble<DataField, RowData, N>,
819) -> RowData
820where
821    F: FnOnce() -> Vec<DataField>,
822{
823    let Some(name) = provider else {
824        return cache_query_fields_async_with_scope(
825            sql,
826            local_cache_scope,
827            c_params,
828            build_query_params,
829            cache,
830        )
831        .await;
832    };
833    let handle = match runtime().provider_by_name(name) {
834        Ok(handle) => handle,
835        Err(err) => {
836            warn_kdb!("[kdb] named provider '{name}' unavailable: {}", err);
837            return Vec::new();
838        }
839    };
840    let generation = Some(handle.generation.0);
841    let provider_kind = Some(handle.kind.clone());
842    // provider 粒度 scope:与默认 provider 区分,本地缓存代际按 provider 独立跟踪。
843    let scope = local_cache_scope ^ stable_hash(name);
844    if let Some(hit) = local_cache_hit(scope, c_params, cache, generation, provider_kind) {
845        return hit;
846    }
847    let query_params = build_query_params();
848    match runtime()
849        .execute_first_row_fields_for_async(name, sql, &query_params, CachePolicy::UseGlobal)
850        .await
851    {
852        Ok(row) => {
853            cache.save_scoped(scope, c_params, row.clone());
854            row
855        }
856        Err(err) => {
857            warn_kdb!("[kdb] query error: {}", err);
858            Vec::new()
859        }
860    }
861}
862
863pub fn cache_query<const N: usize>(
864    sql: &str,
865    c_params: &[DataField; N],
866    named_params: &[(&str, &dyn ToSql)],
867    cache: &mut impl CacheAble<DataField, RowData, N>,
868) -> RowData {
869    let query_fields = match named_params_to_fields(named_params) {
870        Ok(fields) => fields,
871        Err(err) => {
872            warn_kdb!("[kdb] query param conversion error: {}", err);
873            return Vec::new();
874        }
875    };
876
877    cache_query_fields(sql, c_params, &query_fields, cache)
878}
879
880pub async fn cache_query_async<const N: usize>(
881    sql: &str,
882    c_params: &[DataField; N],
883    named_params: &[(&str, &dyn ToSql)],
884    cache: &mut impl CacheAble<DataField, RowData, N>,
885) -> RowData {
886    let query_fields = match named_params_to_fields(named_params) {
887        Ok(fields) => fields,
888        Err(err) => {
889            warn_kdb!("[kdb] query param conversion error: {}", err);
890            return Vec::new();
891        }
892    };
893
894    cache_query_fields_async(sql, c_params, &query_fields, cache).await
895}
896
897fn ensure_wal(authority_uri: &str) -> KnowledgeResult<()> {
898    if let Ok(conn) = Connection::open_with_flags(
899        authority_uri,
900        OpenFlags::SQLITE_OPEN_READ_WRITE
901            | OpenFlags::SQLITE_OPEN_CREATE
902            | OpenFlags::SQLITE_OPEN_URI,
903    ) {
904        let _ = conn.execute_batch(
905            "PRAGMA journal_mode=WAL;\nPRAGMA synchronous=NORMAL;\nPRAGMA temp_store=MEMORY;",
906        );
907    }
908    Ok(())
909}
910
911pub fn init_wal_pool_from_authority(authority_uri: &str, pool_size: u32) -> KnowledgeResult<()> {
912    ensure_wal(authority_uri)?;
913    let flags = OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI;
914    let mem = MemDB::new_file(authority_uri, pool_size, flags)?;
915    init_mem_provider(mem)
916}
917
918pub fn init_thread_cloned_from_knowdb(
919    root: &Path,
920    knowdb_conf: &Path,
921    authority_uri: &str,
922    dict: &orion_variate::EnvDict,
923) -> KnowledgeResult<()> {
924    let (conf, conf_abs, _) = parse_knowdb_conf(root, knowdb_conf, dict)?;
925    if let Some(provider_cfg) = conf.provider() {
926        // New-style [provider.sqldb] / [[provider.sqldb]]
927        if let Some(specs) = provider_cfg.sqldb
928            && !specs.is_empty()
929        {
930            let expected = SqlProviderSpec::validate_specs(&specs)?;
931            // 默认 provider:生效名为 `default`(历史单库)者优先,否则取第一个。
932            let default_name = if expected.contains(DEFAULT_SQLDB_NAME) {
933                DEFAULT_SQLDB_NAME.to_string()
934            } else {
935                specs[0].effective_name().to_string()
936            };
937            // 外部 SQL provider 只读(仅 SELECT),初始化中途失败时允许部分安装:
938            // 已连通的库照常可用、连不上的库不可查询,错误由调用方记录后继续。
939            // 因此不做"先全部建池再统一提交"的事务化处理。
940            for spec in &specs {
941                let name = spec.effective_name().to_string();
942                let is_default = name == default_name;
943                // 无显式 name 的历史单库保留 connection_uri 作为 datasource 种子
944                let seed = match &spec.name {
945                    Some(_) => name.clone(),
946                    None => spec.connection_uri.clone(),
947                };
948                match spec.kind {
949                    SqlProviderKind::Postgres => {
950                        info_ctrl!(
951                            "init postgres knowdb provider({name}:{}) ",
952                            conf_abs.display(),
953                        );
954                        install_postgres_provider(
955                            &name,
956                            build_postgres_config(spec),
957                            &seed,
958                            is_default,
959                        )?;
960                    }
961                    SqlProviderKind::Mysql => {
962                        info_ctrl!("init mysql knowdb provider({name}:{}) ", conf_abs.display(),);
963                        install_mysql_provider(&name, build_mysql_config(spec), &seed, is_default)?;
964                    }
965                }
966            }
967            runtime().configure_result_cache(
968                conf.cache.enabled,
969                conf.cache.capacity,
970                Duration::from_millis(conf.cache.ttl_ms.max(1)),
971            );
972            runtime().prune_named_providers_except(&expected);
973            return Ok(());
974        }
975        // New-style [provider.redis]
976        if let Some(redis_cfg) = provider_cfg.redis {
977            info_ctrl!("init redis knowdb provider({}) ", conf_abs.display(),);
978            crate::redis::init_with_opts(
979                "knowdb",
980                &redis_cfg.connection_uri,
981                redis_cfg.pool_size,
982                redis_cfg.command_timeout_ms,
983            )?;
984            runtime().configure_redis_cache(conf.cache.enabled, conf.cache.capacity);
985            register_fun_map(conf.fun.clone());
986            // redis-only 配置不注册任何 sqldb 命名 provider,清空过期命名注册
987            runtime().prune_named_providers_except(&HashSet::new());
988            return Ok(());
989        }
990    }
991
992    crate::loader::build_authority_from_knowdb(root, knowdb_conf, authority_uri, dict)?;
993    let ro_uri = if let Some(rest) = authority_uri.strip_prefix("file:") {
994        let path_part = rest.split('?').next().unwrap_or(rest);
995        format!("file:{}?mode=ro&uri=true", path_part)
996    } else {
997        authority_uri.to_string()
998    };
999
1000    info_ctrl!("init authority knowdb success({}) ", knowdb_conf.display(),);
1001    init_thread_cloned_from_authority(&ro_uri)?;
1002    runtime().configure_result_cache(
1003        conf.cache.enabled,
1004        conf.cache.capacity,
1005        Duration::from_millis(conf.cache.ttl_ms.max(1)),
1006    );
1007    // 本地 authority 模式:仅保留默认 provider,清理残留的命名 provider
1008    runtime().prune_named_providers_except(&HashSet::from([DEFAULT_SQLDB_NAME.to_string()]));
1009    Ok(())
1010}
1011
1012fn stable_hash(value: &str) -> u64 {
1013    let mut hasher = DefaultHasher::new();
1014    value.hash(&mut hasher);
1015    hasher.finish()
1016}
1017
1018#[cfg(test)]
1019mod tests {
1020    use super::*;
1021    use crate::cache::FieldQueryCache;
1022    use crate::error::KnowReason;
1023    use crate::mem::memdb::MemDB;
1024    use crate::mem::query_util::{COLNAME_CACHE, metadata_cache_key_for_scope};
1025    use crate::runtime::fields_to_params;
1026    use crate::telemetry::{
1027        CacheLayer, CacheTelemetryEvent, KnowledgeTelemetry, QueryTelemetryEvent,
1028        ReloadTelemetryEvent, reset_telemetry,
1029    };
1030    use orion_error::conversion::ToStructError;
1031    use orion_variate::EnvDict;
1032    use std::fs;
1033    use std::hint::black_box;
1034    use std::path::PathBuf;
1035    use std::sync::atomic::{AtomicU64, Ordering};
1036    use std::time::{Duration, Instant};
1037
1038    #[derive(Default)]
1039    struct TestTelemetry {
1040        reload_success: AtomicU64,
1041        reload_failure: AtomicU64,
1042        local_hits: AtomicU64,
1043        local_misses: AtomicU64,
1044        result_hits: AtomicU64,
1045        result_misses: AtomicU64,
1046        metadata_hits: AtomicU64,
1047        metadata_misses: AtomicU64,
1048        query_success: AtomicU64,
1049        query_failure: AtomicU64,
1050    }
1051
1052    impl KnowledgeTelemetry for TestTelemetry {
1053        fn on_cache(&self, event: &CacheTelemetryEvent) {
1054            let counter = match (event.layer, event.outcome) {
1055                (CacheLayer::Local, CacheOutcome::Hit) => &self.local_hits,
1056                (CacheLayer::Local, CacheOutcome::Miss) => &self.local_misses,
1057                (CacheLayer::Result, CacheOutcome::Hit) => &self.result_hits,
1058                (CacheLayer::Result, CacheOutcome::Miss) => &self.result_misses,
1059                (CacheLayer::Metadata, CacheOutcome::Hit) => &self.metadata_hits,
1060                (CacheLayer::Metadata, CacheOutcome::Miss) => &self.metadata_misses,
1061            };
1062            counter.fetch_add(1, Ordering::Relaxed);
1063        }
1064
1065        fn on_reload(&self, event: &ReloadTelemetryEvent) {
1066            let counter = match event.outcome {
1067                crate::telemetry::ReloadOutcome::Success => &self.reload_success,
1068                crate::telemetry::ReloadOutcome::Failure => &self.reload_failure,
1069            };
1070            counter.fetch_add(1, Ordering::Relaxed);
1071        }
1072
1073        fn on_query(&self, event: &QueryTelemetryEvent) {
1074            let counter = if event.success {
1075                &self.query_success
1076            } else {
1077                &self.query_failure
1078            };
1079            counter.fetch_add(1, Ordering::Relaxed);
1080        }
1081    }
1082
1083    fn perf_env_usize(key: &str, default: usize) -> usize {
1084        std::env::var(key)
1085            .ok()
1086            .and_then(|value| value.parse::<usize>().ok())
1087            .unwrap_or(default)
1088    }
1089
1090    fn seed_perf_provider(rows: usize) {
1091        let db = MemDB::instance();
1092        db.execute("CREATE TABLE perf_kv (id INTEGER PRIMARY KEY, value TEXT)")
1093            .expect("create perf_kv");
1094        db.execute("BEGIN IMMEDIATE").expect("begin perf_kv load");
1095        for id in 1..=rows {
1096            let sql = format!("INSERT INTO perf_kv (id, value) VALUES ({id}, 'value_{id}')");
1097            db.execute(sql.as_str()).expect("insert perf_kv row");
1098        }
1099        db.execute("COMMIT").expect("commit perf_kv load");
1100        init_mem_provider(db).expect("init perf provider");
1101    }
1102
1103    #[tokio::test(flavor = "current_thread")]
1104    async fn query_async_works_with_mem_provider() {
1105        let _guard = crate::runtime::runtime_test_guard().lock_async().await;
1106        let db = MemDB::instance();
1107        db.execute("CREATE TABLE async_kv (id INTEGER PRIMARY KEY, value TEXT)")
1108            .expect("create async_kv");
1109        db.execute("INSERT INTO async_kv (id, value) VALUES (1, 'hello')")
1110            .expect("insert async_kv row");
1111        init_mem_provider(db).expect("init mem provider");
1112
1113        let row = query_fields_async(
1114            "SELECT value FROM async_kv WHERE id=:id",
1115            &[DataField::from_digit(":id", 1)],
1116        )
1117        .await
1118        .expect("query async row");
1119        assert_eq!(row.len(), 1);
1120        assert_eq!(row[0].to_string(), "chars(hello)");
1121    }
1122
1123    #[derive(Clone)]
1124    struct PerfQuery {
1125        cache_key: [DataField; 1],
1126        query_params: [DataField; 1],
1127        bypass_req: QueryRequest,
1128        global_req: QueryRequest,
1129    }
1130
1131    fn build_perf_workload(ops: usize, hotset: usize) -> Vec<PerfQuery> {
1132        (0..ops)
1133            .map(|idx| {
1134                let id = ((idx * 17) % hotset + 1) as i64;
1135                let cache_key = [DataField::from_digit("id", id)];
1136                let query_params = [DataField::from_digit(":id", id)];
1137                let bypass_req = QueryRequest::first_row(
1138                    "SELECT value FROM perf_kv WHERE id=:id",
1139                    fields_to_params(&query_params),
1140                    CachePolicy::Bypass,
1141                );
1142                let global_req = QueryRequest::first_row(
1143                    "SELECT value FROM perf_kv WHERE id=:id",
1144                    fields_to_params(&query_params),
1145                    CachePolicy::UseGlobal,
1146                );
1147                PerfQuery {
1148                    cache_key,
1149                    query_params,
1150                    bypass_req,
1151                    global_req,
1152                }
1153            })
1154            .collect()
1155    }
1156
1157    #[derive(Debug, Clone, Copy)]
1158    struct PerfCounters {
1159        result_hits: u64,
1160        result_misses: u64,
1161        local_hits: u64,
1162        local_misses: u64,
1163        metadata_hits: u64,
1164        metadata_misses: u64,
1165    }
1166
1167    #[derive(Debug, Clone)]
1168    struct PerfResult {
1169        name: &'static str,
1170        elapsed: Duration,
1171        ops: usize,
1172        counters: PerfCounters,
1173    }
1174
1175    impl PerfResult {
1176        fn qps(&self) -> f64 {
1177            let secs = self.elapsed.as_secs_f64();
1178            if secs == 0.0 {
1179                self.ops as f64
1180            } else {
1181                self.ops as f64 / secs
1182            }
1183        }
1184    }
1185
1186    fn snapshot_delta(before: &RuntimeSnapshot, after: &RuntimeSnapshot) -> PerfCounters {
1187        PerfCounters {
1188            result_hits: after
1189                .result_cache_hits
1190                .saturating_sub(before.result_cache_hits),
1191            result_misses: after
1192                .result_cache_misses
1193                .saturating_sub(before.result_cache_misses),
1194            local_hits: after
1195                .local_cache_hits
1196                .saturating_sub(before.local_cache_hits),
1197            local_misses: after
1198                .local_cache_misses
1199                .saturating_sub(before.local_cache_misses),
1200            metadata_hits: after
1201                .metadata_cache_hits
1202                .saturating_sub(before.metadata_cache_hits),
1203            metadata_misses: after
1204                .metadata_cache_misses
1205                .saturating_sub(before.metadata_cache_misses),
1206        }
1207    }
1208
1209    fn run_bypass_perf(workload: &[PerfQuery], rows: usize) -> PerfResult {
1210        seed_perf_provider(rows);
1211        let before = runtime_snapshot();
1212        let started = Instant::now();
1213        for item in workload {
1214            let row = runtime()
1215                .execute(&item.bypass_req)
1216                .expect("execute bypass request")
1217                .into_row();
1218            black_box(row);
1219        }
1220        let elapsed = started.elapsed();
1221        let after = runtime_snapshot();
1222        PerfResult {
1223            name: "bypass",
1224            elapsed,
1225            ops: workload.len(),
1226            counters: snapshot_delta(&before, &after),
1227        }
1228    }
1229
1230    fn run_global_cache_perf(workload: &[PerfQuery], rows: usize) -> PerfResult {
1231        seed_perf_provider(rows);
1232        let before = runtime_snapshot();
1233        let started = Instant::now();
1234        for item in workload {
1235            let row = runtime()
1236                .execute(&item.global_req)
1237                .expect("execute global-cache request")
1238                .into_row();
1239            black_box(row);
1240        }
1241        let elapsed = started.elapsed();
1242        let after = runtime_snapshot();
1243        PerfResult {
1244            name: "global_cache",
1245            elapsed,
1246            ops: workload.len(),
1247            counters: snapshot_delta(&before, &after),
1248        }
1249    }
1250
1251    fn run_local_cache_perf(workload: &[PerfQuery], rows: usize) -> PerfResult {
1252        seed_perf_provider(rows);
1253        let mut cache = FieldQueryCache::with_capacity(workload.len().max(1));
1254        let before = runtime_snapshot();
1255        let started = Instant::now();
1256        for item in workload {
1257            let row = cache_query_fields(
1258                "SELECT value FROM perf_kv WHERE id=:id",
1259                &item.cache_key,
1260                &item.query_params,
1261                &mut cache,
1262            );
1263            black_box(row);
1264        }
1265        let elapsed = started.elapsed();
1266        let after = runtime_snapshot();
1267        PerfResult {
1268            name: "local_cache",
1269            elapsed,
1270            ops: workload.len(),
1271            counters: snapshot_delta(&before, &after),
1272        }
1273    }
1274
1275    fn print_perf_result(result: &PerfResult) {
1276        eprintln!(
1277            "[wp-knowledge][cache-perf] scenario={} elapsed_ms={} qps={:.0} result_hit={} result_miss={} local_hit={} local_miss={} metadata_hit={} metadata_miss={}",
1278            result.name,
1279            result.elapsed.as_millis(),
1280            result.qps(),
1281            result.counters.result_hits,
1282            result.counters.result_misses,
1283            result.counters.local_hits,
1284            result.counters.local_misses,
1285            result.counters.metadata_hits,
1286            result.counters.metadata_misses,
1287        );
1288    }
1289
1290    fn uniq_cache_cfg_tmp_dir() -> PathBuf {
1291        use rand::{Rng, rng};
1292        let rnd: u64 = rng().next_u64();
1293        std::env::temp_dir().join(format!("wpk_cache_cfg_{}", rnd))
1294    }
1295
1296    fn write_minimal_knowdb_with_cache(
1297        root: &Path,
1298        enabled: bool,
1299        capacity: usize,
1300        ttl_ms: u64,
1301    ) -> std::path::PathBuf {
1302        let models = root.join("models").join("knowledge");
1303        let example_dir = models.join("example");
1304        fs::create_dir_all(&example_dir).expect("create knowdb models/example");
1305        fs::write(
1306            models.join("knowdb.toml"),
1307            format!(
1308                r#"
1309version = 2
1310base_dir = "."
1311
1312[cache]
1313enabled = {enabled}
1314capacity = {capacity}
1315ttl_ms = {ttl_ms}
1316
1317[csv]
1318has_header = false
1319
1320[[tables]]
1321name = "example"
1322columns.by_index = [0,1]
1323"#
1324            ),
1325        )
1326        .expect("write knowdb.toml");
1327        fs::write(
1328            example_dir.join("create.sql"),
1329            r#"
1330CREATE TABLE IF NOT EXISTS {table} (
1331  id INTEGER PRIMARY KEY,
1332  value TEXT NOT NULL
1333);
1334"#,
1335        )
1336        .expect("write create.sql");
1337        fs::write(
1338            example_dir.join("insert.sql"),
1339            "INSERT INTO {table} (id, value) VALUES (?1, ?2);\n",
1340        )
1341        .expect("write insert.sql");
1342        fs::write(example_dir.join("data.csv"), "1,alpha\n").expect("write data.csv");
1343        models.join("knowdb.toml")
1344    }
1345
1346    fn write_provider_only_knowdb_with_cache(
1347        root: &Path,
1348        provider_kind: &str,
1349        connection_uri: &str,
1350        enabled: bool,
1351        capacity: usize,
1352        ttl_ms: u64,
1353    ) -> std::path::PathBuf {
1354        let models = root.join("models").join("knowledge");
1355        fs::create_dir_all(&models).expect("create knowdb models");
1356        fs::write(
1357            models.join("knowdb.toml"),
1358            format!(
1359                r#"
1360version = 2
1361base_dir = "."
1362
1363[cache]
1364enabled = {enabled}
1365capacity = {capacity}
1366ttl_ms = {ttl_ms}
1367
1368[provider.sqldb]
1369kind = "{provider_kind}"
1370connection_uri = "{connection_uri}"
1371"#
1372            ),
1373        )
1374        .expect("write provider knowdb.toml");
1375        models.join("knowdb.toml")
1376    }
1377
1378    fn restore_default_result_cache_config() {
1379        runtime().configure_result_cache(true, 1024, Duration::from_millis(30_000));
1380    }
1381
1382    #[test]
1383    fn provider_can_be_replaced() {
1384        let _guard = crate::runtime::runtime_test_guard()
1385            .lock()
1386            .expect("provider test guard");
1387        let db1 = MemDB::instance();
1388        db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1389            .expect("create table in db1");
1390        db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1391            .expect("seed db1");
1392        init_mem_provider(db1).expect("init provider db1");
1393        let row = query_row("SELECT value FROM t WHERE id = 1").expect("query db1");
1394        assert_eq!(row[0].to_string(), "chars(first)");
1395
1396        let db2 = MemDB::instance();
1397        db2.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1398            .expect("create table in db2");
1399        db2.execute("INSERT INTO t (id, value) VALUES (1, 'second')")
1400            .expect("seed db2");
1401        init_mem_provider(db2).expect("replace provider with db2");
1402        let row = query_row("SELECT value FROM t WHERE id = 1").expect("query db2");
1403        assert_eq!(row[0].to_string(), "chars(second)");
1404    }
1405
1406    #[test]
1407    fn sqlite_metadata_cache_uses_provider_scope_after_reload() {
1408        let _guard = crate::runtime::runtime_test_guard()
1409            .lock()
1410            .expect("provider test guard");
1411        COLNAME_CACHE.write().expect("metadata cache lock").clear();
1412
1413        let db = MemDB::instance();
1414        db.execute("CREATE TABLE cache_scope_t (id INTEGER PRIMARY KEY, value TEXT)")
1415            .expect("create table");
1416        db.execute("INSERT INTO cache_scope_t (id, value) VALUES (1, 'scope-old')")
1417            .expect("seed table");
1418
1419        let old_scope = MetadataCacheScope {
1420            datasource_id: DatasourceId("sqlite:old".to_string()),
1421            generation: Generation(1),
1422        };
1423        let new_scope = MetadataCacheScope {
1424            datasource_id: DatasourceId("sqlite:new".to_string()),
1425            generation: Generation(2),
1426        };
1427        let old_provider = MemProvider {
1428            db: db.clone(),
1429            metadata_scope: old_scope.clone(),
1430        };
1431
1432        install_provider(
1433            ProviderKind::SqliteAuthority,
1434            new_scope.datasource_id.clone(),
1435            |_generation| {
1436                Ok(Arc::new(MemProvider {
1437                    db: db.clone(),
1438                    metadata_scope: new_scope.clone(),
1439                }))
1440            },
1441        )
1442        .expect("install new provider");
1443
1444        let row = old_provider
1445            .query_row("SELECT value FROM cache_scope_t WHERE id = 1")
1446            .expect("old provider query");
1447        assert_eq!(row[0].to_string(), "chars(scope-old)");
1448
1449        let cache = COLNAME_CACHE.read().expect("metadata cache lock");
1450        assert!(cache.contains(&metadata_cache_key_for_scope(
1451            &old_scope,
1452            "SELECT value FROM cache_scope_t WHERE id = 1",
1453        )));
1454        assert!(!cache.contains(&metadata_cache_key_for_scope(
1455            &new_scope,
1456            "SELECT value FROM cache_scope_t WHERE id = 1",
1457        )));
1458    }
1459
1460    #[tokio::test(flavor = "current_thread")]
1461    async fn async_query_uses_runtime_bridge() {
1462        let _guard = crate::runtime::runtime_test_guard().lock_async().await;
1463        let db = MemDB::instance();
1464        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1465            .expect("create table");
1466        db.execute("INSERT INTO t (id, value) VALUES (1, 'async-first')")
1467            .expect("seed table");
1468        init_mem_provider(db).expect("init provider");
1469
1470        let row = query_row_async("SELECT value FROM t WHERE id = 1")
1471            .await
1472            .expect("async query row");
1473        assert_eq!(row[0].to_string(), "chars(async-first)");
1474    }
1475
1476    #[tokio::test(flavor = "current_thread")]
1477    async fn async_cache_query_fields_hits_local_cache() {
1478        let _guard = crate::runtime::runtime_test_guard().lock_async().await;
1479        let db = MemDB::instance();
1480        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1481            .expect("create table");
1482        db.execute("INSERT INTO t (id, value) VALUES (1, 'async-cache')")
1483            .expect("seed table");
1484        init_mem_provider(db).expect("init provider");
1485
1486        let key = [DataField::from_digit("id", 1)];
1487        let params = [DataField::from_digit(":id", 1)];
1488        let mut cache = FieldQueryCache::default();
1489
1490        let first = cache_query_fields_async(
1491            "SELECT value FROM t WHERE id=:id",
1492            &key,
1493            &params,
1494            &mut cache,
1495        )
1496        .await;
1497        let second = cache_query_fields_async(
1498            "SELECT value FROM t WHERE id=:id",
1499            &key,
1500            &params,
1501            &mut cache,
1502        )
1503        .await;
1504
1505        assert_eq!(first[0].to_string(), "chars(async-cache)");
1506        assert_eq!(second[0].to_string(), "chars(async-cache)");
1507    }
1508
1509    #[test]
1510    fn local_cache_is_cleared_when_generation_changes() {
1511        let _guard = crate::runtime::runtime_test_guard()
1512            .lock()
1513            .expect("provider test guard");
1514        let db1 = MemDB::instance();
1515        db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1516            .expect("create table in db1");
1517        db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1518            .expect("seed db1");
1519        init_mem_provider(db1).expect("init provider db1");
1520
1521        let key = [DataField::from_digit("id", 1)];
1522        let params = [DataField::from_digit(":id", 1)];
1523        let mut cache = FieldQueryCache::default();
1524        let row = cache_query_fields(
1525            "SELECT value FROM t WHERE id=:id",
1526            &key,
1527            &params,
1528            &mut cache,
1529        );
1530        assert_eq!(row[0].to_string(), "chars(first)");
1531
1532        let db2 = MemDB::instance();
1533        db2.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1534            .expect("create table in db2");
1535        db2.execute("INSERT INTO t (id, value) VALUES (1, 'second')")
1536            .expect("seed db2");
1537        init_mem_provider(db2).expect("replace provider with db2");
1538
1539        let row = cache_query_fields(
1540            "SELECT value FROM t WHERE id=:id",
1541            &key,
1542            &params,
1543            &mut cache,
1544        );
1545        assert_eq!(row[0].to_string(), "chars(second)");
1546    }
1547
1548    #[test]
1549    fn local_cache_is_scoped_by_sql_text() {
1550        let _guard = crate::runtime::runtime_test_guard()
1551            .lock()
1552            .expect("provider test guard");
1553        let db = MemDB::instance();
1554        db.execute("CREATE TABLE t1 (id INTEGER PRIMARY KEY, value TEXT)")
1555            .expect("create t1");
1556        db.execute("CREATE TABLE t2 (id INTEGER PRIMARY KEY, value TEXT)")
1557            .expect("create t2");
1558        db.execute("INSERT INTO t1 (id, value) VALUES (1, 'first')")
1559            .expect("seed t1");
1560        db.execute("INSERT INTO t2 (id, value) VALUES (1, 'second')")
1561            .expect("seed t2");
1562        init_mem_provider(db).expect("init provider");
1563
1564        let key = [DataField::from_digit("id", 1)];
1565        let params = [DataField::from_digit(":id", 1)];
1566        let mut cache = FieldQueryCache::default();
1567
1568        let row = cache_query_fields(
1569            "SELECT value FROM t1 WHERE id=:id",
1570            &key,
1571            &params,
1572            &mut cache,
1573        );
1574        assert_eq!(row[0].to_string(), "chars(first)");
1575
1576        let row = cache_query_fields(
1577            "SELECT value FROM t2 WHERE id=:id",
1578            &key,
1579            &params,
1580            &mut cache,
1581        );
1582        assert_eq!(row[0].to_string(), "chars(second)");
1583    }
1584
1585    #[test]
1586    fn runtime_snapshot_tracks_generation_and_cache_size() {
1587        let _guard = crate::runtime::runtime_test_guard()
1588            .lock()
1589            .expect("provider test guard");
1590        let db = MemDB::instance();
1591        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1592            .expect("create table");
1593        db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1594            .expect("seed table");
1595        init_mem_provider(db).expect("init provider");
1596
1597        let mut cache = FieldQueryCache::default();
1598        let key = [DataField::from_digit("id", 1)];
1599        let params = [DataField::from_digit(":id", 1)];
1600        let row = cache_query_fields(
1601            "SELECT value FROM t WHERE id=:id",
1602            &key,
1603            &params,
1604            &mut cache,
1605        );
1606        assert_eq!(row[0].to_string(), "chars(first)");
1607
1608        let snapshot = runtime_snapshot();
1609        assert!(matches!(
1610            snapshot.provider_kind,
1611            Some(ProviderKind::SqliteAuthority)
1612        ));
1613        assert!(snapshot.generation.is_some());
1614        assert!(snapshot.result_cache_len >= 1);
1615        assert!(snapshot.result_cache_capacity >= snapshot.result_cache_len);
1616        assert!(snapshot.metadata_cache_capacity >= snapshot.metadata_cache_len);
1617        assert!(snapshot.reload_successes >= 1);
1618    }
1619
1620    #[test]
1621    fn metadata_cache_is_scoped_by_generation() {
1622        let _guard = crate::runtime::runtime_test_guard()
1623            .lock()
1624            .expect("provider test guard");
1625        let sql = "SELECT value FROM t WHERE id = 1";
1626        let before = runtime_snapshot().metadata_cache_len;
1627
1628        let db1 = MemDB::instance();
1629        db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1630            .expect("create table in db1");
1631        db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1632            .expect("seed db1");
1633        init_mem_provider(db1).expect("init provider db1");
1634        let row = query_row(sql).expect("query db1");
1635        assert_eq!(row[0].to_string(), "chars(first)");
1636        let after_first = runtime_snapshot().metadata_cache_len;
1637        assert!(
1638            after_first > before,
1639            "metadata cache did not record first generation entry: before={before} after_first={after_first}"
1640        );
1641
1642        let db2 = MemDB::instance();
1643        db2.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1644            .expect("create table in db2");
1645        db2.execute("INSERT INTO t (id, value) VALUES (1, 'second')")
1646            .expect("seed db2");
1647        init_mem_provider(db2).expect("replace provider with db2");
1648        let row = query_row(sql).expect("query db2");
1649        assert_eq!(row[0].to_string(), "chars(second)");
1650        let after_second = runtime_snapshot().metadata_cache_len;
1651        assert!(
1652            after_second > after_first,
1653            "metadata cache did not keep a distinct generation entry: after_first={after_first} after_second={after_second}"
1654        );
1655    }
1656
1657    #[test]
1658    fn failed_provider_reload_keeps_previous_provider() {
1659        let _guard = crate::runtime::runtime_test_guard()
1660            .lock()
1661            .expect("provider test guard");
1662        let db1 = MemDB::instance();
1663        db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1664            .expect("create table in db1");
1665        db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1666            .expect("seed db1");
1667        init_mem_provider(db1).expect("init provider db1");
1668        let before_generation = current_generation();
1669
1670        let reload_err = install_provider(
1671            ProviderKind::SqliteAuthority,
1672            datasource_id_for(ProviderKind::SqliteAuthority, "reload-failure"),
1673            |_generation| {
1674                Err(KnowReason::from_logic()
1675                    .to_err()
1676                    .with_detail("expected reload failure"))
1677            },
1678        );
1679        assert!(reload_err.is_err());
1680
1681        let row = query_row("SELECT value FROM t WHERE id = 1").expect("query previous provider");
1682        assert_eq!(row[0].to_string(), "chars(first)");
1683        assert_eq!(current_generation(), before_generation);
1684    }
1685
1686    #[test]
1687    fn runtime_snapshot_records_cache_counters() {
1688        let _guard = crate::runtime::runtime_test_guard()
1689            .lock()
1690            .expect("provider test guard");
1691        let db = MemDB::instance();
1692        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1693            .expect("create table");
1694        db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1695            .expect("seed table");
1696        init_mem_provider(db).expect("init provider");
1697
1698        let before = runtime_snapshot();
1699        let mut cache = FieldQueryCache::default();
1700        let key = [DataField::from_digit("id", 1)];
1701        let params = [DataField::from_digit(":id", 1)];
1702        let row = cache_query_fields(
1703            "SELECT value FROM t WHERE id=:id",
1704            &key,
1705            &params,
1706            &mut cache,
1707        );
1708        assert_eq!(row[0].to_string(), "chars(first)");
1709        let row = cache_query_fields(
1710            "SELECT value FROM t WHERE id=:id",
1711            &key,
1712            &params,
1713            &mut cache,
1714        );
1715        assert_eq!(row[0].to_string(), "chars(first)");
1716
1717        let after = runtime_snapshot();
1718        assert!(after.local_cache_hits > before.local_cache_hits);
1719        assert!(after.local_cache_misses > before.local_cache_misses);
1720        assert!(after.result_cache_misses > before.result_cache_misses);
1721        assert!(after.metadata_cache_misses > before.metadata_cache_misses);
1722    }
1723
1724    #[test]
1725    fn telemetry_receives_reload_cache_and_query_events() {
1726        let _guard = crate::runtime::runtime_test_guard()
1727            .lock()
1728            .expect("provider test guard");
1729        let telemetry_impl = Arc::new(TestTelemetry::default());
1730        let previous = install_runtime_telemetry(telemetry_impl.clone());
1731
1732        let db = MemDB::instance();
1733        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1734            .expect("create table");
1735        db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1736            .expect("seed table");
1737        init_mem_provider(db).expect("init provider");
1738
1739        let mut cache = FieldQueryCache::default();
1740        let key = [DataField::from_digit("id", 1)];
1741        let params = [DataField::from_digit(":id", 1)];
1742        let row = cache_query_fields(
1743            "SELECT value FROM t WHERE id=:id",
1744            &key,
1745            &params,
1746            &mut cache,
1747        );
1748        assert_eq!(row[0].to_string(), "chars(first)");
1749        let row = cache_query_fields(
1750            "SELECT value FROM t WHERE id=:id",
1751            &key,
1752            &params,
1753            &mut cache,
1754        );
1755        assert_eq!(row[0].to_string(), "chars(first)");
1756
1757        let reload_err = install_provider(
1758            ProviderKind::SqliteAuthority,
1759            datasource_id_for(ProviderKind::SqliteAuthority, "telemetry-failure"),
1760            |_generation| {
1761                Err(KnowReason::from_logic()
1762                    .to_err()
1763                    .with_detail("expected telemetry reload failure"))
1764            },
1765        );
1766        assert!(reload_err.is_err());
1767
1768        install_runtime_telemetry(previous);
1769        reset_telemetry();
1770
1771        assert!(telemetry_impl.reload_success.load(Ordering::Relaxed) >= 1);
1772        assert!(telemetry_impl.reload_failure.load(Ordering::Relaxed) >= 1);
1773        assert!(telemetry_impl.local_hits.load(Ordering::Relaxed) >= 1);
1774        assert!(telemetry_impl.local_misses.load(Ordering::Relaxed) >= 1);
1775        assert!(telemetry_impl.result_misses.load(Ordering::Relaxed) >= 1);
1776        assert!(telemetry_impl.metadata_misses.load(Ordering::Relaxed) >= 1);
1777        assert!(telemetry_impl.query_success.load(Ordering::Relaxed) >= 1);
1778    }
1779
1780    #[test]
1781    #[ignore = "manual perf comparison; run with cargo test cache_perf_reports_cache_vs_no_cache -- --ignored --nocapture"]
1782    fn cache_perf_reports_cache_vs_no_cache() {
1783        let _guard = crate::runtime::runtime_test_guard()
1784            .lock()
1785            .expect("provider test guard");
1786        let rows = perf_env_usize("WP_KDB_PERF_ROWS", 10_000).max(1);
1787        let ops = perf_env_usize("WP_KDB_PERF_OPS", 120_000).max(1);
1788        let hotset = perf_env_usize("WP_KDB_PERF_HOTSET", 128).clamp(1, rows);
1789        let workload = build_perf_workload(ops, hotset);
1790
1791        eprintln!(
1792            "[wp-knowledge][cache-perf] rows={} ops={} hotset={} sql=SELECT value FROM perf_kv WHERE id=:id",
1793            rows, ops, hotset
1794        );
1795
1796        let bypass = run_bypass_perf(&workload, rows);
1797        let global = run_global_cache_perf(&workload, rows);
1798        let local = run_local_cache_perf(&workload, rows);
1799
1800        print_perf_result(&bypass);
1801        print_perf_result(&global);
1802        print_perf_result(&local);
1803
1804        eprintln!(
1805            "[wp-knowledge][cache-perf] speedup global_vs_bypass={:.2}x local_vs_bypass={:.2}x",
1806            bypass.elapsed.as_secs_f64() / global.elapsed.as_secs_f64(),
1807            bypass.elapsed.as_secs_f64() / local.elapsed.as_secs_f64(),
1808        );
1809
1810        assert_eq!(bypass.counters.result_hits, 0);
1811        assert_eq!(bypass.counters.result_misses, 0);
1812        assert_eq!(bypass.counters.local_hits, 0);
1813        assert_eq!(bypass.counters.local_misses, 0);
1814        assert!(global.counters.result_hits > 0);
1815        assert!(global.counters.result_misses > 0);
1816        assert_eq!(global.counters.local_hits, 0);
1817        assert_eq!(global.counters.local_misses, 0);
1818        assert!(local.counters.local_hits > 0);
1819        assert!(local.counters.local_misses > 0);
1820        assert!(local.counters.result_misses > 0);
1821    }
1822
1823    #[test]
1824    fn init_thread_cloned_from_knowdb_applies_result_cache_config() {
1825        let _guard = crate::runtime::runtime_test_guard()
1826            .lock()
1827            .expect("provider test guard");
1828        let root = uniq_cache_cfg_tmp_dir();
1829        let conf_path = write_minimal_knowdb_with_cache(&root, true, 7, 5);
1830        let auth_file = root.join(".run").join("authority.sqlite");
1831        fs::create_dir_all(auth_file.parent().expect("authority parent"))
1832            .expect("create authority parent");
1833        let authority_uri = format!("file:{}?mode=rwc&uri=true", auth_file.display());
1834
1835        init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default())
1836            .expect("init knowdb with cache config");
1837
1838        let snapshot = runtime_snapshot();
1839        assert!(snapshot.result_cache_enabled);
1840        assert_eq!(snapshot.result_cache_capacity, 7);
1841        assert_eq!(snapshot.result_cache_ttl_ms, 5);
1842
1843        let req = QueryRequest::first_row(
1844            "SELECT value FROM example WHERE id=:id",
1845            fields_to_params(&[DataField::from_digit(":id", 1)]),
1846            CachePolicy::UseGlobal,
1847        );
1848        let before = runtime_snapshot();
1849        let row = runtime()
1850            .execute(&req)
1851            .expect("first result-cache query")
1852            .into_row();
1853        assert_eq!(row[0].to_string(), "chars(alpha)");
1854        let row = runtime()
1855            .execute(&req)
1856            .expect("second result-cache query")
1857            .into_row();
1858        assert_eq!(row[0].to_string(), "chars(alpha)");
1859        std::thread::sleep(Duration::from_millis(12));
1860        let row = runtime()
1861            .execute(&req)
1862            .expect("expired result-cache query")
1863            .into_row();
1864        assert_eq!(row[0].to_string(), "chars(alpha)");
1865        let after = runtime_snapshot();
1866
1867        assert!(after.result_cache_hits > before.result_cache_hits);
1868        assert!(after.result_cache_misses >= before.result_cache_misses + 2);
1869
1870        restore_default_result_cache_config();
1871        let _ = fs::remove_dir_all(&root);
1872    }
1873
1874    #[test]
1875    fn disabled_result_cache_from_knowdb_config_forces_bypass() {
1876        let _guard = crate::runtime::runtime_test_guard()
1877            .lock()
1878            .expect("provider test guard");
1879        let root = uniq_cache_cfg_tmp_dir();
1880        let conf_path = write_minimal_knowdb_with_cache(&root, false, 3, 30_000);
1881        let auth_file = root.join(".run").join("authority.sqlite");
1882        fs::create_dir_all(auth_file.parent().expect("authority parent"))
1883            .expect("create authority parent");
1884        let authority_uri = format!("file:{}?mode=rwc&uri=true", auth_file.display());
1885
1886        init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default())
1887            .expect("init knowdb with cache disabled");
1888
1889        let snapshot = runtime_snapshot();
1890        assert!(!snapshot.result_cache_enabled);
1891        assert_eq!(snapshot.result_cache_capacity, 3);
1892        assert_eq!(snapshot.result_cache_ttl_ms, 30_000);
1893
1894        let req = QueryRequest::first_row(
1895            "SELECT value FROM example WHERE id=:id",
1896            fields_to_params(&[DataField::from_digit(":id", 1)]),
1897            CachePolicy::UseGlobal,
1898        );
1899        let before = runtime_snapshot();
1900        let _ = runtime()
1901            .execute(&req)
1902            .expect("first bypassed result-cache query");
1903        let _ = runtime()
1904            .execute(&req)
1905            .expect("second bypassed result-cache query");
1906        let after = runtime_snapshot();
1907
1908        assert_eq!(after.result_cache_hits, before.result_cache_hits);
1909        assert_eq!(after.result_cache_misses, before.result_cache_misses);
1910
1911        restore_default_result_cache_config();
1912        let _ = fs::remove_dir_all(&root);
1913    }
1914
1915    #[test]
1916    fn failed_knowdb_provider_init_does_not_apply_cache_config() {
1917        let _guard = crate::runtime::runtime_test_guard()
1918            .lock()
1919            .expect("provider test guard");
1920        restore_default_result_cache_config();
1921
1922        let db = MemDB::instance();
1923        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1924            .expect("create table");
1925        db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1926            .expect("seed table");
1927        init_mem_provider(db).expect("init provider");
1928
1929        let before = runtime_snapshot();
1930        let root = uniq_cache_cfg_tmp_dir();
1931        let conf_path = write_provider_only_knowdb_with_cache(
1932            &root,
1933            "mysql",
1934            "not-a-valid-mysql-url",
1935            false,
1936            3,
1937            5,
1938        );
1939        let authority_uri = format!(
1940            "file:{}?mode=rwc&uri=true",
1941            root.join("unused.sqlite").display()
1942        );
1943
1944        let err =
1945            init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default());
1946        assert!(err.is_err());
1947
1948        let after = runtime_snapshot();
1949        assert_eq!(after.result_cache_enabled, before.result_cache_enabled);
1950        assert_eq!(after.result_cache_capacity, before.result_cache_capacity);
1951        assert_eq!(after.result_cache_ttl_ms, before.result_cache_ttl_ms);
1952
1953        let row = query_row("SELECT value FROM t WHERE id = 1").expect("query previous provider");
1954        assert_eq!(row[0].to_string(), "chars(first)");
1955
1956        let _ = fs::remove_dir_all(&root);
1957    }
1958
1959    // 安装一个命名内存 provider(不替换默认 provider)。
1960    fn install_named_mem_provider(name: &str, db: MemDB) {
1961        let datasource_id = DatasourceId::from_seed(ProviderKind::SqliteAuthority, name);
1962        let ds_for_scope = datasource_id.clone();
1963        runtime()
1964            .install_provider_named(
1965                name,
1966                ProviderKind::SqliteAuthority,
1967                datasource_id,
1968                |generation| {
1969                    Ok(Arc::new(MemProvider {
1970                        db,
1971                        metadata_scope: MetadataCacheScope {
1972                            datasource_id: ds_for_scope.clone(),
1973                            generation,
1974                        },
1975                    }))
1976                },
1977                false,
1978            )
1979            .expect("install named mem provider");
1980    }
1981
1982    #[test]
1983    fn cache_query_fields_route_separates_named_and_default_providers() {
1984        let _guard = crate::runtime::runtime_test_guard()
1985            .lock()
1986            .expect("provider test guard");
1987
1988        // 默认 provider:route_t.value = 'default'
1989        let db_default = MemDB::instance();
1990        db_default
1991            .execute("CREATE TABLE route_t (id INTEGER PRIMARY KEY, value TEXT)")
1992            .expect("create default table");
1993        db_default
1994            .execute("INSERT INTO route_t (id, value) VALUES (1, 'default')")
1995            .expect("seed default");
1996        init_mem_provider(db_default).expect("init default provider");
1997
1998        // 命名 provider "geo":route_t.value = 'geo'
1999        let db_geo = MemDB::instance();
2000        db_geo
2001            .execute("CREATE TABLE route_t (id INTEGER PRIMARY KEY, value TEXT)")
2002            .expect("create geo table");
2003        db_geo
2004            .execute("INSERT INTO route_t (id, value) VALUES (1, 'geo')")
2005            .expect("seed geo");
2006        install_named_mem_provider("geo", db_geo);
2007
2008        let sql = "SELECT value FROM route_t WHERE id=:id";
2009        let c_params = [DataField::from_digit("id", 1)];
2010        let query_params = [DataField::from_digit(":id", 1)];
2011        let mut cache = FieldQueryCache::default();
2012
2013        let named =
2014            cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
2015        assert_eq!(named[0].to_string(), "chars(geo)");
2016
2017        // 同一 cache 实例切到默认 provider,本地缓存必须按 provider 隔离
2018        let default = cache_query_fields_route(None, sql, 0, &c_params, &query_params, &mut cache);
2019        assert_eq!(default[0].to_string(), "chars(default)");
2020
2021        // 再查命名 provider,仍返回 geo
2022        let named_again =
2023            cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
2024        assert_eq!(named_again[0].to_string(), "chars(geo)");
2025    }
2026
2027    #[test]
2028    fn mixed_provider_queries_keep_local_cache_per_provider() {
2029        let _guard = crate::runtime::runtime_test_guard()
2030            .lock()
2031            .expect("provider test guard");
2032
2033        let db_default = MemDB::instance();
2034        db_default
2035            .execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
2036            .expect("create default table");
2037        db_default
2038            .execute("INSERT INTO t (id, value) VALUES (1, 'default')")
2039            .expect("seed default");
2040        init_mem_provider(db_default).expect("init default provider");
2041
2042        let db_geo = MemDB::instance();
2043        db_geo
2044            .execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
2045            .expect("create geo table");
2046        db_geo
2047            .execute("INSERT INTO t (id, value) VALUES (1, 'geo')")
2048            .expect("seed geo");
2049        install_named_mem_provider("geo", db_geo);
2050
2051        let sql = "SELECT value FROM t WHERE id=:id";
2052        let c_params = [DataField::from_digit("id", 1)];
2053        let query_params = [DataField::from_digit(":id", 1)];
2054        let mut cache = FieldQueryCache::default();
2055
2056        let before = runtime_snapshot();
2057        // 1) 命名 provider:miss → 执行
2058        let row =
2059            cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
2060        assert_eq!(row[0].to_string(), "chars(geo)");
2061        // 2) 默认 provider:scope 不同,不应清空 geo 的本地缓存
2062        let row = cache_query_fields_route(None, sql, 0, &c_params, &query_params, &mut cache);
2063        assert_eq!(row[0].to_string(), "chars(default)");
2064        // 3) 再查命名 provider:应命中本地缓存(不重新执行)
2065        let row =
2066            cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
2067        assert_eq!(row[0].to_string(), "chars(geo)");
2068        let after = runtime_snapshot();
2069        assert!(
2070            after.local_cache_hits > before.local_cache_hits,
2071            "命名 provider 二次查询应命中本地缓存(按 provider 隔离代际)"
2072        );
2073    }
2074
2075    #[test]
2076    fn local_authority_init_prunes_stale_named_providers() {
2077        let _guard = crate::runtime::runtime_test_guard()
2078            .lock()
2079            .expect("provider test guard");
2080
2081        // 先安装命名 provider,模拟上一份配置的残留
2082        let db = MemDB::instance();
2083        db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
2084            .expect("create table");
2085        db.execute("INSERT INTO t (id, value) VALUES (1, 'x')")
2086            .expect("seed table");
2087        install_named_mem_provider("geo", db);
2088        assert!(provider_exists("geo"));
2089
2090        // 再用本地 authority 配置初始化 → 应清理 geo,仅保留 default
2091        let root = uniq_cache_cfg_tmp_dir();
2092        let conf_path = write_minimal_knowdb_with_cache(&root, true, 16, 30_000);
2093        let auth_file = root.join(".run").join("authority.sqlite");
2094        fs::create_dir_all(auth_file.parent().expect("authority parent"))
2095            .expect("create authority parent");
2096        let authority_uri = format!("file:{}?mode=rwc&uri=true", auth_file.display());
2097
2098        init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default())
2099            .expect("init local authority");
2100
2101        assert!(!provider_exists("geo"));
2102        assert!(provider_exists(DEFAULT_SQLDB_NAME));
2103
2104        restore_default_result_cache_config();
2105        let _ = fs::remove_dir_all(&root);
2106    }
2107
2108    #[test]
2109    fn cache_query_fields_route_unknown_provider_returns_empty() {
2110        let _guard = crate::runtime::runtime_test_guard()
2111            .lock()
2112            .expect("provider test guard");
2113        let c_params = [DataField::from_digit("id", 1)];
2114        let query_params = [DataField::from_digit(":id", 1)];
2115        let mut cache = FieldQueryCache::default();
2116        let out = cache_query_fields_route(
2117            Some("nope"),
2118            "SELECT value FROM t WHERE id=:id",
2119            0,
2120            &c_params,
2121            &query_params,
2122            &mut cache,
2123        );
2124        assert!(out.is_empty());
2125    }
2126
2127    #[test]
2128    fn query_for_named_provider_returns_rows() {
2129        let _guard = crate::runtime::runtime_test_guard()
2130            .lock()
2131            .expect("provider test guard");
2132        let db = MemDB::instance();
2133        db.execute("CREATE TABLE named_t (id INTEGER PRIMARY KEY, value TEXT)")
2134            .expect("create named table");
2135        db.execute("INSERT INTO named_t (id, value) VALUES (1, 'named')")
2136            .expect("seed named");
2137        install_named_mem_provider("geo", db);
2138
2139        let rows = query_for("geo", "SELECT value FROM named_t WHERE id = 1").expect("query geo");
2140        assert_eq!(rows.len(), 1);
2141        assert_eq!(rows[0][0].to_string(), "chars(named)");
2142        assert!(provider_exists("geo"));
2143        assert!(!provider_exists("nope"));
2144        assert!(provider_names().iter().any(|name| name == "geo"));
2145    }
2146
2147    #[tokio::test(flavor = "current_thread")]
2148    async fn query_async_for_named_provider_works() {
2149        let _guard = crate::runtime::runtime_test_guard().lock_async().await;
2150        let db = MemDB::instance();
2151        db.execute("CREATE TABLE named_t (id INTEGER PRIMARY KEY, value TEXT)")
2152            .expect("create named table");
2153        db.execute("INSERT INTO named_t (id, value) VALUES (1, 'async-named')")
2154            .expect("seed named");
2155        install_named_mem_provider("geo", db);
2156
2157        let rows = query_async_for("geo", "SELECT value FROM named_t WHERE id = 1")
2158            .await
2159            .expect("query geo");
2160        assert_eq!(rows[0][0].to_string(), "chars(async-named)");
2161    }
2162}