wp_knowledge/mem/
memdb.rs

1use crate::DBQuery;
2use crate::cache::CacheAble;
3use crate::error::{KnowReason, KnowledgeResult};
4use crate::mem::RowData;
5use crate::mem::stub::StubMDB;
6use csv::Reader;
7use enum_dispatch::enum_dispatch;
8use lazy_static::lazy_static;
9use orion_error::conversion::ErrorWith;
10use orion_error::conversion::{SourceRawErr, ToStructError};
11use r2d2_sqlite::SqliteConnectionManager;
12use rusqlite::OpenFlags;
13use rusqlite::Params;
14use rusqlite::ToSql;
15use rusqlite::types::ToSqlOutput;
16use rusqlite::types::Value;
17use std::path::PathBuf;
18use wp_log::debug_kdb;
19use wp_log::info_kdb;
20use wp_log::warn_kdb;
21use wp_model_core::model;
22use wp_model_core::model::DataField;
23
24use super::SqlNamedParam;
25use crate::loader::ProviderKind;
26use crate::runtime::MetadataCacheScope;
27
28lazy_static! {
29    // Important: Use a single SQLite in-memory connection so schema/data persist across calls.
30    // r2d2 with `memory()` creates isolated DBs per connection; limit pool to size=1 to reuse
31    // the same connection and avoid "no such table" issues when different checkouts observe
32    // different ephemeral databases.
33    pub static ref MEM_SQLITE_INS: r2d2::Pool<SqliteConnectionManager> =
34        r2d2::Pool::builder()
35            .max_size(1)
36            .build(SqliteConnectionManager::memory())
37            .expect("init SQLite memory pool (size=1) failed");
38}
39
40#[derive(Debug, Clone)]
41pub struct MemDB {
42    conn: r2d2::Pool<SqliteConnectionManager>,
43}
44
45#[derive(Debug, Clone)]
46#[enum_dispatch(DBQuery)]
47pub enum MDBEnum {
48    Stub(StubMDB),
49    Use(MemDB),
50}
51impl Default for MDBEnum {
52    fn default() -> Self {
53        MDBEnum::Stub(StubMDB {})
54    }
55}
56impl MDBEnum {
57    pub fn global() -> Self {
58        MDBEnum::Use(MemDB::global())
59    }
60    pub fn load_test() -> KnowledgeResult<()> {
61        MemDB::load_test()?;
62        Ok(())
63    }
64}
65
66pub fn cache_query<const N: usize, P: Params>(
67    db: &MDBEnum,
68    sql: &str,
69    c_params: &[DataField; N],
70    q_params: P,
71    cache: &mut impl CacheAble<DataField, RowData, N>,
72) -> RowData {
73    crate::cache_util::cache_query_impl(c_params, cache, || db.query_row_params(sql, q_params))
74}
75impl ToSql for SqlNamedParam {
76    fn to_sql(&self) -> rusqlite::Result<rusqlite::types::ToSqlOutput<'_>> {
77        match self.0.get_value() {
78            model::Value::Bool(v) => Ok(ToSqlOutput::Owned(Value::Integer(if *v { 1 } else { 0 }))),
79            model::Value::Null => Ok(ToSqlOutput::Owned(Value::Null)),
80            model::Value::Chars(v) => Ok(ToSqlOutput::Owned(Value::Text(v.to_string()))),
81            model::Value::Symbol(v) => Ok(ToSqlOutput::Owned(Value::Text(v.to_string()))),
82            model::Value::Time(v) => Ok(ToSqlOutput::Owned(Value::Text(v.to_string()))),
83            model::Value::Digit(v) => Ok(ToSqlOutput::Owned(Value::Integer(*v))),
84            // 任意精度整数：超出 i64 范围，以十进制文本绑定；SQLite 数值比较按 affinity 自动转换
85            model::Value::BigUint(v) => Ok(ToSqlOutput::Owned(Value::Text(v.to_string()))),
86            model::Value::Hex(v) => Ok(ToSqlOutput::Owned(Value::Text(v.to_string()))),
87            model::Value::Float(v) => Ok(ToSqlOutput::Owned(Value::Real(*v))),
88            model::Value::IpNet(v) => Ok(ToSqlOutput::Owned(Value::Text(v.to_string()))),
89            model::Value::IpAddr(v) => Ok(ToSqlOutput::Owned(Value::Text(v.to_string()))),
90            model::Value::Ignore(_) => Ok(ToSqlOutput::Owned(Value::Null)),
91            model::Value::Obj(v) => Ok(ToSqlOutput::Owned(Value::Text(format!("{:?}", v)))),
92            model::Value::Array(v) => Ok(ToSqlOutput::Owned(Value::Text(format!("{:?}", v)))),
93            model::Value::Domain(v) => Ok(ToSqlOutput::Owned(Value::Text(v.0.to_string()))),
94            model::Value::Url(v) => Ok(ToSqlOutput::Owned(Value::Text(v.0.to_string()))),
95            model::Value::Email(v) => Ok(ToSqlOutput::Owned(Value::Text(v.0.to_string()))),
96            model::Value::IdCard(v) => Ok(ToSqlOutput::Owned(Value::Text(v.0.to_string()))),
97            model::Value::MobilePhone(v) => Ok(ToSqlOutput::Owned(Value::Text(v.0.to_string()))),
98        }
99    }
100}
101
102impl DBQuery for MemDB {
103    fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
104        let conn = self
105            .conn
106            .get()
107            .source_raw_err(KnowReason::from_res(), "source error")
108            .doing("get memdb connect")?;
109        let _ = crate::sqlite_ext::register_builtin(&conn);
110        super::query_util::query_cached(&conn, sql, [])
111    }
112
113    fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
114        let conn = self
115            .conn
116            .get()
117            .source_raw_err(KnowReason::from_res(), "source error")
118            .doing("get memdb connect")?;
119        // Ensure SQLite UDFs are available on this connection (ip4_int/cidr4_* etc.)
120        let _ = crate::sqlite_ext::register_builtin(&conn);
121        super::query_util::query_first_row_cached(&conn, sql, [])
122    }
123
124    fn query_row_params<P: Params>(&self, sql: &str, params: P) -> KnowledgeResult<RowData> {
125        debug_kdb!("[memdb] query_row_params: {}", sql);
126        let conn = self
127            .conn
128            .get()
129            .source_raw_err(KnowReason::from_res(), "source error")?;
130        // Ensure SQLite UDFs are available on this connection
131        let _ = crate::sqlite_ext::register_builtin(&conn);
132        super::query_util::query_first_row_cached(&conn, sql, params)
133    }
134
135    fn query_row_tdos<P: Params>(
136        &self,
137        _sql: &str,
138        _params: &[DataField; 2],
139    ) -> KnowledgeResult<RowData> {
140        //let data: [TDOParams; 2] = [TDOParams(&params[0]), TDOParams(&params[1])];
141        //params.iter().for_each(|x| data.push(TDOParams(x)));
142        //self.query_row_params(sql, data)
143        todo!();
144    }
145}
146impl MemDB {
147    pub fn query_with_scope(
148        &self,
149        scope: &MetadataCacheScope,
150        sql: &str,
151    ) -> KnowledgeResult<Vec<RowData>> {
152        let conn = self
153            .conn
154            .get()
155            .source_raw_err(KnowReason::from_res(), "source error")
156            .doing("get memdb connect")?;
157        let _ = crate::sqlite_ext::register_builtin(&conn);
158        super::query_util::query_cached_with_scope(
159            &conn,
160            scope,
161            Some(ProviderKind::SqliteAuthority),
162            sql,
163            [],
164        )
165    }
166
167    pub fn query_row_with_scope(
168        &self,
169        scope: &MetadataCacheScope,
170        sql: &str,
171    ) -> KnowledgeResult<RowData> {
172        let conn = self
173            .conn
174            .get()
175            .source_raw_err(KnowReason::from_res(), "source error")
176            .doing("get memdb connect")?;
177        let _ = crate::sqlite_ext::register_builtin(&conn);
178        super::query_util::query_first_row_cached_with_scope(
179            &conn,
180            scope,
181            Some(ProviderKind::SqliteAuthority),
182            sql,
183            [],
184        )
185    }
186
187    pub fn query_fields_with_scope(
188        &self,
189        scope: &MetadataCacheScope,
190        sql: &str,
191        params: &[DataField],
192    ) -> KnowledgeResult<Vec<RowData>> {
193        let conn = self
194            .conn
195            .get()
196            .source_raw_err(KnowReason::from_res(), "source error")
197            .doing("get memdb connect")?;
198        let _ = crate::sqlite_ext::register_builtin(&conn);
199        let named_params = params
200            .iter()
201            .cloned()
202            .map(SqlNamedParam)
203            .collect::<Vec<_>>();
204        let refs: Vec<(&str, &dyn ToSql)> = named_params
205            .iter()
206            .map(|param| (param.0.get_name(), param as &dyn ToSql))
207            .collect();
208        super::query_util::query_cached_with_scope(
209            &conn,
210            scope,
211            Some(ProviderKind::SqliteAuthority),
212            sql,
213            refs.as_slice(),
214        )
215    }
216
217    pub fn query_named_fields_with_scope(
218        &self,
219        scope: &MetadataCacheScope,
220        sql: &str,
221        params: &[DataField],
222    ) -> KnowledgeResult<RowData> {
223        self.query_fields_with_scope(scope, sql, params)
224            .map(|rows| rows.into_iter().next().unwrap_or_default())
225    }
226
227    pub fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
228        let conn = self
229            .conn
230            .get()
231            .source_raw_err(KnowReason::from_res(), "source error")
232            .doing("get memdb connect")?;
233        let _ = crate::sqlite_ext::register_builtin(&conn);
234        let named_params = params
235            .iter()
236            .cloned()
237            .map(SqlNamedParam)
238            .collect::<Vec<_>>();
239        let refs: Vec<(&str, &dyn ToSql)> = named_params
240            .iter()
241            .map(|param| (param.0.get_name(), param as &dyn ToSql))
242            .collect();
243        super::query_util::query_cached(&conn, sql, refs.as_slice())
244    }
245
246    pub fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
247        self.query_fields(sql, params)
248            .map(|rows| rows.into_iter().next().unwrap_or_default())
249    }
250
251    pub fn instance() -> Self {
252        // Provide a single-connection pool for a consistent in-memory DB view
253        let manager = SqliteConnectionManager::memory();
254        let pool = r2d2::Pool::builder()
255            .max_size(1)
256            .build(manager)
257            .expect("init SQLite memory pool (size=1) failed");
258        Self { conn: pool }
259    }
260    /// Experimental: shared in-memory SQLite via URI with a pool size > 1.
261    /// Requires SQLite compiled with shared-cache support.
262    pub fn shared_pool(max_size: u32) -> KnowledgeResult<Self> {
263        // Shared in-memory URI. Every connection to this URI shares same DB.
264        // Note: this depends on platform SQLite features.
265        let uri = "file:wp_knowledge_shm?mode=memory&cache=shared";
266        let manager = SqliteConnectionManager::file(uri).with_flags(
267            OpenFlags::SQLITE_OPEN_READ_WRITE
268                | OpenFlags::SQLITE_OPEN_CREATE
269                | OpenFlags::SQLITE_OPEN_URI,
270        );
271        let pool = r2d2::Pool::builder()
272            .max_size(max_size)
273            .build(manager)
274            .source_raw_err(KnowReason::from_res(), "source error")?;
275        Ok(Self { conn: pool })
276    }
277
278    /// Create a MemDB backed by a file path with custom flags and pool size.
279    pub fn new_file(
280        path: &str,
281        max_size: u32,
282        flags: rusqlite::OpenFlags,
283    ) -> KnowledgeResult<Self> {
284        let manager = r2d2_sqlite::SqliteConnectionManager::file(path).with_flags(flags);
285        let pool = r2d2::Pool::builder()
286            .max_size(max_size)
287            .build(manager)
288            .source_raw_err(KnowReason::from_res(), "source error")?;
289        Ok(Self { conn: pool })
290    }
291    // V1 init_load_by_conf removed: use loader::build_authority_from_knowdb for V2
292
293    /// Execute a closure with a checked-out SQLite connection from the pool.
294    /// Useful for one-time prepared statements or specialized operations.
295    pub fn with_conn<T, F: FnOnce(&rusqlite::Connection) -> anyhow::Result<T>>(
296        &self,
297        f: F,
298    ) -> anyhow::Result<T> {
299        let pooled = self.conn.get()?;
300        let conn_ref: &rusqlite::Connection = &pooled;
301        f(conn_ref)
302    }
303
304    pub fn table_create(&self, sql: &str) -> KnowledgeResult<()> {
305        let conn = self
306            .conn
307            .get()
308            .source_raw_err(KnowReason::from_res(), "source error")?;
309        conn.execute(sql, ())
310            .source_raw_err(KnowReason::from_rule(), "source error")?;
311        debug_kdb!("crate table: {} ", sql);
312        Ok(())
313    }
314    pub fn execute(&self, sql: &str) -> KnowledgeResult<()> {
315        let conn = self
316            .conn
317            .get()
318            .source_raw_err(KnowReason::from_res(), "source error")?;
319        conn.execute(sql, ())
320            .source_raw_err(KnowReason::from_rule(), "source error")?;
321        debug_kdb!("execute: {} ", sql);
322        Ok(())
323    }
324
325    pub fn table_clean(&self, sql: &str) -> KnowledgeResult<()> {
326        let conn = self
327            .conn
328            .get()
329            .source_raw_err(KnowReason::from_res(), "source error")?;
330        conn.execute(sql, ())
331            .source_raw_err(KnowReason::from_rule(), "source error")?;
332        debug_kdb!("clean table: {} ", sql);
333        Ok(())
334    }
335
336    pub fn table_load(
337        &self,
338        sql: &str,
339        csv_path: PathBuf,
340        cols: Vec<usize>,
341        max: usize,
342    ) -> KnowledgeResult<usize> {
343        info_kdb!("load table data in {}", csv_path.display());
344        if !csv_path.exists() {
345            warn_kdb!("{} not find, load knowdb failed", csv_path.display());
346            return Ok(0);
347        }
348        let mut rdr =
349            Reader::from_path(&csv_path).source_raw_err(KnowReason::from_res(), "source error")?;
350        let conn = self
351            .conn
352            .get()
353            .source_raw_err(KnowReason::from_res(), "source error")?;
354        let mut load_cnt: usize = 0;
355        // Prepare once outside loop for performance
356        let mut stmt = conn
357            .prepare(sql)
358            .source_raw_err(KnowReason::from_rule(), "source error")?;
359        for (idx, result) in rdr.records().enumerate() {
360            if load_cnt >= max {
361                break;
362            }
363            let record = result.map_err(|e| {
364                KnowReason::from_rule().to_err().with_detail(format!(
365                    "read csv record failed at line {}: {}",
366                    idx + 1,
367                    e
368                ))
369            })?;
370
371            // Basic bounds check to avoid panic on bad column indices
372            if let Some(max_col) = cols.iter().max()
373                && *max_col >= record.len()
374            {
375                return Err(KnowReason::from_rule().to_err().with_detail(format!(
376                    "csv has insufficient columns at line {}: need index {}, got {} columns",
377                    idx + 1,
378                    *max_col,
379                    record.len()
380                )));
381            }
382
383            // Unified dynamic binding (strict): any missing column is an error
384            let mut vec: Vec<&str> = Vec::with_capacity(cols.len());
385            for &ci in &cols {
386                let v = record.get(ci).ok_or_else(|| {
387                    KnowReason::from_rule().to_err().with_detail(format!(
388                        "line {} col {} missing",
389                        idx + 1,
390                        ci
391                    ))
392                })?;
393                vec.push(v);
394            }
395            let params = rusqlite::params_from_iter(vec);
396            stmt.execute(params)
397                .source_raw_err(KnowReason::from_rule(), "source error")?;
398            load_cnt += 1;
399        }
400        info_kdb!("from {} load data cnt: {}", csv_path.display(), load_cnt);
401        Ok(load_cnt)
402    }
403
404    pub fn check_data(&self, table: &str, scope: (usize, usize)) -> KnowledgeResult<usize> {
405        let conn = self
406            .conn
407            .get()
408            .source_raw_err(KnowReason::from_res(), "source error")?;
409        let count_sql = format!("select count(*) from {}", table);
410        let count: usize = conn
411            .query_row(count_sql.as_str(), (), |row| row.get(0))
412            .source_raw_err(KnowReason::from_rule(), "source error")?;
413        if count >= scope.0 {
414            Ok(count)
415        } else {
416            Err(KnowReason::from_conf()
417                .to_err()
418                .with_detail("table data less")
419                .with_context(("table", table))
420                .with_context(("count", count.to_string())))
421
422            /*
423            Err(anyhow!(
424                "data less! , load data count {} <= min {}",
425                count,
426                scope.0,
427            ))
428            */
429        }
430    }
431
432    pub fn global() -> Self {
433        Self {
434            conn: MEM_SQLITE_INS.clone(),
435        }
436    }
437    pub fn load_test() -> KnowledgeResult<Self> {
438        let db = Self::global();
439        debug_kdb!("[memdb] load_test invoked");
440        db.table_create(EXAMPLE_CREATE_SQL)?;
441        // 通过 crate 根目录定位测试字典，避免 cwd 影响
442        let csv = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("src/mem/dict/example.csv");
443        let _ = db.table_clean(EXAMPLE_CLEAN_SQL);
444        db.table_load(EXAMPLE_INSERT_SQL, csv, vec![0, 1], 100)?;
445        // quick sanity check
446        if let Ok(cnt) = db.check_data("example", (1, usize::MAX)) {
447            debug_kdb!("[memdb] example rows loaded = {}", cnt);
448        }
449        Ok(db)
450    }
451}
452pub const EXAMPLE_CREATE_SQL: &str = r#"CREATE TABLE IF NOT EXISTS example (
453    id   INTEGER PRIMARY KEY,
454    name TEXT NOT NULL,
455    pinying TEXT NOT NULL
456    )"#;
457pub const EXAMPLE_CLEAN_SQL: &str = "DELETE FROM example";
458pub const EXAMPLE_INSERT_SQL: &str = r#"INSERT INTO example(name,pinying) VALUES (?1, ?2 ) "#;
459
460#[cfg(test)]
461mod tests {
462
463    use std::{fs::File, io::Read};
464
465    use super::*;
466    // V1 TableConf removed
467    use crate::error::{KnowReason, KnowledgeResult};
468    use crate::mem::ToSqlParams;
469
470    use orion_conf::EnvTomlLoad;
471    use orion_error::conversion::SourceErr;
472    use orion_variate::EnvDict;
473    use serde::Serialize;
474    use std::fs;
475    use wp_data_fmt::{Csv, RecordFormatter};
476
477    #[test]
478    fn test_load() -> KnowledgeResult<()> {
479        let db = MemDB::instance();
480        db.table_create(EXAMPLE_CREATE_SQL)?;
481        let loaded = db.table_load(
482            EXAMPLE_INSERT_SQL,
483            PathBuf::from("src/mem/dict/example.csv"),
484            vec![0, 1],
485            100,
486        )?;
487        assert_eq!(loaded, 10);
488        let fmt = Csv::default();
489        let tdos = db.query_row("select * from example;")?;
490        for obj in tdos {
491            println!("{}", fmt.fmt_field(&obj.into()));
492        }
493        Ok(())
494    }
495
496    #[test]
497    fn test_csv_off_by_one() -> KnowledgeResult<()> {
498        let db = MemDB::instance();
499        db.table_create(EXAMPLE_CREATE_SQL)?;
500        // Expect only 1 row loaded when max=1 (no off-by-one)
501        let loaded = db.table_load(
502            EXAMPLE_INSERT_SQL,
503            PathBuf::from("src/mem/dict/example.csv"),
504            vec![0, 1],
505            1,
506        )?;
507        assert_eq!(loaded, 1);
508        Ok(())
509    }
510
511    #[test]
512    fn test_row_null_mapping() -> KnowledgeResult<()> {
513        let db = MemDB::instance();
514        db.execute("CREATE TABLE tnull (v TEXT)")?;
515        db.execute("INSERT INTO tnull (v) VALUES (NULL)")?;
516        let row = db.query_row("SELECT v FROM tnull")?;
517        assert_eq!(row.len(), 1);
518        assert_eq!(row[0].get_name(), "v");
519        // Ensure NULL becomes a Value::Null rather than panic
520        assert!(matches!(row[0].get_value(), model::Value::Null));
521        Ok(())
522    }
523
524    #[test]
525    fn test_row_blob_mapping() -> KnowledgeResult<()> {
526        let db = MemDB::instance();
527        db.execute("CREATE TABLE tblob (b BLOB)")?;
528        // Insert ASCII 'ABC' as blob
529        db.execute("INSERT INTO tblob (b) VALUES (X'414243')")?;
530        let row = db.query_row("SELECT b FROM tblob")?;
531        assert_eq!(row.len(), 1);
532        assert_eq!(row[0].get_name(), "b");
533        // lossy utf8 decode should yield "ABC"
534        assert_eq!(row[0].to_string(), "chars(ABC)");
535        Ok(())
536    }
537
538    #[test]
539    fn test_csv_missing_column_error() -> KnowledgeResult<()> {
540        use std::fs;
541        use std::io::Write;
542        let db = MemDB::instance();
543        db.table_create(EXAMPLE_CREATE_SQL)?;
544        // Create a temp csv with only 1 column per row
545        let mut path = std::env::temp_dir();
546        path.push("wp_knowledge_csv_missing_col.csv");
547        {
548            let mut f =
549                fs::File::create(&path).source_raw_err(KnowReason::from_res(), "source error")?;
550            writeln!(f, "name").source_raw_err(KnowReason::from_res(), "source error")?;
551            writeln!(f, "only_one_col").source_raw_err(KnowReason::from_res(), "source error")?;
552        }
553        let res = db.table_load(
554            EXAMPLE_INSERT_SQL,
555            path.clone(),
556            vec![0, 1], // request 2 columns but csv has 1
557            10,
558        );
559        assert!(res.is_err());
560        let e = format!("{}", res.err().unwrap());
561        assert!(e.contains("line"));
562        assert!(e.contains("insufficient columns"));
563        // cleanup
564        let _ = fs::remove_file(&path);
565        Ok(())
566    }
567
568    #[test]
569    fn test_global_persistence_across_handles() -> KnowledgeResult<()> {
570        // Create table via one global handle
571        {
572            let db1 = MemDB::global();
573            db1.execute("CREATE TABLE IF NOT EXISTS gtest (v TEXT)")?;
574            db1.execute("INSERT INTO gtest (v) VALUES ('ok')")?;
575        }
576        // Read via a new global handle; should see the same in-memory DB
577        {
578            let db2 = MemDB::global();
579            let rows = db2.query_row("SELECT v FROM gtest")?;
580            assert_eq!(rows.len(), 1);
581            assert_eq!(rows[0].to_string(), "chars(ok)");
582        }
583        Ok(())
584    }
585
586    #[test]
587    fn test_init_by_conf() -> KnowledgeResult<()> {
588        let db = MemDB::global();
589        db.table_create(EXAMPLE_CREATE_SQL)?;
590        let _ = db.table_clean(EXAMPLE_CLEAN_SQL);
591        db.table_load(
592            EXAMPLE_INSERT_SQL,
593            PathBuf::from("src/mem/dict/example.csv"),
594            vec![0, 1],
595            100,
596        )?;
597        Ok(())
598    }
599
600    // V1 conf serde test removed
601
602    #[test]
603    fn test_alter_level() -> KnowledgeResult<()> {
604        let db = MemDB::global();
605        // ensure clean state across global in-memory handle
606        let _ = db.execute("DROP TABLE IF EXISTS alert_cat_level");
607        db.table_create(
608            r#"CREATE TABLE IF NOT EXISTS alert_cat_level (
609                id   INTEGER PRIMARY KEY,
610                log_type TEXT NOT NULL,
611                level1_code TEXT NOT NULL,
612                level1_name TEXT NOT NULL,
613                level2_code TEXT NOT NULL,
614                level2_name TEXT NOT NULL,
615                original_code TEXT NOT NULL,
616                original_name TEXT NOT NULL
617            )"#,
618        )?;
619        let _ = db.table_clean("DELETE FROM alert_cat_level");
620        db.table_load(
621            r#"INSERT INTO alert_cat_level (log_type, level1_code, level1_name, level2_code, level2_name, original_code, original_name) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)"#,
622            PathBuf::from("src/mem/dict/event_cat_level.csv"),
623            vec![0, 1, 2, 3, 4, 5, 6],
624            2000,
625        )?;
626
627        let sql = "select level1_code from alert_cat_level where log_type = :log_type  and  original_code = :code ";
628        let result = db.query_row_params(
629            //"select level1_code from alert_cat_level where log_type = 'jowto_server_alert_log' and original_code = '00000002'",
630            sql,
631            &[(":log_type", "app_log"), (":code", "00000002")],
632        )?;
633        assert_eq!(result, vec![DataField::from_chars("level1_code", "105")]);
634
635        let px = [
636            SqlNamedParam(DataField::from_chars(":code", "00000002")),
637            SqlNamedParam(DataField::from_chars(":log_type", "app_log")),
638        ];
639
640        let p = px.to_params();
641        let result = db.query_row_params(sql, &p)?;
642        assert_eq!(result, vec![DataField::from_chars("level1_code", "105")]);
643
644        Ok(())
645    }
646
647    #[test]
648    fn test_tosql_bind_various_types() -> KnowledgeResult<()> {
649        use chrono::NaiveDate;
650        use std::net::{IpAddr, Ipv4Addr};
651        use wp_model_core::model::types::value::ObjectValue;
652        use wp_model_core::model::{DateTimeValue, HexT};
653
654        let db = MemDB::instance();
655        db.execute("CREATE TABLE p (v)")?;
656
657        // Bool -> integer 1
658        {
659            let sql = "INSERT INTO p (v) VALUES (:v)";
660            let p = [SqlNamedParam(DataField::from_bool(":v", true))];
661            db.query_row_params(sql, &p.to_params())?;
662            let row = db.query_row("SELECT v FROM p ORDER BY rowid DESC LIMIT 1")?;
663            assert!(matches!(row[0].get_value(), model::Value::Digit(1)));
664        }
665        // Null
666        {
667            let sql = "INSERT INTO p (v) VALUES (:v)";
668            let p = [SqlNamedParam(DataField::new(
669                model::DataType::default(),
670                ":v",
671                model::Value::Null,
672            ))];
673            db.query_row_params(sql, &p.to_params())?;
674            let row = db.query_row("SELECT v FROM p ORDER BY rowid DESC LIMIT 1")?;
675            assert!(matches!(row[0].get_value(), model::Value::Null));
676        }
677        // Time -> text
678        {
679            let dt: DateTimeValue = NaiveDate::from_ymd_opt(2023, 1, 1)
680                .unwrap()
681                .and_hms_opt(0, 0, 0)
682                .unwrap();
683            let sql = "INSERT INTO p (v) VALUES (:v)";
684            let p = [SqlNamedParam(DataField::from_time(":v", dt))];
685            db.query_row_params(sql, &p.to_params())?;
686            let row = db.query_row("SELECT v FROM p ORDER BY rowid DESC LIMIT 1")?;
687            assert!(matches!(row[0].get_value(), model::Value::Chars(_)));
688        }
689        // Hex -> text
690        {
691            let sql = "INSERT INTO p (v) VALUES (:v)";
692            let p = [SqlNamedParam(DataField::from_hex(":v", HexT(0xABCD)))];
693            db.query_row_params(sql, &p.to_params())?;
694            let row = db.query_row("SELECT v FROM p ORDER BY rowid DESC LIMIT 1")?;
695            assert!(matches!(row[0].get_value(), model::Value::Chars(_)));
696        }
697        // IpAddr -> text
698        {
699            let sql = "INSERT INTO p (v) VALUES (:v)";
700            let p = [SqlNamedParam(DataField::from_ip(
701                ":v",
702                IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)),
703            ))];
704            db.query_row_params(sql, &p.to_params())?;
705            let row = db.query_row("SELECT v FROM p ORDER BY rowid DESC LIMIT 1")?;
706            assert_eq!(row[0].to_string(), "chars(1.2.3.4)");
707        }
708        // Obj -> text (debug)
709        {
710            let mut obj = ObjectValue::new();
711            obj.insert("k".to_string(), DataField::from_chars("", "v"));
712            let sql = "INSERT INTO p (v) VALUES (:v)";
713            let p = [SqlNamedParam(DataField::from_obj(":v", obj))];
714            db.query_row_params(sql, &p.to_params())?;
715            let row = db.query_row("SELECT v FROM p ORDER BY rowid DESC LIMIT 1")?;
716            assert!(matches!(row[0].get_value(), model::Value::Chars(_)));
717        }
718        // Array -> text (debug)
719        {
720            let arr = vec![DataField::from_chars("", "a"), DataField::from_digit("", 1)];
721            let sql = "INSERT INTO p (v) VALUES (:v)";
722            let p = [SqlNamedParam(DataField::from_arr(":v", arr))];
723            db.query_row_params(sql, &p.to_params())?;
724            let row = db.query_row("SELECT v FROM p ORDER BY rowid DESC LIMIT 1")?;
725            assert!(matches!(row[0].get_value(), model::Value::Chars(_)));
726        }
727        // BigUint -> 十进制文本（超出 i64 范围，SQLite 按 affinity 数值比较）
728        {
729            use num_bigint::BigUint;
730            use std::str::FromStr;
731            let sql = "INSERT INTO p (v) VALUES (:v)";
732            let p = [SqlNamedParam(DataField::new(
733                model::DataType::BigInt,
734                ":v",
735                model::Value::BigUint(
736                    BigUint::from_str("382824323044708348099391746388336347272").unwrap(),
737                ),
738            ))];
739            db.query_row_params(sql, &p.to_params())?;
740            let row = db.query_row("SELECT v FROM p ORDER BY rowid DESC LIMIT 1")?;
741            assert!(matches!(row[0].get_value(), model::Value::Chars(_)));
742            assert_eq!(
743                row[0].get_chars(),
744                Some("382824323044708348099391746388336347272")
745            );
746        }
747        Ok(())
748    }
749
750    #[test]
751    fn test_column_alias_names() -> KnowledgeResult<()> {
752        let db = MemDB::instance();
753        // Create a simple one-shot table/view using alias
754        db.execute("CREATE TABLE ctest (a INTEGER, b TEXT)")?;
755        db.execute("INSERT INTO ctest (a,b) VALUES (42,'x')")?;
756        let row = db.query_row("SELECT a AS 'the number', b AS 'the text' FROM ctest LIMIT 1")?;
757        assert_eq!(row.len(), 2);
758        assert_eq!(row[0].get_name(), "the number");
759        assert_eq!(row[1].get_name(), "the text");
760        Ok(())
761    }
762
763    #[test]
764    fn test_concurrent_inserts() -> KnowledgeResult<()> {
765        use std::thread;
766        let db = MemDB::global();
767        db.execute("CREATE TABLE IF NOT EXISTS concur (v INTEGER)")?;
768        let threads: Vec<_> = (0..4)
769            .map(|_| {
770                thread::spawn(|| {
771                    let dbt = MemDB::global();
772                    for _ in 0..10 {
773                        let _ = dbt.execute("INSERT INTO concur (v) VALUES (1)");
774                    }
775                })
776            })
777            .collect();
778        for t in threads {
779            t.join().unwrap();
780        }
781        let row = db.query_row("SELECT SUM(v) AS total FROM concur")?;
782        // total should be 40
783        assert_eq!(row[0].to_string(), "digit(40)");
784        Ok(())
785    }
786
787    #[test]
788    fn test_query_returns_all_rows() -> KnowledgeResult<()> {
789        let db = MemDB::instance();
790        db.execute("CREATE TABLE multi (id INTEGER, name TEXT)")?;
791        let rows = db.query("SELECT * FROM multi")?;
792        assert!(rows.is_empty(), "empty table should return empty vec");
793        db.execute("INSERT INTO multi (id, name) VALUES (1, 'alice')")?;
794        db.execute("INSERT INTO multi (id, name) VALUES (2, 'bob')")?;
795        db.execute("INSERT INTO multi (id, name) VALUES (3, 'charlie')")?;
796
797        let rows = db.query("SELECT id, name FROM multi ORDER BY id")?;
798        assert_eq!(rows.len(), 3, "should return all 3 rows");
799
800        Ok(())
801    }
802
803    #[allow(dead_code)]
804    fn load_toml_conf<T: serde::de::DeserializeOwned>(path: &str) -> KnowledgeResult<T> {
805        let mut f = File::open(path)
806            .source_raw_err(KnowReason::from_res(), "source error")
807            .doing(format!("conf file not found: {}", path))?;
808        let mut buffer = Vec::with_capacity(10240);
809        f.read_to_end(&mut buffer)
810            .source_raw_err(KnowReason::from_res(), "source error")?;
811        let conf_data =
812            String::from_utf8(buffer).source_raw_err(KnowReason::from_rule(), "source error")?;
813        let dict = EnvDict::new();
814        let conf: T = T::env_parse_toml(conf_data.as_str(), &dict)
815            .source_err(KnowReason::from_conf(), "parse toml config")?;
816        Ok(conf)
817    }
818
819    #[allow(dead_code)]
820    fn export_toml_local<T: Serialize>(val: &T, path: &str) -> KnowledgeResult<()> {
821        let data =
822            toml::to_string_pretty(val).source_raw_err(KnowReason::from_rule(), "source error")?;
823        if let Some(parent) = std::path::Path::new(path).parent() {
824            fs::create_dir_all(parent).source_raw_err(KnowReason::from_res(), "source error")?;
825        }
826        fs::write(path, data).source_raw_err(KnowReason::from_res(), "source error")?;
827        Ok(())
828    }
829}
wp_knowledge/mem/memdb.rs

wp_knowledge/mem/
memdb.rs
