inklog/integrations/infra/
database.rs

1// Copyright (c) 2026 Kirky.X
2// SPDX-License-Identifier: MIT
3//! Database trait - 抽象数据库操作
4//!
5//! 提供日志记录批量写入和健康检查的抽象接口。
6
7use std::sync::Arc;
8
9use crate::InklogError;
10use crate::LogRecord;
11use async_trait::async_trait;
12
13/// Database trait - 抽象数据库操作
14///
15/// 提供日志记录批量写入和健康检查接口。
16/// 实现必须保证线程安全（`Send + Sync`）。
17///
18/// # 实现要求
19///
20/// - 所有方法使用 `&self`（不可变引用），支持并发访问
21/// - 批量插入应该是原子操作（全部成功或全部失败）
22/// - 健康检查应该是轻量级的
23///
24/// # 示例
25///
26/// ```ignore
27/// use inklog::infrastructure::Database;
28/// use inklog::log_record::LogRecord;
29/// use tracing::Level;
30///
31/// async fn example(db: &dyn Database) {
32///     let records = vec![
33///         LogRecord::new(Level::INFO, "module".to_string(), "message".to_string()),
34///     ];
35///     
36///     let count = db.insert_batch(&records).await.unwrap();
37///     assert_eq!(count, 1);
38///     
39///     if db.is_healthy().await {
40///         println!("Database is healthy");
41///     }
42/// }
43/// ```
44#[async_trait]
45pub trait Database: Send + Sync {
46    /// 批量插入日志记录
47    ///
48    /// # 参数
49    ///
50    /// * `records` - 日志记录切片
51    ///
52    /// # 返回
53    ///
54    /// 成功返回成功插入的记录数 `Ok(count)`，失败返回 `Err(InklogError)`
55    ///
56    /// # 注意
57    ///
58    /// 实现应该保证原子性，要么全部插入成功，要么全部失败
59    async fn insert_batch(&self, records: &[LogRecord]) -> Result<usize, InklogError>;
60
61    /// 检查数据库健康状态
62    ///
63    /// # 返回
64    ///
65    /// 数据库连接正常返回 `true`，否则返回 `false`
66    ///
67    /// # 注意
68    ///
69    /// 此方法应该是轻量级的，适合频繁调用
70    async fn is_healthy(&self) -> bool;
71}
72
73// ============================================================================
74// DbNexusAdapter - dbnexus 适配器实现 (条件编译)
75// ============================================================================
76
77#[cfg(any(
78    feature = "sqlite",
79    feature = "postgres",
80    feature = "mysql",
81    feature = "duckdb"
82))]
83use dbnexus::ConnectionPool;
84#[cfg(any(
85    feature = "sqlite",
86    feature = "postgres",
87    feature = "mysql",
88    feature = "duckdb"
89))]
90use dbnexus::database::pool::DbPool;
91#[cfg(any(
92    feature = "sqlite",
93    feature = "postgres",
94    feature = "mysql",
95    feature = "duckdb"
96))]
97use dbnexus::foundation::config::DbConfig;
98
99#[cfg(any(
100    feature = "sqlite",
101    feature = "postgres",
102    feature = "mysql",
103    feature = "duckdb"
104))]
105use crate::domain::config::database::DatabaseDriver;
106
107/// dbnexus 适配器
108///
109/// 将 dbnexus 库的 `DbPool` 适配为 `Database` trait。
110/// 使用 Sea-ORM 进行批量插入操作。
111///
112/// # 功能要求
113///
114/// - 需要启用 `dbnexus` feature
115/// - 支持 PostgreSQL、MySQL、SQLite、DuckDB 数据库
116///
117/// # 示例
118///
119/// ```ignore
120/// use inklog::infrastructure::database::{Database, DbNexusAdapter};
121///
122/// #[tokio::main]
123/// async fn main() -> Result<(), Box<dyn std::error::Error>> {
124///     let db = DbNexusAdapter::new("postgres://user:pass@localhost/logs", 10).await?;
125///     
126///     let healthy = db.is_healthy().await;
127///     println!("Database healthy: {}", healthy);
128///     
129///     Ok(())
130/// }
131/// ```
132#[cfg(any(
133    feature = "sqlite",
134    feature = "postgres",
135    feature = "mysql",
136    feature = "duckdb"
137))]
138pub struct DbNexusAdapter {
139    pool: Arc<dyn ConnectionPool + Send + Sync>,
140    table_name: String,
141    admin_role: String,
142}
143
144#[cfg(any(
145    feature = "sqlite",
146    feature = "postgres",
147    feature = "mysql",
148    feature = "duckdb"
149))]
150impl DbNexusAdapter {
151    /// 创建新的 dbnexus 适配器
152    ///
153    /// # 参数
154    ///
155    /// * `url` - 数据库连接字符串
156    /// * `pool_size` - 连接池大小（最大连接数）
157    ///
158    /// # 返回
159    ///
160    /// 成功返回 `Ok(Self)`，失败返回 `Err(InklogError)`
161    ///
162    /// # 错误
163    ///
164    /// - `InklogError::DatabaseError` - 连接池创建失败
165    ///
166    /// # 示例
167    ///
168    /// ```ignore
169    /// // PostgreSQL
170    /// let db = DbNexusAdapter::new("postgres://user:pass@localhost/logs", 10).await?;
171    ///
172    /// // MySQL
173    /// let db = DbNexusAdapter::new("mysql://user:pass@localhost/logs", 10).await?;
174    ///
175    /// // SQLite
176    /// let db = DbNexusAdapter::new("sqlite://logs.db", 1).await?;
177    /// ```
178    pub async fn new(url: &str, pool_size: u32) -> Result<Self, InklogError> {
179        Self::with_table_name(url, pool_size, crate::support::io::sink::entity::TABLE_NAME).await
180    }
181
182    /// 创建带有自定义表名的适配器
183    ///
184    /// # 参数
185    ///
186    /// * `url` - 数据库连接字符串
187    /// * `pool_size` - 连接池大小（最大连接数）
188    /// * `table_name` - 日志表名称
189    pub async fn with_table_name(
190        url: &str,
191        pool_size: u32,
192        table_name: &str,
193    ) -> Result<Self, InklogError> {
194        Self::with_full_config(url, pool_size, table_name, None, "admin").await
195    }
196
197    /// 创建带完整配置的适配器
198    ///
199    /// 支持自定义权限配置文件路径和管理员角色名。
200    /// 构造完成后自动调用 `ensure_table_exists()` 创建日志表。
201    ///
202    /// # 参数
203    ///
204    /// * `url` - 数据库连接字符串
205    /// * `pool_size` - 连接池大小（最大连接数）
206    /// * `table_name` - 日志表名称
207    /// * `permissions_path` - 权限配置文件路径，`None` 时不启用权限校验
208    /// * `admin_role` - 管理员角色名
209    pub async fn with_full_config(
210        url: &str,
211        pool_size: u32,
212        table_name: &str,
213        permissions_path: Option<String>,
214        admin_role: &str,
215    ) -> Result<Self, InklogError> {
216        validate_table_name(table_name)?;
217
218        // 创建 DbConfig
219        let config = DbConfig {
220            url: url.to_string(),
221            pool_config: dbnexus::foundation::config::PoolConfig {
222                max_connections: pool_size,
223                min_connections: 1,
224                idle_timeout: 300,
225                acquire_timeout: 5000,
226            },
227            permissions_path,
228            migrations_dir: None,
229            auto_migrate: false,
230            migration_timeout: 60,
231            admin_role: admin_role.to_string(),
232            warmup_timeout: 30,
233            warmup_retries: 3,
234            cache_config: dbnexus::foundation::config::CacheConfig::default(),
235            retry_policy: Some(dbnexus::reliability::retry::RetryPolicy::default()),
236        };
237
238        // 使用 DbPool::with_config 创建连接池
239        let pool = DbPool::with_config(config).await.map_err(|e| {
240            let mut args = fluent_bundle::FluentArgs::new();
241            args.set("err", e.to_string());
242            InklogError::DatabaseError {
243                message: crate::i18n::tr_args("db-pool_create_failed", args),
244                source: Some(Box::new(e)),
245            }
246        })?;
247
248        let adapter = Self {
249            pool: Arc::new(pool),
250            table_name: table_name.to_string(),
251            admin_role: admin_role.to_string(),
252        };
253
254        // 自动建表
255        adapter.ensure_table_exists(url).await?;
256
257        Ok(adapter)
258    }
259
260    /// 从现有 DbPool 创建适配器
261    ///
262    /// 用于需要共享连接池的场景。
263    ///
264    /// # 参数
265    ///
266    /// * `pool` - 已创建的连接池实例
267    /// * `table_name` - 日志表名称
268    pub fn from_pool(pool: DbPool, table_name: &str) -> Result<Self, InklogError> {
269        validate_table_name(table_name)?;
270        Ok(Self {
271            pool: Arc::new(pool),
272            table_name: table_name.to_string(),
273            admin_role: "admin".to_string(),
274        })
275    }
276
277    /// 从已有的 `ConnectionPool` trait 对象创建适配器
278    ///
279    /// 用于 trait-kit DI 场景：kit 提供 `Arc<dyn ConnectionPool + Send + Sync>`，
280    /// 直接包装为 `Database` trait 实现，无需重新创建连接池。
281    ///
282    /// # 参数
283    ///
284    /// * `pool` - dbnexus 连接池 trait 对象（通常来自 `kit.require::<DbNexusModule>()`）
285    /// * `table_name` - 日志表名称
286    pub fn from_connection_pool(
287        pool: Arc<dyn ConnectionPool + Send + Sync>,
288        table_name: &str,
289    ) -> Result<Self, InklogError> {
290        validate_table_name(table_name)?;
291        Ok(Self {
292            pool,
293            table_name: table_name.to_string(),
294            admin_role: "admin".to_string(),
295        })
296    }
297
298    /// 获取底层连接池引用
299    pub fn pool(&self) -> &dyn ConnectionPool {
300        self.pool.as_ref()
301    }
302
303    /// 获取底层连接池的 `Arc` 克隆
304    ///
305    /// 用于需要共享连接池创建新适配器的场景（见 [`Self::from_connection_pool`]）。
306    pub fn pool_arc(&self) -> Arc<dyn ConnectionPool + Send + Sync> {
307        Arc::clone(&self.pool)
308    }
309
310    /// 获取表名
311    pub fn table_name(&self) -> &str {
312        &self.table_name
313    }
314
315    /// 确保日志表存在，若不存在则自动创建。
316    ///
317    /// 使用管理员角色获取 session 并执行 DDL。
318    /// 驱动类型通过 URL 前缀自动识别。
319    async fn ensure_table_exists(&self, url: &str) -> Result<(), InklogError> {
320        let driver = detect_driver_from_url(url);
321        let ddl = generate_create_table_sql(&self.table_name, &driver);
322        let session = self.pool.get_session(&self.admin_role).await.map_err(|e| {
323            let mut args = fluent_bundle::FluentArgs::new();
324            args.set("err", e.to_string());
325            InklogError::DatabaseError {
326                message: crate::i18n::tr_args("db-session_failed", args),
327                source: Some(Box::new(e)),
328            }
329        })?;
330        session.execute_raw_ddl(&ddl).await.map_err(|e| {
331            let mut args = fluent_bundle::FluentArgs::new();
332            args.set("err", e.to_string());
333            InklogError::DatabaseError {
334                message: crate::i18n::tr_args("db-ensure_table_failed", args),
335                source: Some(Box::new(e)),
336            }
337        })?;
338        Ok(())
339    }
340}
341
342/// Validate that a table name contains only safe identifier characters.
343///
344/// Rejects names that don't match `^[a-zA-Z_][a-zA-Z0-9_]*$` to prevent SQL injection
345/// via table name interpolation.
346#[cfg(any(
347    feature = "sqlite",
348    feature = "postgres",
349    feature = "mysql",
350    feature = "duckdb"
351))]
352fn validate_table_name(name: &str) -> Result<(), InklogError> {
353    if name.is_empty() {
354        return Err(InklogError::ConfigError(crate::i18n::tr("db-table_empty")));
355    }
356    let mut chars = name.chars();
357    let first = chars.next().unwrap();
358    if !first.is_ascii_alphabetic() && first != '_' {
359        let mut args = fluent_bundle::FluentArgs::new();
360        args.set("name", name.to_string());
361        return Err(InklogError::ConfigError(crate::i18n::tr_args(
362            "db-table_invalid_start",
363            args,
364        )));
365    }
366    for c in chars {
367        if !c.is_ascii_alphanumeric() && c != '_' {
368            let mut args = fluent_bundle::FluentArgs::new();
369            args.set("name", name.to_string());
370            args.set("char", c.to_string());
371            return Err(InklogError::ConfigError(crate::i18n::tr_args(
372                "db-table_invalid_char",
373                args,
374            )));
375        }
376    }
377    Ok(())
378}
379
380/// 转义 SQL 字符串中的单引号，防止 SQL 注入。
381///
382/// 所有通过 `insert_batch` 写入的字符串字段必须经过此函数。
383/// 采用标准 SQL 转义规则：将单引号 `'` 替换为双单引号 `''`。
384#[cfg(any(
385    feature = "sqlite",
386    feature = "postgres",
387    feature = "mysql",
388    feature = "duckdb"
389))]
390#[inline]
391fn escape_sql_string(s: &str) -> String {
392    s.replace('\'', "''")
393}
394
395/// 根据数据库驱动生成 CREATE TABLE DDL 语句。
396///
397/// 不同后端使用不同的数据类型：
398/// - SQLite: `INTEGER PRIMARY KEY AUTOINCREMENT`, `TEXT`
399/// - PostgreSQL: `BIGSERIAL PRIMARY KEY`, `TIMESTAMPTZ`, `TEXT`
400/// - MySQL: `BIGINT AUTO_INCREMENT PRIMARY KEY`, `TIMESTAMP`, `TEXT`
401/// - DuckDB: `BIGINT AUTOINCREMENT PRIMARY KEY`, `TIMESTAMP`, `TEXT`
402#[cfg(any(
403    feature = "sqlite",
404    feature = "postgres",
405    feature = "mysql",
406    feature = "duckdb"
407))]
408fn generate_create_table_sql(table_name: &str, driver: &DatabaseDriver) -> String {
409    match driver {
410        DatabaseDriver::SQLite => format!(
411            "CREATE TABLE IF NOT EXISTS {} (\
412                id INTEGER PRIMARY KEY AUTOINCREMENT, \
413                timestamp TEXT NOT NULL, \
414                level TEXT NOT NULL, \
415                target TEXT NOT NULL, \
416                message TEXT NOT NULL, \
417                fields TEXT, \
418                file TEXT, \
419                line INTEGER, \
420                thread_id TEXT NOT NULL\
421            )",
422            table_name
423        ),
424        DatabaseDriver::PostgreSQL => format!(
425            "CREATE TABLE IF NOT EXISTS {} (\
426                id BIGSERIAL PRIMARY KEY, \
427                timestamp TIMESTAMPTZ NOT NULL, \
428                level TEXT NOT NULL, \
429                target TEXT NOT NULL, \
430                message TEXT NOT NULL, \
431                fields TEXT, \
432                file TEXT, \
433                line INTEGER, \
434                thread_id TEXT NOT NULL\
435            )",
436            table_name
437        ),
438        DatabaseDriver::MySQL => format!(
439            "CREATE TABLE IF NOT EXISTS {} (\
440                id BIGINT AUTO_INCREMENT PRIMARY KEY, \
441                timestamp TIMESTAMP NOT NULL, \
442                level TEXT NOT NULL, \
443                target TEXT NOT NULL, \
444                message TEXT NOT NULL, \
445                fields TEXT, \
446                file TEXT, \
447                line INTEGER, \
448                thread_id TEXT NOT NULL\
449            )",
450            table_name
451        ),
452        DatabaseDriver::DuckDB => format!(
453            "CREATE TABLE IF NOT EXISTS {} (\
454                id BIGINT AUTOINCREMENT PRIMARY KEY, \
455                timestamp TIMESTAMP NOT NULL, \
456                level TEXT NOT NULL, \
457                target TEXT NOT NULL, \
458                message TEXT NOT NULL, \
459                fields TEXT, \
460                file TEXT, \
461                line INTEGER, \
462                thread_id TEXT NOT NULL\
463            )",
464            table_name
465        ),
466    }
467}
468
469/// 从数据库 URL 推断驱动类型。
470#[cfg(any(
471    feature = "sqlite",
472    feature = "postgres",
473    feature = "mysql",
474    feature = "duckdb"
475))]
476fn detect_driver_from_url(url: &str) -> DatabaseDriver {
477    if url.starts_with("sqlite:") || url.starts_with("sqlite3:") {
478        DatabaseDriver::SQLite
479    } else if url.starts_with("postgres:") || url.starts_with("postgresql:") {
480        DatabaseDriver::PostgreSQL
481    } else if url.starts_with("mysql:") {
482        DatabaseDriver::MySQL
483    } else if url.starts_with("duckdb:") || url.starts_with("duckdb://") {
484        DatabaseDriver::DuckDB
485    } else {
486        // 默认回退到 PostgreSQL
487        DatabaseDriver::PostgreSQL
488    }
489}
490
491#[cfg(any(
492    feature = "sqlite",
493    feature = "postgres",
494    feature = "mysql",
495    feature = "duckdb"
496))]
497#[async_trait]
498impl Database for DbNexusAdapter {
499    async fn insert_batch(&self, records: &[LogRecord]) -> Result<usize, InklogError> {
500        if records.is_empty() {
501            return Ok(0);
502        }
503
504        let start = std::time::Instant::now();
505
506        // 获取写会话 (使用配置的管理员角色)
507        let session = self.pool.get_session(&self.admin_role).await.map_err(|e| {
508            let mut args = fluent_bundle::FluentArgs::new();
509            args.set("err", e.to_string());
510            InklogError::DatabaseError {
511                message: crate::i18n::tr_args("db-session_failed", args),
512                source: Some(Box::new(e)),
513            }
514        })?;
515
516        // 构建所有记录的 INSERT SQL 语句
517        let sqls: Vec<String> = records
518            .iter()
519            .map(|record| {
520                let timestamp = record.timestamp.to_rfc3339();
521                let level = escape_sql_string(&record.level);
522                let target = escape_sql_string(&record.target);
523                let message = escape_sql_string(&record.message);
524                let fields_json =
525                    serde_json::to_string(&record.fields).unwrap_or_else(|_| "{}".to_string());
526                let fields_escaped = escape_sql_string(&fields_json);
527                let file = record
528                    .file
529                    .as_ref()
530                    .map(|f| format!("'{}'", escape_sql_string(f)))
531                    .unwrap_or_else(|| "NULL".to_string());
532                let line = record
533                    .line
534                    .map(|l| l.to_string())
535                    .unwrap_or_else(|| "NULL".to_string());
536                let thread_id = escape_sql_string(&record.thread_id);
537
538                format!(
539                    "INSERT INTO {} (timestamp, level, target, message, fields, file, line, thread_id) \
540                     VALUES ('{}', '{}', '{}', '{}', '{}', {}, {}, '{}')",
541                    self.table_name,
542                    timestamp,
543                    level,
544                    target,
545                    message,
546                    fields_escaped,
547                    file,
548                    line,
549                    thread_id
550                )
551            })
552            .collect();
553
554        // 在事务中执行全部语句——原子性：全部成功或全部失败
555        let sql_refs: Vec<&str> = sqls.iter().map(|s| s.as_str()).collect();
556        session
557            .batch_execute_in_transaction(sql_refs)
558            .await
559            .map_err(|e| {
560                let elapsed_ms = start.elapsed().as_millis();
561                tracing::warn!(
562                    table = %self.table_name,
563                    error = %e,
564                    elapsed_ms = elapsed_ms,
565                    "Database batch insert failed"
566                );
567                let mut args = fluent_bundle::FluentArgs::new();
568                args.set("err", e.to_string());
569                let msg = crate::i18n::tr_args("db-batch_insert_failed", args);
570                InklogError::DatabaseError {
571                    message: msg,
572                    source: Some(Box::new(e)),
573                }
574            })?;
575
576        let elapsed_ms = start.elapsed().as_millis();
577        tracing::debug!(
578            table = %self.table_name,
579            count = records.len(),
580            elapsed_ms = elapsed_ms,
581            "Database batch insert succeeded"
582        );
583
584        Ok(records.len())
585    }
586
587    async fn is_healthy(&self) -> bool {
588        // 健康检查：仅验证连接池能获取管理员会话。
589        //
590        // 不执行 `SELECT 1` 之类的探针 SQL，原因：
591        // dbnexus 启用 `sql-parser` + `permission` features 后，
592        // `execute_raw` 要求 SQL 必须含表名以做权限校验，
593        // 无表名的 `SELECT 1` 会被 `Permission("SQL statement requires a valid
594        // table name for permission checking")` 拒绝。
595        //
596        // `get_session` 内部调用 `acquire_connection`：
597        // - 空闲队列有连接时直接返回（池已验证过初始可达性，见 `with_config`）
598        // - 空闲队列为空时调用 `create_connection` 重新建立连接，失败则返回 Err
599        //
600        // 因此 `get_session` 成功即可认为数据库当前可达，符合 `is_healthy()`
601        // 文档要求的"轻量级，适合频繁调用"。
602        match self.pool.get_session(&self.admin_role).await {
603            Ok(_) => true,
604            Err(e) => {
605                let mut args = fluent_bundle::FluentArgs::new();
606                args.set("err", e.to_string());
607                tracing::warn!(
608                    "{}",
609                    crate::i18n::tr_args("warn-db_health_check_failed", args)
610                );
611                false
612            }
613        }
614    }
615}
616
617// ============================================================================
618// 非 dbnexus feature 时的占位实现
619// ============================================================================
620
621#[cfg(not(any(
622    feature = "sqlite",
623    feature = "postgres",
624    feature = "mysql",
625    feature = "duckdb"
626)))]
627/// DbNexusAdapter - 仅在启用 `dbnexus` feature 时可用
628///
629/// 当未启用 `dbnexus` feature 时，此类型不存在。
630/// 使用 `MockDatabaseAdapter` 作为测试替代方案。
631pub struct DbNexusAdapter {
632    _phantom: (),
633}
634
635#[cfg(not(any(
636    feature = "sqlite",
637    feature = "postgres",
638    feature = "mysql",
639    feature = "duckdb"
640)))]
641impl DbNexusAdapter {
642    /// 此方法仅在启用 `dbnexus` feature 时可用
643    #[deprecated(note = "Enable 'dbnexus' feature to use DbNexusAdapter")]
644    pub async fn new(_url: &str, _pool_size: u32) -> Result<Self, InklogError> {
645        Err(InklogError::DatabaseError {
646            message: "DbNexusAdapter requires 'dbnexus' feature to be enabled".to_string(),
647            source: None,
648        })
649    }
650}
651
652// ============================================================================
653// MockDatabaseAdapter - 测试用 Mock 实现
654// ============================================================================
655
656use std::sync::RwLock;
657use std::sync::atomic::{AtomicBool, Ordering};
658
659/// Mock 数据库适配器，用于单元测试
660///
661/// 提供内存存储，支持健康状态控制。
662/// 所有操作都在内存中完成，不依赖外部数据库。
663///
664/// # 线程安全
665///
666/// 使用 `RwLock` 保护记录存储，使用 `AtomicBool` 管理健康状态，
667/// 确保多线程环境下的安全性。
668///
669/// # 示例
670///
671/// ```rust
672/// use inklog::integrations::infra::database::{Database, MockDatabaseAdapter};
673/// use inklog::LogRecord;
674/// use tracing::Level;
675///
676/// #[tokio::main]
677/// async fn main() {
678///     let db = MockDatabaseAdapter::new();
679///
680///     // 插入记录
681///     let records = vec![LogRecord::new(
682///         Level::INFO,
683///         "test::module".to_string(),
684///         "Test message".to_string(),
685///     )];
686///     let count = db.insert_batch(&records).await.unwrap();
687///     assert_eq!(count, 1);
688///
689///     // 健康检查
690///     assert!(db.is_healthy().await);
691///
692///     // 模拟故障
693///     db.set_healthy(false);
694///     assert!(!db.is_healthy().await);
695/// }
696/// ```
697pub struct MockDatabaseAdapter {
698    /// 存储的日志记录
699    records: RwLock<Vec<LogRecord>>,
700    /// 健康状态
701    healthy: Arc<AtomicBool>,
702}
703
704impl MockDatabaseAdapter {
705    /// 创建新的 Mock 数据库适配器
706    ///
707    /// 初始化为健康状态（`healthy = true`）。
708    pub fn new() -> Self {
709        Self {
710            records: RwLock::new(Vec::new()),
711            healthy: Arc::new(AtomicBool::new(true)),
712        }
713    }
714
715    /// 设置健康状态
716    ///
717    /// 用于测试中模拟数据库故障和恢复场景。
718    ///
719    /// # 参数
720    ///
721    /// * `healthy` - 新的健康状态
722    pub fn set_healthy(&self, healthy: bool) {
723        self.healthy.store(healthy, Ordering::SeqCst);
724    }
725
726    /// 获取存储的记录数量
727    ///
728    /// 用于测试验证插入操作。
729    pub fn record_count(&self) -> usize {
730        self.records.read().unwrap().len()
731    }
732
733    /// 获取所有存储的记录
734    ///
735    /// 返回记录的克隆，用于测试验证。
736    pub fn get_records(&self) -> Vec<LogRecord> {
737        self.records.read().unwrap().clone()
738    }
739
740    /// 清空所有记录
741    ///
742    /// 用于测试重置状态。
743    pub fn clear(&self) {
744        self.records.write().unwrap().clear();
745    }
746}
747
748impl Default for MockDatabaseAdapter {
749    fn default() -> Self {
750        Self::new()
751    }
752}
753
754#[cfg(test)]
755impl MockDatabaseAdapter {
756    /// Returns the number of records stored (for test verification)
757    pub fn stored_count(&self) -> usize {
758        self.records.read().unwrap().len()
759    }
760}
761
762#[async_trait]
763impl Database for MockDatabaseAdapter {
764    async fn insert_batch(&self, records: &[LogRecord]) -> Result<usize, InklogError> {
765        if records.is_empty() {
766            return Ok(0);
767        }
768
769        let mut stored = self.records.write().unwrap();
770        stored.extend_from_slice(records);
771        Ok(records.len())
772    }
773
774    async fn is_healthy(&self) -> bool {
775        self.healthy.load(Ordering::SeqCst)
776    }
777}
778
779// ============================================================================
780// 测试
781// ============================================================================
782
783#[cfg(test)]
784mod tests {
785    use super::*;
786    use tracing::Level;
787
788    // ============================================================================
789    // DbNexusAdapter 测试 (需要 feature)
790    // ============================================================================
791
792    #[cfg(feature = "sqlite")]
793    #[tokio::test]
794    async fn test_dbnexus_adapter_health_check() {
795        // 创建临时权限配置文件
796        let temp_dir = std::env::temp_dir();
797        let perm_path = temp_dir.join("inklog_health_perm.yaml");
798        let perm_content = r#"roles:
799  admin:
800    tables:
801      - name: "*"
802        operations: ["select", "insert", "update", "delete"]
803"#;
804        std::fs::write(&perm_path, perm_content).expect("Failed to write permissions file");
805
806        // 创建 DbConfig（使用不同的数据库文件）
807        let db_path = temp_dir.join("inklog_health.db");
808        let db_url = format!("sqlite:{}?mode=rwc", db_path.to_string_lossy());
809
810        let config = DbConfig {
811            url: db_url,
812            pool_config: dbnexus::foundation::config::PoolConfig {
813                max_connections: 1,
814                min_connections: 1,
815                idle_timeout: 300,
816                acquire_timeout: 30000,
817            },
818            permissions_path: Some(perm_path.to_string_lossy().to_string()),
819            migrations_dir: None,
820            auto_migrate: false,
821            migration_timeout: 60,
822            admin_role: "admin".to_string(),
823            warmup_timeout: 60,
824            warmup_retries: 5,
825            cache_config: dbnexus::foundation::config::CacheConfig::default(),
826            retry_policy: Some(dbnexus::reliability::retry::RetryPolicy::default()),
827        };
828
829        let pool = DbPool::with_config(config)
830            .await
831            .expect("Failed to create pool");
832        let db = DbNexusAdapter::from_pool(pool, "logs").expect("from_pool should succeed");
833
834        // 创建表用于健康检查
835        let session = db
836            .pool
837            .get_session("admin")
838            .await
839            .expect("Failed to get session");
840        session
841            .execute_raw_ddl(
842                "CREATE TABLE IF NOT EXISTS logs (
843                id INTEGER PRIMARY KEY AUTOINCREMENT,
844                timestamp TEXT NOT NULL,
845                level TEXT NOT NULL,
846                target TEXT NOT NULL,
847                message TEXT NOT NULL,
848                fields TEXT,
849                file TEXT,
850                line INTEGER,
851                thread_id TEXT NOT NULL
852            )",
853            )
854            .await
855            .expect("Failed to create table");
856        drop(session);
857
858        // 直接测试健康检查逻辑 - 使用有效的表名进行查询
859        let session = db
860            .pool
861            .get_session("admin")
862            .await
863            .expect("Failed to get session");
864        let result = session.execute_raw("SELECT COUNT(*) FROM logs").await;
865        assert!(
866            result.is_ok(),
867            "Health check query failed: {:?}",
868            result.err()
869        );
870        drop(session);
871
872        drop(db);
873
874        let _ = std::fs::remove_file(&perm_path);
875        let _ = std::fs::remove_file(&db_path);
876    }
877
878    #[cfg(feature = "sqlite")]
879    #[tokio::test]
880    async fn test_dbnexus_adapter_insert_batch() {
881        // 创建临时权限配置文件
882        let temp_dir = std::env::temp_dir();
883        let perm_path = temp_dir.join("inklog_batch_perm.yaml");
884        let perm_content = r#"roles:
885  admin:
886    tables:
887      - name: "*"
888        operations: ["select", "insert", "update", "delete"]
889"#;
890        std::fs::write(&perm_path, perm_content).expect("Failed to write permissions file");
891
892        // 创建 DbConfig（使用不同的数据库文件）
893        let db_path = temp_dir.join("inklog_batch.db");
894        let db_url = format!("sqlite:{}?mode=rwc", db_path.to_string_lossy());
895
896        let config = DbConfig {
897            url: db_url,
898            pool_config: dbnexus::foundation::config::PoolConfig {
899                max_connections: 2,
900                min_connections: 1,
901                idle_timeout: 300,
902                acquire_timeout: 30000,
903            },
904            permissions_path: Some(perm_path.to_string_lossy().to_string()),
905            migrations_dir: None,
906            auto_migrate: false,
907            migration_timeout: 60,
908            admin_role: "admin".to_string(),
909            warmup_timeout: 60,
910            warmup_retries: 5,
911            cache_config: dbnexus::foundation::config::CacheConfig::default(),
912            retry_policy: Some(dbnexus::reliability::retry::RetryPolicy::default()),
913        };
914
915        let pool = DbPool::with_config(config)
916            .await
917            .expect("Failed to create pool");
918        let db = DbNexusAdapter::from_pool(pool, "logs").expect("from_pool should succeed");
919
920        // 创建 logs 表
921        let session = db
922            .pool
923            .get_session("admin")
924            .await
925            .expect("Failed to get session");
926        let create_result = session
927            .execute_raw_ddl(
928                "CREATE TABLE IF NOT EXISTS logs (
929                id INTEGER PRIMARY KEY AUTOINCREMENT,
930                timestamp TEXT NOT NULL,
931                level TEXT NOT NULL,
932                target TEXT NOT NULL,
933                message TEXT NOT NULL,
934                fields TEXT,
935                file TEXT,
936                line INTEGER,
937                thread_id TEXT NOT NULL
938            )",
939            )
940            .await;
941        assert!(
942            create_result.is_ok(),
943            "Failed to create table: {:?}",
944            create_result.err()
945        );
946        drop(session);
947
948        let records = vec![LogRecord::new(
949            tracing::Level::INFO,
950            "test::module".to_string(),
951            "Test message".to_string(),
952        )];
953
954        let count = db.insert_batch(&records).await.expect("Failed to insert");
955        assert_eq!(count, 1);
956
957        drop(db);
958
959        let _ = std::fs::remove_file(&perm_path);
960        let _ = std::fs::remove_file(&db_path);
961    }
962
963    #[cfg(not(any(
964        feature = "sqlite",
965        feature = "postgres",
966        feature = "mysql",
967        feature = "duckdb"
968    )))]
969    #[allow(deprecated)]
970    #[tokio::test]
971    async fn test_dbnexus_adapter_not_available_without_feature() {
972        let result = DbNexusAdapter::new("test", 1).await;
973        assert!(result.is_err());
974        if let Err(InklogError::DatabaseError { .. }) = result {
975            // Expected
976        } else {
977            panic!("Expected DatabaseError");
978        }
979    }
980
981    // ============================================================================
982    // MockDatabaseAdapter 测试
983    // ============================================================================
984
985    #[tokio::test]
986    async fn test_mock_database_insert_batch() {
987        let db = MockDatabaseAdapter::new();
988
989        let records = vec![
990            LogRecord::new(Level::INFO, "module1".to_string(), "message1".to_string()),
991            LogRecord::new(Level::WARN, "module2".to_string(), "message2".to_string()),
992        ];
993
994        let count = db.insert_batch(&records).await.unwrap();
995        assert_eq!(count, 2);
996        assert_eq!(db.record_count(), 2);
997    }
998
999    #[tokio::test]
1000    async fn test_mock_database_insert_empty_batch() {
1001        let db = MockDatabaseAdapter::new();
1002
1003        let records: Vec<LogRecord> = vec![];
1004        let count = db.insert_batch(&records).await.unwrap();
1005        assert_eq!(count, 0);
1006        assert_eq!(db.record_count(), 0);
1007    }
1008
1009    #[tokio::test]
1010    async fn test_mock_database_is_healthy() {
1011        let db = MockDatabaseAdapter::new();
1012
1013        // 初始状态应该是健康的
1014        assert!(db.is_healthy().await);
1015
1016        // 设置为不健康
1017        db.set_healthy(false);
1018        assert!(!db.is_healthy().await);
1019
1020        // 恢复健康
1021        db.set_healthy(true);
1022        assert!(db.is_healthy().await);
1023    }
1024
1025    #[tokio::test]
1026    async fn test_mock_database_get_records() {
1027        let db = MockDatabaseAdapter::new();
1028
1029        let records = vec![
1030            LogRecord::new(Level::INFO, "module".to_string(), "message1".to_string()),
1031            LogRecord::new(Level::ERROR, "module".to_string(), "message2".to_string()),
1032        ];
1033
1034        db.insert_batch(&records).await.unwrap();
1035
1036        let stored = db.get_records();
1037        assert_eq!(stored.len(), 2);
1038        assert_eq!(stored[0].message, "message1");
1039        assert_eq!(stored[1].message, "message2");
1040    }
1041
1042    #[tokio::test]
1043    async fn test_mock_database_clear() {
1044        let db = MockDatabaseAdapter::new();
1045
1046        let records = vec![LogRecord::new(
1047            Level::INFO,
1048            "module".to_string(),
1049            "message".to_string(),
1050        )];
1051
1052        db.insert_batch(&records).await.unwrap();
1053        assert_eq!(db.record_count(), 1);
1054
1055        db.clear();
1056        assert_eq!(db.record_count(), 0);
1057    }
1058
1059    #[tokio::test]
1060    async fn test_mock_database_default() {
1061        let db = MockDatabaseAdapter::default();
1062
1063        assert!(db.is_healthy().await);
1064        assert_eq!(db.record_count(), 0);
1065    }
1066
1067    #[tokio::test]
1068    async fn test_mock_database_multiple_inserts() {
1069        let db = MockDatabaseAdapter::new();
1070
1071        // 第一次插入
1072        let records1 = vec![LogRecord::new(
1073            Level::INFO,
1074            "module1".to_string(),
1075            "message1".to_string(),
1076        )];
1077        db.insert_batch(&records1).await.unwrap();
1078        assert_eq!(db.record_count(), 1);
1079
1080        // 第二次插入
1081        let records2 = vec![LogRecord::new(
1082            Level::WARN,
1083            "module2".to_string(),
1084            "message2".to_string(),
1085        )];
1086        db.insert_batch(&records2).await.unwrap();
1087        assert_eq!(db.record_count(), 2);
1088
1089        // 验证记录顺序
1090        let stored = db.get_records();
1091        assert_eq!(stored[0].message, "message1");
1092        assert_eq!(stored[1].message, "message2");
1093    }
1094
1095    // ============================================================================
1096    // DbNexusAdapter getter 与空批量插入测试
1097    // 覆盖行：181-183 (Ok(Self)), 203-204 (pool()), 208-209 (table_name()), 218 (Ok(0))
1098    // ============================================================================
1099
1100    #[cfg(feature = "sqlite")]
1101    #[tokio::test]
1102    async fn test_dbnexus_adapter_with_table_name_creates_instance() {
1103        // 覆盖行 181-183：with_table_name 成功路径返回 Ok(Self { pool, table_name })
1104        let temp_dir = std::env::temp_dir();
1105        let db_path = temp_dir.join("inklog_with_table_name.db");
1106        let db_url = format!("sqlite:{}?mode=rwc", db_path.to_string_lossy());
1107
1108        let adapter = DbNexusAdapter::with_table_name(&db_url, 1, "custom_logs")
1109            .await
1110            .expect("with_table_name should succeed");
1111
1112        // 覆盖行 208-209：table_name() getter
1113        assert_eq!(adapter.table_name(), "custom_logs");
1114
1115        let _ = std::fs::remove_file(&db_path);
1116    }
1117
1118    #[cfg(feature = "sqlite")]
1119    #[tokio::test]
1120    async fn test_dbnexus_adapter_pool_getter_returns_underlying_pool() {
1121        // 覆盖行 203-204：pool() getter
1122        let temp_dir = std::env::temp_dir();
1123        let db_path = temp_dir.join("inklog_pool_getter.db");
1124        let db_url = format!("sqlite:{}?mode=rwc", db_path.to_string_lossy());
1125
1126        let adapter = DbNexusAdapter::new(&db_url, 1)
1127            .await
1128            .expect("new should succeed");
1129
1130        // pool() 返回底层 DbPool 引用——验证可获取 admin 会话即可
1131        let _session = adapter
1132            .pool()
1133            .get_session("admin")
1134            .await
1135            .expect("should get session from underlying pool");
1136
1137        let _ = std::fs::remove_file(&db_path);
1138    }
1139
1140    #[cfg(feature = "sqlite")]
1141    #[tokio::test]
1142    async fn test_dbnexus_adapter_insert_empty_batch_returns_zero() {
1143        // 覆盖行 218：insert_batch 收到空切片时立即返回 Ok(0)
1144        let temp_dir = std::env::temp_dir();
1145        let db_path = temp_dir.join("inklog_empty_batch.db");
1146        let db_url = format!("sqlite:{}?mode=rwc", db_path.to_string_lossy());
1147
1148        let adapter = DbNexusAdapter::new(&db_url, 1)
1149            .await
1150            .expect("new should succeed");
1151
1152        let empty: Vec<LogRecord> = vec![];
1153        let count = adapter
1154            .insert_batch(&empty)
1155            .await
1156            .expect("empty batch should succeed");
1157        assert_eq!(count, 0, "empty batch must return 0");
1158
1159        let _ = std::fs::remove_file(&db_path);
1160    }
1161
1162    // ============================================================================
1163    // T003: SQL injection via unescaped level field
1164    // ============================================================================
1165
1166    #[test]
1167    fn test_level_field_single_quote_escaping() {
1168        // Verify that the escaping pattern used in insert_batch correctly neutralizes
1169        // SQL injection via single quotes in the level field.
1170        let malicious_level = "INFO'OR'1'='1";
1171        let escaped = malicious_level.replace('\'', "''");
1172        assert_eq!(escaped, "INFO''OR''1''=''1");
1173        // After escaping, every original single quote should be doubled,
1174        // so the string cannot break out of a SQL string literal.
1175        assert_eq!(
1176            escaped.matches('\'').count(),
1177            8,
1178            "all 4 original quotes should be doubled"
1179        );
1180    }
1181
1182    // ============================================================================
1183    // T004: Table name validation
1184    // ============================================================================
1185
1186    #[cfg(any(
1187        feature = "sqlite",
1188        feature = "postgres",
1189        feature = "mysql",
1190        feature = "duckdb"
1191    ))]
1192    #[test]
1193    fn test_table_name_accepts_valid_names() {
1194        assert!(validate_table_name("logs").is_ok());
1195        assert!(validate_table_name("my_logs").is_ok());
1196        assert!(validate_table_name("_private").is_ok());
1197        assert!(validate_table_name("Logs123").is_ok());
1198        assert!(validate_table_name("a").is_ok());
1199    }
1200
1201    #[cfg(any(
1202        feature = "sqlite",
1203        feature = "postgres",
1204        feature = "mysql",
1205        feature = "duckdb"
1206    ))]
1207    #[test]
1208    fn test_table_name_rejects_sql_injection() {
1209        // SQL injection attempts
1210        assert!(validate_table_name("logs; DROP TABLE users").is_err());
1211        assert!(validate_table_name("logs' OR '1'='1").is_err());
1212        assert!(validate_table_name("logs--comment").is_err());
1213        assert!(validate_table_name("logs\";").is_err());
1214        // Empty name
1215        assert!(validate_table_name("").is_err());
1216        // Starts with digit
1217        assert!(validate_table_name("123logs").is_err());
1218        // Contains dot (schema.table)
1219        assert!(validate_table_name("public.logs").is_err());
1220    }
1221
1222    // ============================================================================
1223    // T009: escape_sql_string 单元测试
1224    // ============================================================================
1225
1226    #[cfg(any(
1227        feature = "sqlite",
1228        feature = "postgres",
1229        feature = "mysql",
1230        feature = "duckdb"
1231    ))]
1232    #[test]
1233    fn test_escape_sql_string_empty() {
1234        assert_eq!(escape_sql_string(""), "");
1235    }
1236
1237    #[cfg(any(
1238        feature = "sqlite",
1239        feature = "postgres",
1240        feature = "mysql",
1241        feature = "duckdb"
1242    ))]
1243    #[test]
1244    fn test_escape_sql_string_no_special_chars() {
1245        assert_eq!(escape_sql_string("hello world"), "hello world");
1246        assert_eq!(escape_sql_string("abc123"), "abc123");
1247    }
1248
1249    #[cfg(any(
1250        feature = "sqlite",
1251        feature = "postgres",
1252        feature = "mysql",
1253        feature = "duckdb"
1254    ))]
1255    #[test]
1256    fn test_escape_sql_string_single_quote() {
1257        assert_eq!(escape_sql_string("it's"), "it''s");
1258        assert_eq!(escape_sql_string("'"), "''");
1259    }
1260
1261    #[cfg(any(
1262        feature = "sqlite",
1263        feature = "postgres",
1264        feature = "mysql",
1265        feature = "duckdb"
1266    ))]
1267    #[test]
1268    fn test_escape_sql_string_multiple_quotes() {
1269        assert_eq!(escape_sql_string("a'b'c'd"), "a''b''c''d");
1270        assert_eq!(escape_sql_string("''''"), "''''''''");
1271    }
1272
1273    #[cfg(any(
1274        feature = "sqlite",
1275        feature = "postgres",
1276        feature = "mysql",
1277        feature = "duckdb"
1278    ))]
1279    #[test]
1280    fn test_escape_sql_string_unicode() {
1281        assert_eq!(escape_sql_string("こんにちは"), "こんにちは");
1282        assert_eq!(escape_sql_string("世界'it's"), "世界''it''s");
1283    }
1284
1285    // ============================================================================
1286    // T010: generate_create_table_sql 测试
1287    // ============================================================================
1288
1289    #[cfg(any(
1290        feature = "sqlite",
1291        feature = "postgres",
1292        feature = "mysql",
1293        feature = "duckdb"
1294    ))]
1295    #[test]
1296    fn test_generate_create_table_sql_sqlite() {
1297        let ddl = generate_create_table_sql("logs", &DatabaseDriver::SQLite);
1298        assert!(ddl.contains("CREATE TABLE IF NOT EXISTS logs"));
1299        assert!(ddl.contains("INTEGER PRIMARY KEY AUTOINCREMENT"));
1300        assert!(ddl.contains("TEXT NOT NULL"));
1301        assert!(ddl.contains("line INTEGER"));
1302    }
1303
1304    #[cfg(any(
1305        feature = "sqlite",
1306        feature = "postgres",
1307        feature = "mysql",
1308        feature = "duckdb"
1309    ))]
1310    #[test]
1311    fn test_generate_create_table_sql_postgres() {
1312        let ddl = generate_create_table_sql("logs", &DatabaseDriver::PostgreSQL);
1313        assert!(ddl.contains("CREATE TABLE IF NOT EXISTS logs"));
1314        assert!(ddl.contains("BIGSERIAL PRIMARY KEY"));
1315        assert!(ddl.contains("TIMESTAMPTZ NOT NULL"));
1316        assert!(ddl.contains("TEXT NOT NULL"));
1317    }
1318
1319    #[cfg(any(
1320        feature = "sqlite",
1321        feature = "postgres",
1322        feature = "mysql",
1323        feature = "duckdb"
1324    ))]
1325    #[test]
1326    fn test_generate_create_table_sql_mysql() {
1327        let ddl = generate_create_table_sql("logs", &DatabaseDriver::MySQL);
1328        assert!(ddl.contains("CREATE TABLE IF NOT EXISTS logs"));
1329        assert!(ddl.contains("BIGINT AUTO_INCREMENT PRIMARY KEY"));
1330        assert!(ddl.contains("TIMESTAMP NOT NULL"));
1331        assert!(ddl.contains("TEXT NOT NULL"));
1332    }
1333
1334    // ============================================================================
1335    // T011: 自定义 admin_role 构造测试
1336    // ============================================================================
1337
1338    #[cfg(feature = "sqlite")]
1339    #[tokio::test]
1340    async fn test_dbnexus_adapter_custom_admin_role() {
1341        let temp_dir = std::env::temp_dir();
1342
1343        // 创建权限配置文件，允许 superadmin 角色
1344        let perm_path = temp_dir.join("inklog_custom_role_perm.yaml");
1345        let perm_content = r#"roles:
1346  superadmin:
1347    tables:
1348      - name: "*"
1349        operations: ["select", "insert", "update", "delete"]
1350"#;
1351        std::fs::write(&perm_path, perm_content).expect("Failed to write permissions file");
1352
1353        let db_path = temp_dir.join("inklog_custom_role.db");
1354        let db_url = format!("sqlite:{}?mode=rwc", db_path.to_string_lossy());
1355
1356        let adapter = DbNexusAdapter::with_full_config(
1357            &db_url,
1358            1,
1359            "logs",
1360            Some(perm_path.to_string_lossy().to_string()),
1361            "superadmin",
1362        )
1363        .await
1364        .expect("with_full_config should succeed");
1365
1366        // Verify the adapter stores the custom role
1367        assert_eq!(adapter.admin_role, "superadmin");
1368        assert_eq!(adapter.table_name(), "logs");
1369
1370        let _ = std::fs::remove_file(&perm_path);
1371        let _ = std::fs::remove_file(&db_path);
1372    }
1373}
inklog/integrations/infra/database.rs

inklog/integrations/infra/
database.rs
