sz_orm_audit/

lib.rs

//! # SZ-ORM Audit — SQL Audit Log
//!
//! Provides SQL execution audit records and performs case-insensitive masking
//! on sensitive keywords such as password/token/credit_card, ensuring that
//! audit logs never leak sensitive information.
//!
//! ## Main Types
//!
//! - [`SqlAuditContext`] — Audit context (SQL/user/timestamp)
//! - [`SqlAuditor`] — Audit executor

use serde::{Deserialize, Serialize};
use std::sync::Mutex;

#[cfg(feature = "data-lineage")]
pub mod lineage;

#[cfg(feature = "data-quality")]
pub mod data_quality;

#[cfg(feature = "lineage-viz")]
pub use lineage::{downstream_impact, upstream_trace, ImpactEdge};
#[cfg(feature = "data-lineage")]
pub use lineage::{
    EdgeType, LineageDialect, LineageEdge, LineageError, LineageExportFormat, LineageGraph,
    LineageNode, LineageNodeId, LineageTracker, LineageUpdate, NodeType,
};

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SqlAuditContext {
    pub sql: String,
    pub user: String,
    pub timestamp: i64,
}

/// Sensitive keywords that should be masked in audit logs. Matching is
/// case-insensitive on the ASCII bytes of the SQL string.
const SENSITIVE_KEYWORDS: &[&str] = &[
    "password",
    "pwd",
    "passwd",
    "secret",
    "token",
    "api_key",
    "apikey",
    "access_key",
    "accesskey",
    "session",
    "credit_card",
    "creditcard",
    "cvv",
    "ssn",
];

pub struct SqlAuditor {
    logs: Mutex<Vec<SqlAuditContext>>,
}

impl SqlAuditor {
    pub fn new() -> Self {
        Self {
            logs: Mutex::new(vec![]),
        }
    }

    /// Log an audit entry. The SQL is masked for sensitive keywords before
    /// being stored in the in-memory buffer.
    pub fn log(&self, ctx: &SqlAuditContext) {
        let masked_sql = mask_sensitive(&ctx.sql);
        let entry = SqlAuditContext {
            sql: masked_sql,
            user: ctx.user.clone(),
            timestamp: ctx.timestamp,
        };
        let mut logs = self
            .logs
            .lock()
            .expect("SqlAuditor logs lock poisoned (log)");
        logs.push(entry);
    }

    /// Return a snapshot of all stored audit entries.
    pub fn get_logs(&self) -> Vec<SqlAuditContext> {
        let logs = self
            .logs
            .lock()
            .expect("SqlAuditor logs lock poisoned (get_logs)");
        logs.iter().cloned().collect()
    }

    /// Flush all stored audit entries to a JSON file at `path`.
    /// Returns the number of entries written.
    pub fn flush(&self, path: &str) -> Result<usize, String> {
        let logs = self
            .logs
            .lock()
            .expect("SqlAuditor logs lock poisoned (flush)");
        let snapshot: Vec<&SqlAuditContext> = logs.iter().collect();
        let json = serde_json::to_string_pretty(&snapshot).map_err(|e| e.to_string())?;
        std::fs::write(path, json).map_err(|e| e.to_string())?;
        Ok(logs.len())
    }

    /// Mask all sensitive keywords in `sql` with `******`. Matching is
    /// case-insensitive.
    pub fn mask_sensitive(&self, sql: &str) -> String {
        mask_sensitive(sql)
    }
}

impl Default for SqlAuditor {
    fn default() -> Self {
        Self::new()
    }
}

/// Mask all sensitive keywords in `sql` with `******`. Matching is
/// case-insensitive over the ASCII bytes of the string.
fn mask_sensitive(sql: &str) -> String {
    let lower = sql.to_ascii_lowercase();
    let mut result = String::with_capacity(sql.len());
    let mut i = 0;
    let bytes = sql.as_bytes();
    let lower_bytes = lower.as_bytes();
    while i < bytes.len() {
        let mut matched_len: Option<usize> = None;
        for keyword in SENSITIVE_KEYWORDS {
            let kw_bytes = keyword.as_bytes();
            if i + kw_bytes.len() <= bytes.len() && &lower_bytes[i..i + kw_bytes.len()] == kw_bytes
            {
                // Only treat as a keyword match if it's not part of a longer identifier.
                // Boundary check: previous and next char must be non-alphanumeric/underscore
                let prev_ok = i == 0 || !is_ident_char(bytes[i - 1]);
                let next_idx = i + kw_bytes.len();
                let next_ok = next_idx >= bytes.len() || !is_ident_char(bytes[next_idx]);
                if prev_ok && next_ok {
                    matched_len = Some(kw_bytes.len());
                    break;
                }
            }
        }
        if let Some(kw_len) = matched_len {
            result.push_str("******");
            i += kw_len;
        } else {
            // Push one char (handles UTF-8 properly since we step by char).
            let ch = sql[i..]
                .chars()
                .next()
                .expect("i < bytes.len() guarantees non-empty slice");
            result.push(ch);
            i += ch.len_utf8();
        }
    }
    result
}

/// Returns true if `b` is an ASCII identifier character (alphanumeric or '_').
fn is_ident_char(b: u8) -> bool {
    b.is_ascii_alphanumeric() || b == b'_'
}

// ============================================================================
// 审计规则配置（允许/拒绝列表）
// ============================================================================

/// Audit rules: allow/deny lists used to decide whether a given SQL statement
/// should be recorded in the audit log.
///
/// Rule evaluation order:
/// 1. If the SQL matches any pattern in the deny list → not recorded
/// 2. If the allow list is empty → record all SQL not denied
/// 3. If the allow list is non-empty → record only SQL that matches the allow list
#[derive(Debug, Clone, Default)]
pub struct AuditRules {
    /// Allow-list patterns (case-insensitive substring match); empty means allow all
    allow_patterns: Vec<String>,
    /// Deny-list patterns (case-insensitive substring match); a match means deny
    deny_patterns: Vec<String>,
}

impl AuditRules {
    pub fn new() -> Self {
        Self::default()
    }

    /// Add an allow pattern (case-insensitive substring match)
    pub fn allow(mut self, pattern: impl Into<String>) -> Self {
        self.allow_patterns
            .push(pattern.into().to_ascii_lowercase());
        self
    }

    /// Add a deny pattern (case-insensitive substring match)
    pub fn deny(mut self, pattern: impl Into<String>) -> Self {
        self.deny_patterns.push(pattern.into().to_ascii_lowercase());
        self
    }

    /// Determine whether the given SQL should be audit-recorded
    pub fn should_audit(&self, sql: &str) -> bool {
        let lower = sql.to_ascii_lowercase();
        // 拒绝列表优先
        for pat in &self.deny_patterns {
            if lower.contains(pat) {
                return false;
            }
        }
        // 允许列表为空 → 允许所有
        if self.allow_patterns.is_empty() {
            return true;
        }
        // 允许列表非空 → 仅允许命中项
        self.allow_patterns.iter().any(|pat| lower.contains(pat))
    }

    /// Return the number of allow patterns
    pub fn allow_count(&self) -> usize {
        self.allow_patterns.len()
    }

    /// Return the number of deny patterns
    pub fn deny_count(&self) -> usize {
        self.deny_patterns.len()
    }
}

// ============================================================================
// 审计日志轮转策略（按大小/时间）
// ============================================================================

/// Audit log rotation policy configuration.
///
/// - `max_entries`: maximum number of log entries retained in memory; once
///   exceeded, rotation is triggered automatically (old logs are cleared or
///   flushed to disk)
/// - `max_age_ms`: maximum age of a log entry in milliseconds; once exceeded,
///   rotation is triggered
#[derive(Debug, Clone)]
pub struct RotationPolicy {
    /// Maximum entry count (0 means unlimited)
    pub max_entries: usize,
    /// Maximum age in milliseconds (0 means unlimited)
    pub max_age_ms: i64,
}

impl RotationPolicy {
    /// Create an unlimited rotation policy
    pub fn none() -> Self {
        Self {
            max_entries: 0,
            max_age_ms: 0,
        }
    }

    /// Rotate by entry count
    pub fn by_size(max_entries: usize) -> Self {
        Self {
            max_entries,
            max_age_ms: 0,
        }
    }

    /// Rotate by age (milliseconds)
    pub fn by_age(max_age_ms: i64) -> Self {
        Self {
            max_entries: 0,
            max_age_ms,
        }
    }

    /// Rotate by both entry count and age
    pub fn by_size_and_age(max_entries: usize, max_age_ms: i64) -> Self {
        Self {
            max_entries,
            max_age_ms,
        }
    }

    /// Determine whether rotation is needed
    fn needs_rotation(&self, entry_count: usize, oldest_ts: i64, now_ts: i64) -> bool {
        if self.max_entries > 0 && entry_count >= self.max_entries {
            return true;
        }
        if self.max_age_ms > 0 && oldest_ts > 0 && (now_ts - oldest_ts) > self.max_age_ms {
            return true;
        }
        false
    }
}

impl Default for RotationPolicy {
    fn default() -> Self {
        Self::none()
    }
}

// ============================================================================
// 带轮转和规则的审计器
// ============================================================================

/// Enhanced auditor with rotation policy and audit rules.
///
/// Built on top of `SqlAuditor`, it adds:
/// - Log rotation (automatic cleanup of old logs by size or age)
/// - Audit rules (allow/deny list filtering)
pub struct RotatingAuditor {
    logs: Mutex<Vec<SqlAuditContext>>,
    rules: AuditRules,
    policy: RotationPolicy,
    /// Number of rotations (cleanups) performed
    rotations: Mutex<usize>,
}

impl RotatingAuditor {
    pub fn new(policy: RotationPolicy, rules: AuditRules) -> Self {
        Self {
            logs: Mutex::new(vec![]),
            rules,
            policy,
            rotations: Mutex::new(0),
        }
    }

    /// Create an auditor that rotates only by size, with no rule filtering
    pub fn with_max_entries(max_entries: usize) -> Self {
        Self::new(RotationPolicy::by_size(max_entries), AuditRules::new())
    }

    /// Create an auditor that rotates only by age, with no rule filtering
    pub fn with_max_age(max_age_ms: i64) -> Self {
        Self::new(RotationPolicy::by_age(max_age_ms), AuditRules::new())
    }

    /// Record an audit log, automatically applying rule filtering and rotation policy
    pub fn log(&self, ctx: &SqlAuditContext) -> bool {
        // 规则过滤
        if !self.rules.should_audit(&ctx.sql) {
            return false;
        }
        let masked_sql = mask_sensitive(&ctx.sql);
        let entry = SqlAuditContext {
            sql: masked_sql,
            user: ctx.user.clone(),
            timestamp: ctx.timestamp,
        };
        let mut logs = self
            .logs
            .lock()
            .expect("RotatingAuditor logs lock poisoned (log)");

        // 在添加新条目前检查轮转（确保新条目不被立即清除）
        let now = ctx.timestamp;
        let oldest = logs.first().map(|e| e.timestamp).unwrap_or(now);
        if self.policy.needs_rotation(logs.len(), oldest, now) {
            logs.clear();
            *self
                .rotations
                .lock()
                .expect("RotatingAuditor rotations lock poisoned (log)") += 1;
        }

        logs.push(entry);
        true
    }

    /// Return a snapshot of the current logs
    pub fn get_logs(&self) -> Vec<SqlAuditContext> {
        self.logs
            .lock()
            .expect("RotatingAuditor logs lock poisoned (get_logs)")
            .clone()
    }

    /// Return the number of rotations performed
    pub fn rotation_count(&self) -> usize {
        *self
            .rotations
            .lock()
            .expect("RotatingAuditor rotations lock poisoned (rotation_count)")
    }

    /// Manually trigger rotation (clears the current logs)
    pub fn rotate(&self) -> usize {
        let mut logs = self
            .logs
            .lock()
            .expect("RotatingAuditor logs lock poisoned (rotate)");
        let count = logs.len();
        logs.clear();
        *self
            .rotations
            .lock()
            .expect("RotatingAuditor rotations lock poisoned (rotate)") += 1;
        count
    }

    /// Return the current number of log entries
    pub fn len(&self) -> usize {
        self.logs
            .lock()
            .expect("RotatingAuditor logs lock poisoned (len)")
            .len()
    }

    /// Whether the auditor is empty
    pub fn is_empty(&self) -> bool {
        self.logs
            .lock()
            .expect("RotatingAuditor logs lock poisoned (is_empty)")
            .is_empty()
    }
}

// ============================================================================
// 异步审计写入器
// ============================================================================

/// Asynchronous audit writer: writes audit logs via a background thread to
/// avoid blocking the main thread.
///
/// Uses an `std::sync::mpsc` channel to send logs to a background thread that
/// handles storage. Call `shutdown` to wait for the background thread to exit
/// and return all written logs.
pub struct AsyncAuditWriter {
    sender: std::sync::mpsc::Sender<AsyncCommand>,
    handle: Mutex<Option<std::thread::JoinHandle<Vec<SqlAuditContext>>>>,
}

enum AsyncCommand {
    Log(SqlAuditContext),
    Shutdown,
}

impl AsyncAuditWriter {
    /// Create an async writer and start the background thread
    pub fn new() -> Self {
        let (sender, receiver) = std::sync::mpsc::channel::<AsyncCommand>();
        let handle = std::thread::spawn(move || {
            let mut logs: Vec<SqlAuditContext> = Vec::new();
            for cmd in receiver {
                match cmd {
                    AsyncCommand::Log(ctx) => {
                        let masked_sql = mask_sensitive(&ctx.sql);
                        logs.push(SqlAuditContext {
                            sql: masked_sql,
                            user: ctx.user,
                            timestamp: ctx.timestamp,
                        });
                    }
                    AsyncCommand::Shutdown => break,
                }
            }
            logs
        });
        Self {
            sender,
            handle: Mutex::new(Some(handle)),
        }
    }

    /// Asynchronously record an audit log (non-blocking)
    pub fn log(&self, ctx: &SqlAuditContext) -> Result<(), String> {
        self.sender
            .send(AsyncCommand::Log(ctx.clone()))
            .map_err(|e| format!("AsyncAuditWriter channel closed: {}", e))
    }

    /// Shut down the background thread and return all written logs
    pub fn shutdown(&self) -> Result<Vec<SqlAuditContext>, String> {
        let _ = self.sender.send(AsyncCommand::Shutdown);
        let mut handle_guard = self
            .handle
            .lock()
            .expect("AsyncAuditWriter handle lock poisoned (shutdown)");
        if let Some(handle) = handle_guard.take() {
            handle
                .join()
                .map_err(|e| format!("Thread panicked: {:?}", e))
        } else {
            Err("Already shut down".to_string())
        }
    }
}

impl Default for AsyncAuditWriter {
    fn default() -> Self {
        Self::new()
    }
}

// ============================================================================
// 审计日志查询过滤
// ============================================================================

/// Audit log query filter
#[derive(Debug, Clone, Default)]
pub struct AuditQuery {
    /// Filter by username (exact match; None means no filter)
    pub user: Option<String>,
    /// Time range start (millisecond timestamp; None means no lower bound)
    pub from_ts: Option<i64>,
    /// Time range end (millisecond timestamp; None means no upper bound)
    pub to_ts: Option<i64>,
    /// SQL keyword filter (case-insensitive substring match; None means no filter)
    pub sql_contains: Option<String>,
    /// Limit on the number of returned entries (0 means unlimited)
    pub limit: usize,
}

impl AuditQuery {
    pub fn new() -> Self {
        Self::default()
    }

    /// Filter by username
    pub fn by_user(mut self, user: impl Into<String>) -> Self {
        self.user = Some(user.into());
        self
    }

    /// Filter by time range (millisecond timestamps)
    pub fn by_time_range(mut self, from: i64, to: i64) -> Self {
        self.from_ts = Some(from);
        self.to_ts = Some(to);
        self
    }

    /// Filter by SQL keyword (case-insensitive)
    pub fn by_sql_contains(mut self, keyword: impl Into<String>) -> Self {
        self.sql_contains = Some(keyword.into());
        self
    }

    /// Limit the number of returned entries
    pub fn with_limit(mut self, limit: usize) -> Self {
        self.limit = limit;
        self
    }

    /// Apply the query filter to a list of logs
    pub fn filter(&self, logs: &[SqlAuditContext]) -> Vec<SqlAuditContext> {
        let keyword_lower = self.sql_contains.as_ref().map(|s| s.to_ascii_lowercase());
        let mut result: Vec<SqlAuditContext> = logs
            .iter()
            .filter(|entry| {
                if let Some(u) = &self.user {
                    if entry.user != *u {
                        return false;
                    }
                }
                if let Some(from) = self.from_ts {
                    if entry.timestamp < from {
                        return false;
                    }
                }
                if let Some(to) = self.to_ts {
                    if entry.timestamp > to {
                        return false;
                    }
                }
                if let Some(kw) = &keyword_lower {
                    if !entry.sql.to_ascii_lowercase().contains(kw) {
                        return false;
                    }
                }
                true
            })
            .cloned()
            .collect();
        if self.limit > 0 && result.len() > self.limit {
            result.truncate(self.limit);
        }
        result
    }
}

/// Query logs from `SqlAuditor` by the given conditions
pub fn query_logs(auditor: &SqlAuditor, query: &AuditQuery) -> Vec<SqlAuditContext> {
    let logs = auditor.get_logs();
    query.filter(&logs)
}

// ============================================================================
// 审计日志持久化存储
// ============================================================================

/// Audit log persistent storage backend trait
///
/// Abstracts the persistent storage capability for audit logs across different
/// storage media (file, database, object storage, etc.). Implementors must
/// ensure that `append` is thread-safe; they should call `mask_sensitive`
/// themselves before storage to mask sensitive data.
pub trait AuditLogStore: Send + Sync {
    /// Append an audit log entry (already masked); returns whether it succeeded
    fn append(&self, entry: &SqlAuditContext) -> Result<(), String>;
    /// Read all persisted audit logs
    fn read_all(&self) -> Result<Vec<SqlAuditContext>, String>;
    /// Clear the persistent storage
    fn clear(&self) -> Result<(), String>;
}

/// File-based audit log persistent storage (JSONL format: one JSON per line)
///
/// Applicable scenarios: single-machine deployment, lightweight audit archiving,
/// development debugging. For high-concurrency production scenarios, it is
/// recommended to combine `AsyncAuditWriter` + `FileAuditLogStore` so that a
/// background thread performs serial writes and avoids lock contention.
pub struct FileAuditLogStore {
    path: String,
    write_lock: Mutex<()>,
}

impl FileAuditLogStore {
    /// Create a file audit log store; the target file is created on the first
    /// `append` if it does not exist
    pub fn new(path: impl Into<String>) -> Self {
        Self {
            path: path.into(),
            write_lock: Mutex::new(()),
        }
    }

    /// Return the storage file path
    pub fn path(&self) -> &str {
        &self.path
    }
}

impl AuditLogStore for FileAuditLogStore {
    /// Append an audit log entry (JSONL format; automatically masked before write)
    fn append(&self, entry: &SqlAuditContext) -> Result<(), String> {
        let _guard = self
            .write_lock
            .lock()
            .map_err(|e| format!("write_lock poisoned: {}", e))?;
        // 脱敏后序列化，确保落盘内容不含敏感信息
        let masked_sql = mask_sensitive(&entry.sql);
        let stored = SqlAuditContext {
            sql: masked_sql,
            user: entry.user.clone(),
            timestamp: entry.timestamp,
        };
        let line = serde_json::to_string(&stored).map_err(|e| e.to_string())?;
        // 以追加模式打开，每条日志占一行（JSONL）
        use std::io::Write;
        let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)
            .map_err(|e| format!("open '{}' failed: {}", self.path, e))?;
        writeln!(file, "{}", line).map_err(|e| e.to_string())
    }

    /// Read all persisted audit logs (parse JSONL line by line)
    ///
    /// Returns an empty vec when the file does not exist (treated as no logs
    /// persisted yet).
    fn read_all(&self) -> Result<Vec<SqlAuditContext>, String> {
        let content = match std::fs::read_to_string(&self.path) {
            Ok(c) => c,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                return Ok(Vec::new());
            }
            Err(e) => return Err(format!("read failed: {}", e)),
        };
        let mut result = Vec::new();
        for (lineno, line) in content.lines().enumerate() {
            let line = line.trim();
            if line.is_empty() {
                continue;
            }
            let entry: SqlAuditContext = serde_json::from_str(line)
                .map_err(|e| format!("parse line {} failed: {}", lineno + 1, e))?;
            result.push(entry);
        }
        Ok(result)
    }

    /// Clear the persistent file (deletes the file; the next append recreates it)
    fn clear(&self) -> Result<(), String> {
        let _guard = self
            .write_lock
            .lock()
            .map_err(|e| format!("write_lock poisoned: {}", e))?;
        std::fs::remove_file(&self.path).or_else(|e| {
            // 文件不存在视为已清空
            if e.kind() == std::io::ErrorKind::NotFound {
                Ok(())
            } else {
                Err(format!("clear failed: {}", e))
            }
        })
    }
}

// ============================================================================
// #11 修复：审计日志哈希链防篡改（Tamper-Evident Hash Chain）
// ============================================================================

/// Genesis hash (the previous hash of the chain head), fixed as an all-zero
/// 64-character hexadecimal string.
///
/// The first record of every hash chain uses `GENESIS_HASH` as its `prev_hash`,
/// making it easy to verify that the chain start has not been truncated.
pub const GENESIS_HASH: &str = "0000000000000000000000000000000000000000000000000000000000000000";

/// Audit log entry with a hash chain.
///
/// Each record contains:
/// - `prev_hash`: the `current_hash` of the previous record (the first record
///   uses `GENESIS_HASH`)
/// - `current_hash`: the SHA-256 hash of this record (computed from
///   `prev_hash + entry`)
/// - `entry`: the original audit context (already masked)
///
/// Any tampering with a historical record causes `current_hash` to mismatch the
/// `prev_hash` of the next record, which is detected by `verify_chain`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HashChainEntry {
    /// Hash of the previous record (the first record uses [`GENESIS_HASH`])
    pub prev_hash: String,
    /// Hash of this record (SHA-256 hexadecimal string, 64 characters)
    pub current_hash: String,
    /// Original audit context (already masked)
    pub entry: SqlAuditContext,
}

impl HashChainEntry {
    /// Compute the `current_hash` of a single record.
    ///
    /// The hash input is the UTF-8 byte concatenation of
    /// `prev_hash || sql || user || timestamp`, using the SHA-256 algorithm.
    /// This way, tampering with any field changes the hash, which in turn
    /// affects the `prev_hash` of the next record, forming a chain of
    /// verification.
    fn compute_hash(prev_hash: &str, entry: &SqlAuditContext) -> String {
        use sha2::{Digest, Sha256};
        let mut hasher = Sha256::new();
        hasher.update(prev_hash.as_bytes());
        hasher.update(entry.sql.as_bytes());
        hasher.update(entry.user.as_bytes());
        // timestamp 使用固定宽度的字节表示，避免可变长度编码导致歧义
        hasher.update(entry.timestamp.to_le_bytes());
        let result = hasher.finalize();
        // 转为小写十六进制字符串（64 字符）
        hex_encode(&result)
    }

    /// Create the chain head record (prev_hash = GENESIS_HASH)
    pub fn genesis(entry: SqlAuditContext) -> Self {
        let prev_hash = GENESIS_HASH.to_string();
        let current_hash = Self::compute_hash(&prev_hash, &entry);
        Self {
            prev_hash,
            current_hash,
            entry,
        }
    }

    /// Append a record to the given previous hash
    pub fn append(prev_hash: &str, entry: SqlAuditContext) -> Self {
        let current_hash = Self::compute_hash(prev_hash, &entry);
        Self {
            prev_hash: prev_hash.to_string(),
            current_hash,
            entry,
        }
    }
}

/// Encode a byte array as a lowercase hexadecimal string.
///
/// Consistent with the behavior of the `hex` crate's `encode`, but avoids
/// introducing an extra dependency.
fn hex_encode(bytes: &[u8]) -> String {
    const HEX_CHARS: &[u8] = b"0123456789abcdef";
    let mut s = String::with_capacity(bytes.len() * 2);
    for &b in bytes {
        s.push(HEX_CHARS[(b >> 4) as usize] as char);
        s.push(HEX_CHARS[(b & 0x0f) as usize] as char);
    }
    s
}

/// Auditor with a hash chain: all logs are linked by SHA-256 chained hashes,
/// supporting tamper detection.
///
/// # Tamper-Evidence Mechanism
///
/// 1. Each record's `current_hash = SHA256(prev_hash || sql || user || timestamp)`
/// 2. The `prev_hash` of the next record equals the `current_hash` of the
///    previous record
/// 3. Any modification to a historical record changes `current_hash`, which
///    then mismatches the `prev_hash` of the next record
/// 4. Deleting a middle record breaks the chain; inserting a record changes
///    all subsequent hashes
///
/// # Example
///
/// ```
/// use sz_orm_audit::{HashChainAuditor, SqlAuditContext};
///
/// let mut auditor = HashChainAuditor::new();
/// auditor.log(&SqlAuditContext {
///     sql: "SELECT * FROM users".to_string(),
///     user: "admin".to_string(),
///     timestamp: 1000,
/// });
/// // Verify chain integrity
/// assert!(auditor.verify().is_ok());
/// ```
pub struct HashChainAuditor {
    /// Hash chain log entries (stored in append order)
    entries: Mutex<Vec<HashChainEntry>>,
}

impl Default for HashChainAuditor {
    fn default() -> Self {
        Self::new()
    }
}

impl HashChainAuditor {
    /// Create an empty hash chain auditor
    pub fn new() -> Self {
        Self {
            entries: Mutex::new(Vec::new()),
        }
    }

    /// Append an audit log to the end of the hash chain.
    ///
    /// - If the chain is empty, uses [`GENESIS_HASH`] as `prev_hash`
    /// - Otherwise uses the `current_hash` of the last record as `prev_hash`
    ///
    /// The SQL is first masked by `mask_sensitive` before being written to the
    /// chain, ensuring that the stored audit logs contain no sensitive
    /// information.
    pub fn log(&self, ctx: &SqlAuditContext) {
        let masked_sql = mask_sensitive(&ctx.sql);
        let entry = SqlAuditContext {
            sql: masked_sql,
            user: ctx.user.clone(),
            timestamp: ctx.timestamp,
        };
        let mut entries = self
            .entries
            .lock()
            .expect("HashChainAuditor entries lock poisoned (log)");
        let prev_hash = entries
            .last()
            .map(|e| e.current_hash.as_str())
            .unwrap_or(GENESIS_HASH);
        let chain_entry = if entries.is_empty() {
            HashChainEntry::genesis(entry)
        } else {
            HashChainEntry::append(prev_hash, entry)
        };
        entries.push(chain_entry);
    }

    /// Return a snapshot (clone) of all log entries
    pub fn get_entries(&self) -> Vec<HashChainEntry> {
        self.entries
            .lock()
            .expect("HashChainAuditor entries lock poisoned (get_entries)")
            .clone()
    }

    /// Return the number of log entries
    pub fn len(&self) -> usize {
        self.entries
            .lock()
            .expect("HashChainAuditor entries lock poisoned (len)")
            .len()
    }

    /// Whether the auditor is empty
    pub fn is_empty(&self) -> bool {
        self.entries
            .lock()
            .expect("HashChainAuditor entries lock poisoned (is_empty)")
            .is_empty()
    }

    /// Verify the integrity of the hash chain.
    ///
    /// Checks performed:
    /// 1. The `prev_hash` of the first record equals [`GENESIS_HASH`]
    /// 2. The `current_hash` of each record equals
    ///    `compute_hash(prev_hash, entry)`
    /// 3. The `prev_hash` of each adjacent record equals the `current_hash` of
    ///    the previous record
    ///
    /// # Return value
    ///
    /// - `Ok(())`: the chain is intact and has not been tampered with
    /// - `Err(reason)`: the chain has been tampered with; `reason` describes
    ///   the location and type of the first anomaly
    pub fn verify(&self) -> Result<(), String> {
        let entries = self
            .entries
            .lock()
            .expect("HashChainAuditor entries lock poisoned (verify)");
        for (i, entry) in entries.iter().enumerate() {
            // 检查 1：首条记录的 prev_hash 必须为 GENESIS_HASH
            if i == 0 {
                if entry.prev_hash != GENESIS_HASH {
                    return Err(format!(
                        "chain genesis prev_hash mismatch at index 0: expected '{}', got '{}'",
                        GENESIS_HASH, entry.prev_hash
                    ));
                }
            } else {
                // 检查 3：非首条记录的 prev_hash 必须等于上一条的 current_hash
                let prev = &entries[i - 1];
                if entry.prev_hash != prev.current_hash {
                    return Err(format!(
                        "chain broken at index {}: prev_hash '{}' != previous current_hash '{}'",
                        i, entry.prev_hash, prev.current_hash
                    ));
                }
            }
            // 检查 2：current_hash 必须等于重新计算的哈希
            let recomputed = HashChainEntry::compute_hash(&entry.prev_hash, &entry.entry);
            if entry.current_hash != recomputed {
                return Err(format!(
                    "hash mismatch at index {}: stored '{}' != recomputed '{}'",
                    i, entry.current_hash, recomputed
                ));
            }
        }
        Ok(())
    }

    /// Persist the hash chain to a JSONL file (one JSON per line).
    ///
    /// Returns the number of entries written.
    pub fn flush(&self, path: &str) -> Result<usize, String> {
        let entries = self
            .entries
            .lock()
            .expect("HashChainAuditor entries lock poisoned (flush)");
        let snapshot: Vec<&HashChainEntry> = entries.iter().collect();
        let json = serde_json::to_string_pretty(&snapshot).map_err(|e| e.to_string())?;
        std::fs::write(path, json).map_err(|e| e.to_string())?;
        Ok(entries.len())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Test data directory: prefers F:\test\data (user convention), falls back
    /// to the environment variable or system temp (CI/Linux).
    ///
    /// Note: checking only that the directory exists is not enough to guarantee
    /// usability — write permission must also be verified, to avoid test
    /// failures in restricted sandbox environments (e.g. TRAE Sandbox) where
    /// the directory exists but is not writable.
    fn test_data_dir() -> std::path::PathBuf {
        let f_drive = std::path::Path::new("F:\\test\\data");
        if is_dir_writable(f_drive) {
            return f_drive.to_path_buf();
        }
        if let Ok(dir) = std::env::var("SZ_ORM_TEST_DATA_DIR") {
            let p = std::path::PathBuf::from(&dir);
            if is_dir_writable(&p) {
                return p;
            }
        }
        std::env::temp_dir()
    }

    /// Check whether the directory exists and is writable: try to create and
    /// delete a probe file inside it
    fn is_dir_writable(dir: &std::path::Path) -> bool {
        if !dir.exists() {
            return false;
        }
        let probe = dir.join(format!(".probe_{}", std::process::id()));
        match std::fs::File::create(&probe) {
            Ok(_) => {
                let _ = std::fs::remove_file(&probe);
                true
            }
            Err(_) => false,
        }
    }

    fn ctx(sql: &str, user: &str, ts: i64) -> SqlAuditContext {
        SqlAuditContext {
            sql: sql.to_string(),
            user: user.to_string(),
            timestamp: ts,
        }
    }

    #[test]
    fn test_log_stores_in_memory() {
        let a = SqlAuditor::new();
        a.log(&ctx("SELECT * FROM users", "admin", 1000));
        a.log(&ctx("INSERT INTO logs VALUES(1)", "admin", 1001));
        let logs = a.get_logs();
        assert_eq!(logs.len(), 2);
        assert_eq!(logs[0].sql, "SELECT * FROM users");
        assert_eq!(logs[0].user, "admin");
        assert_eq!(logs[0].timestamp, 1000);
        assert_eq!(logs[1].timestamp, 1001);
    }

    #[test]
    fn test_log_masks_sensitive_in_storage() {
        let a = SqlAuditor::new();
        a.log(&ctx(
            "SELECT * FROM users WHERE password='secret'",
            "admin",
            1000,
        ));
        let logs = a.get_logs();
        assert_eq!(logs.len(), 1);
        let stored_sql = &logs[0].sql;
        assert!(!stored_sql.contains("password"));
        assert!(!stored_sql.contains("secret"));
        assert!(stored_sql.contains("******"));
    }

    #[test]
    fn test_mask_sensitive_password() {
        let a = SqlAuditor::new();
        let masked = a.mask_sensitive("SELECT * FROM users WHERE password='secret'");
        assert!(!masked.contains("password"));
        assert!(!masked.contains("secret"));
        assert!(masked.contains("******"));
    }

    #[test]
    fn test_mask_sensitive_case_insensitive() {
        let a = SqlAuditor::new();
        let masked = a.mask_sensitive("UPDATE users SET PASSWORD='abc', Token='x'");
        let lower = masked.to_lowercase();
        assert!(!lower.contains("password"));
        assert!(!lower.contains("token"));
        assert!(masked.contains("******"));
    }

    #[test]
    fn test_mask_sensitive_extended_keywords() {
        let a = SqlAuditor::new();
        let inputs = [
            "pwd",
            "passwd",
            "secret",
            "api_key",
            "access_key",
            "session",
            "credit_card",
            "cvv",
            "ssn",
        ];
        for kw in inputs {
            let sql = format!("SELECT * FROM t WHERE k = '{}'", kw);
            let masked = a.mask_sensitive(&sql);
            let lower = masked.to_lowercase();
            assert!(
                !lower.contains(kw),
                "keyword '{}' should be masked in: {}",
                kw,
                masked
            );
            assert!(masked.contains("******"));
        }
    }

    #[test]
    fn test_mask_sensitive_preserves_non_sensitive() {
        let a = SqlAuditor::new();
        let masked = a.mask_sensitive("SELECT id, name FROM users WHERE active = 1");
        assert_eq!(masked, "SELECT id, name FROM users WHERE active = 1");
    }

    #[test]
    fn test_mask_sensitive_does_not_match_substrings() {
        // "passworded" should not be partially matched as "password"
        // because we require word boundaries.
        let a = SqlAuditor::new();
        let masked = a.mask_sensitive("SELECT * FROM users WHERE note='passworded'");
        // The 'passworded' word should remain intact because of boundary check
        assert!(masked.contains("passworded"));
        // Ensure we did NOT replace anything (no ****** from this substring)
        // Actually, "passworded" still has 'password' as a prefix but our
        // boundary check requires the char AFTER the keyword to be non-ident.
        // In "passworded", after "password" comes "e" which IS an ident char,
        // so it should NOT be matched.
        assert_eq!(masked, "SELECT * FROM users WHERE note='passworded'");
    }

    #[test]
    fn test_mask_sensitive_multiple_occurrences() {
        let a = SqlAuditor::new();
        let masked = a.mask_sensitive("INSERT INTO t (password, token) VALUES ('p1', 't1')");
        // Both keywords should be masked
        let lower = masked.to_lowercase();
        assert!(!lower.contains("password"));
        assert!(!lower.contains("token"));
        // Verify there are at least 2 mask replacements
        let count = masked.matches("******").count();
        assert!(count >= 2, "expected at least 2 masks, got: {}", masked);
    }

    #[test]
    fn test_get_logs_empty_initially() {
        let a = SqlAuditor::new();
        assert!(a.get_logs().is_empty());
    }

    #[test]
    fn test_get_logs_returns_snapshot_independent_of_changes() {
        let a = SqlAuditor::new();
        a.log(&ctx("SELECT 1", "u", 1));
        let snap = a.get_logs();
        a.log(&ctx("SELECT 2", "u", 2));
        assert_eq!(snap.len(), 1, "snapshot should not change after new log");
        assert_eq!(a.get_logs().len(), 2);
    }

    #[test]
    fn test_flush_writes_json_file() {
        let a = SqlAuditor::new();
        a.log(&ctx("SELECT * FROM users WHERE password='p'", "admin", 123));
        a.log(&ctx("INSERT INTO logs VALUES(1)", "user2", 456));
        let path = test_data_dir().join("sz_orm_audit_flush_test.json");
        let path_str = path.to_str().unwrap();
        let count = a.flush(path_str).expect("flush should succeed");
        assert_eq!(count, 2);
        // Read back the file and verify it contains valid JSON
        let content = std::fs::read_to_string(path_str).expect("file should be readable");
        let parsed: Vec<SqlAuditContext> =
            serde_json::from_str(&content).expect("should parse as JSON array");
        assert_eq!(parsed.len(), 2);
        assert_eq!(parsed[0].user, "admin");
        assert_eq!(parsed[1].timestamp, 456);
        // Verify masking was applied during log()
        assert!(!parsed[0].sql.contains("password"));
        // Cleanup
        let _ = std::fs::remove_file(path_str);
    }

    #[test]
    fn test_flush_empty_writes_empty_array() {
        let a = SqlAuditor::new();
        let path = test_data_dir().join("sz_orm_audit_flush_empty_test.json");
        let path_str = path.to_str().unwrap();
        let count = a.flush(path_str).expect("flush should succeed");
        assert_eq!(count, 0);
        let content = std::fs::read_to_string(path_str).expect("file should be readable");
        assert_eq!(content.trim(), "[]");
        let _ = std::fs::remove_file(path_str);
    }

    #[test]
    fn test_default_creates_new_auditor() {
        let a = SqlAuditor::default();
        assert!(a.get_logs().is_empty());
    }

    #[test]
    fn test_original_test_compatibility() {
        // Backward compatibility: the original test asserts that masking
        // "SELECT * FROM users WHERE password='secret'" removes "password".
        let a = SqlAuditor::new();
        let masked = a.mask_sensitive("SELECT * FROM users WHERE password='secret'");
        assert!(!masked.contains("password"));
    }

    // ===== 审计规则配置测试 =====

    #[test]
    fn test_audit_rules_empty_allows_all() {
        let rules = AuditRules::new();
        assert!(rules.should_audit("SELECT * FROM users"));
        assert!(rules.should_audit("DELETE FROM orders"));
        assert_eq!(rules.allow_count(), 0);
        assert_eq!(rules.deny_count(), 0);
    }

    #[test]
    fn test_audit_rules_deny_blocks() {
        let rules = AuditRules::new().deny("pg_catalog");
        assert!(!rules.should_audit("SELECT * FROM pg_catalog.tables"));
        assert!(rules.should_audit("SELECT * FROM users"));
    }

    #[test]
    fn test_audit_rules_allow_filters() {
        let rules = AuditRules::new().allow("select").allow("insert");
        assert!(rules.should_audit("SELECT * FROM users"));
        assert!(rules.should_audit("INSERT INTO logs VALUES(1)"));
        assert!(!rules.should_audit("DELETE FROM users"));
    }

    #[test]
    fn test_audit_rules_deny_overrides_allow() {
        let rules = AuditRules::new().allow("select").deny("password");
        // 包含 password 的 SELECT 应被拒绝
        assert!(!rules.should_audit("SELECT * FROM users WHERE password='x'"));
        // 不含 password 的 SELECT 应被允许
        assert!(rules.should_audit("SELECT * FROM users"));
    }

    #[test]
    fn test_audit_rules_case_insensitive() {
        let rules = AuditRules::new().deny("DROP");
        assert!(!rules.should_audit("drop table users"));
        assert!(!rules.should_audit("DROP TABLE users"));
        assert!(rules.should_audit("SELECT * FROM users"));
    }

    #[test]
    fn test_audit_rules_multiple_deny() {
        let rules = AuditRules::new()
            .deny("drop")
            .deny("truncate")
            .deny("shutdown");
        assert!(!rules.should_audit("DROP TABLE x"));
        assert!(!rules.should_audit("TRUNCATE TABLE y"));
        assert!(!rules.should_audit("SHUTDOWN"));
        assert!(rules.should_audit("SELECT 1"));
    }

    // ===== 轮转策略测试 =====

    #[test]
    fn test_rotation_policy_none_never_rotates() {
        let policy = RotationPolicy::none();
        assert!(!policy.needs_rotation(1_000_000, 0, 1_000_000));
        assert!(!policy.needs_rotation(0, 0, 0));
    }

    #[test]
    fn test_rotation_policy_by_size() {
        let policy = RotationPolicy::by_size(100);
        assert!(!policy.needs_rotation(99, 0, 1000));
        assert!(policy.needs_rotation(100, 0, 1000));
        assert!(policy.needs_rotation(200, 0, 1000));
    }

    #[test]
    fn test_rotation_policy_by_age() {
        let policy = RotationPolicy::by_age(5000);
        // 旧日志 4000ms 前，未超时
        assert!(!policy.needs_rotation(10, 5000, 9000));
        // 旧日志 6000ms 前，已超时
        assert!(policy.needs_rotation(10, 5000, 11000));
    }

    #[test]
    fn test_rotation_policy_by_size_and_age() {
        let policy = RotationPolicy::by_size_and_age(100, 5000);
        // 大小未达上限且时间未超 → 不轮转
        assert!(!policy.needs_rotation(50, 5000, 9000));
        // 大小达上限 → 轮转
        assert!(policy.needs_rotation(100, 5000, 5000));
        // 时间超 → 轮转
        assert!(policy.needs_rotation(10, 5000, 11000));
    }

    // ===== RotatingAuditor 测试 =====

    #[test]
    fn test_rotating_auditor_no_rotation_stores_all() {
        let auditor = RotatingAuditor::new(RotationPolicy::none(), AuditRules::new());
        for i in 0..100 {
            auditor.log(&ctx(&format!("SELECT {}", i), "user", i));
        }
        assert_eq!(auditor.len(), 100);
        assert_eq!(auditor.rotation_count(), 0);
    }

    #[test]
    fn test_rotating_auditor_rotates_by_size() {
        let auditor = RotatingAuditor::with_max_entries(5);
        for i in 0..5 {
            auditor.log(&ctx(&format!("SELECT {}", i), "user", i));
        }
        assert_eq!(auditor.len(), 5);
        assert_eq!(auditor.rotation_count(), 0);
        // 第 6 条触发轮转
        auditor.log(&ctx("SELECT 6", "user", 100));
        assert_eq!(auditor.len(), 1);
        assert_eq!(auditor.rotation_count(), 1);
    }

    #[test]
    fn test_rotating_auditor_rotates_by_age() {
        let auditor = RotatingAuditor::with_max_age(1000);
        auditor.log(&ctx("SELECT 1", "user", 100));
        auditor.log(&ctx("SELECT 2", "user", 200));
        assert_eq!(auditor.len(), 2);
        assert_eq!(auditor.rotation_count(), 0);
        // 时间差超过 1000ms → 轮转
        auditor.log(&ctx("SELECT 3", "user", 1500));
        assert_eq!(auditor.len(), 1);
        assert_eq!(auditor.rotation_count(), 1);
    }

    #[test]
    fn test_rotating_auditor_rules_filter() {
        let rules = AuditRules::new().deny("drop").allow("select");
        let auditor = RotatingAuditor::new(RotationPolicy::none(), rules);
        let logged1 = auditor.log(&ctx("SELECT * FROM users", "u", 1));
        let logged2 = auditor.log(&ctx("DROP TABLE users", "u", 2));
        let logged3 = auditor.log(&ctx("DELETE FROM users", "u", 3));
        assert!(logged1);
        assert!(!logged2);
        assert!(!logged3);
        assert_eq!(auditor.len(), 1);
    }

    #[test]
    fn test_rotating_auditor_manual_rotate() {
        let auditor = RotatingAuditor::with_max_entries(100);
        auditor.log(&ctx("SELECT 1", "u", 1));
        auditor.log(&ctx("SELECT 2", "u", 2));
        let cleared = auditor.rotate();
        assert_eq!(cleared, 2);
        assert!(auditor.is_empty());
        assert_eq!(auditor.rotation_count(), 1);
    }

    #[test]
    fn test_rotating_auditor_masks_sensitive() {
        let auditor = RotatingAuditor::with_max_entries(100);
        auditor.log(&ctx("SELECT * FROM users WHERE password='x'", "u", 1));
        let logs = auditor.get_logs();
        assert_eq!(logs.len(), 1);
        assert!(!logs[0].sql.contains("password"));
        assert!(logs[0].sql.contains("******"));
    }

    #[test]
    fn test_rotating_auditor_get_logs_snapshot() {
        let auditor = RotatingAuditor::with_max_entries(100);
        auditor.log(&ctx("SELECT 1", "u", 1));
        let snap = auditor.get_logs();
        auditor.log(&ctx("SELECT 2", "u", 2));
        assert_eq!(snap.len(), 1, "snapshot should be independent");
        assert_eq!(auditor.len(), 2);
    }

    // ===== 异步审计写入器测试 =====

    #[test]
    fn test_async_writer_log_and_shutdown() {
        let writer = AsyncAuditWriter::new();
        writer
            .log(&ctx("SELECT * FROM users", "admin", 1000))
            .unwrap();
        writer
            .log(&ctx("INSERT INTO logs VALUES(1)", "user2", 2000))
            .unwrap();
        let logs = writer.shutdown().expect("shutdown should succeed");
        assert_eq!(logs.len(), 2);
        assert_eq!(logs[0].user, "admin");
        assert_eq!(logs[1].timestamp, 2000);
    }

    #[test]
    fn test_async_writer_masks_sensitive() {
        let writer = AsyncAuditWriter::new();
        writer
            .log(&ctx("SELECT * FROM users WHERE password='secret'", "u", 1))
            .unwrap();
        let logs = writer.shutdown().unwrap();
        assert_eq!(logs.len(), 1);
        assert!(!logs[0].sql.contains("password"));
    }

    #[test]
    fn test_async_writer_empty_shutdown() {
        let writer = AsyncAuditWriter::new();
        let logs = writer.shutdown().expect("shutdown should succeed");
        assert!(logs.is_empty());
    }

    #[test]
    fn test_async_writer_double_shutdown_errors() {
        let writer = AsyncAuditWriter::new();
        let _ = writer.shutdown().unwrap();
        let result = writer.shutdown();
        assert!(result.is_err(), "double shutdown should error");
    }

    #[test]
    fn test_async_writer_default() {
        let writer = AsyncAuditWriter::default();
        writer.log(&ctx("SELECT 1", "u", 1)).unwrap();
        let logs = writer.shutdown().unwrap();
        assert_eq!(logs.len(), 1);
    }

    // ===== 审计日志查询过滤测试 =====

    #[test]
    fn test_audit_query_by_user() {
        let auditor = SqlAuditor::new();
        auditor.log(&ctx("SELECT 1", "alice", 100));
        auditor.log(&ctx("SELECT 2", "bob", 200));
        auditor.log(&ctx("SELECT 3", "alice", 300));
        let query = AuditQuery::new().by_user("alice");
        let results = query_logs(&auditor, &query);
        assert_eq!(results.len(), 2);
        assert!(results.iter().all(|r| r.user == "alice"));
    }

    #[test]
    fn test_audit_query_by_time_range() {
        let auditor = SqlAuditor::new();
        auditor.log(&ctx("SELECT 1", "u", 100));
        auditor.log(&ctx("SELECT 2", "u", 200));
        auditor.log(&ctx("SELECT 3", "u", 300));
        auditor.log(&ctx("SELECT 4", "u", 400));
        let query = AuditQuery::new().by_time_range(150, 350);
        let results = query_logs(&auditor, &query);
        assert_eq!(results.len(), 2);
        assert!(results
            .iter()
            .all(|r| r.timestamp >= 150 && r.timestamp <= 350));
    }

    #[test]
    fn test_audit_query_by_sql_contains() {
        let auditor = SqlAuditor::new();
        auditor.log(&ctx("SELECT * FROM users", "u", 1));
        auditor.log(&ctx("INSERT INTO orders", "u", 2));
        auditor.log(&ctx("SELECT * FROM orders", "u", 3));
        let query = AuditQuery::new().by_sql_contains("orders");
        let results = query_logs(&auditor, &query);
        assert_eq!(results.len(), 2);
        assert!(results
            .iter()
            .all(|r| r.sql.to_lowercase().contains("orders")));
    }

    #[test]
    fn test_audit_query_sql_contains_case_insensitive() {
        let auditor = SqlAuditor::new();
        auditor.log(&ctx("select * from Users", "u", 1));
        let query = AuditQuery::new().by_sql_contains("USERS");
        let results = query_logs(&auditor, &query);
        assert_eq!(results.len(), 1);
    }

    #[test]
    fn test_audit_query_with_limit() {
        let auditor = SqlAuditor::new();
        for i in 0..10 {
            auditor.log(&ctx(&format!("SELECT {}", i), "u", i));
        }
        let query = AuditQuery::new().with_limit(3);
        let results = query_logs(&auditor, &query);
        assert_eq!(results.len(), 3);
    }

    #[test]
    fn test_audit_query_combined_filters() {
        let auditor = SqlAuditor::new();
        auditor.log(&ctx("SELECT * FROM users", "alice", 100));
        auditor.log(&ctx("INSERT INTO users", "alice", 200));
        auditor.log(&ctx("SELECT * FROM orders", "alice", 300));
        auditor.log(&ctx("SELECT * FROM users", "bob", 400));
        let query = AuditQuery::new()
            .by_user("alice")
            .by_sql_contains("select")
            .with_limit(10);
        let results = query_logs(&auditor, &query);
        assert_eq!(results.len(), 2);
        assert!(results.iter().all(|r| r.user == "alice"));
    }

    #[test]
    fn test_audit_query_empty_returns_all() {
        let auditor = SqlAuditor::new();
        auditor.log(&ctx("SELECT 1", "u", 1));
        auditor.log(&ctx("SELECT 2", "u", 2));
        let query = AuditQuery::new();
        let results = query_logs(&auditor, &query);
        assert_eq!(results.len(), 2);
    }

    #[test]
    fn test_audit_query_no_match_returns_empty() {
        let auditor = SqlAuditor::new();
        auditor.log(&ctx("SELECT 1", "u", 1));
        let query = AuditQuery::new().by_user("nonexistent");
        let results = query_logs(&auditor, &query);
        assert!(results.is_empty());
    }

    #[test]
    fn test_audit_query_filter_directly() {
        let logs = vec![
            ctx("SELECT 1", "a", 10),
            ctx("SELECT 2", "b", 20),
            ctx("SELECT 3", "a", 30),
        ];
        let query = AuditQuery::new().by_user("a");
        let results = query.filter(&logs);
        assert_eq!(results.len(), 2);
    }

    #[test]
    fn test_audit_query_limit_zero_means_no_limit() {
        let logs = vec![ctx("SELECT 1", "a", 10), ctx("SELECT 2", "a", 20)];
        let query = AuditQuery::new().with_limit(0);
        let results = query.filter(&logs);
        assert_eq!(results.len(), 2);
    }

    // ===== 审计日志持久化存储测试 =====

    #[test]
    fn test_file_audit_log_store_append_and_read_all() {
        let path = test_data_dir().join("sz_orm_audit_store_append.jsonl");
        let path_str = path.to_str().unwrap();
        let store = FileAuditLogStore::new(path_str);
        // 清理可能残留的旧文件
        let _ = store.clear();

        store
            .append(&ctx("SELECT * FROM users", "alice", 1000))
            .unwrap();
        store
            .append(&ctx("INSERT INTO logs VALUES(1)", "bob", 2000))
            .unwrap();

        let logs = store.read_all().unwrap();
        assert_eq!(logs.len(), 2);
        assert_eq!(logs[0].user, "alice");
        assert_eq!(logs[0].sql, "SELECT * FROM users");
        assert_eq!(logs[1].user, "bob");
        assert_eq!(logs[1].timestamp, 2000);

        let _ = store.clear();
    }

    #[test]
    fn test_file_audit_log_store_masks_sensitive() {
        let path = test_data_dir().join("sz_orm_audit_store_mask.jsonl");
        let path_str = path.to_str().unwrap();
        let store = FileAuditLogStore::new(path_str);
        let _ = store.clear();

        store
            .append(&ctx(
                "SELECT * FROM users WHERE password='secret'",
                "admin",
                1000,
            ))
            .unwrap();

        let logs = store.read_all().unwrap();
        assert_eq!(logs.len(), 1);
        // 落盘内容应已脱敏
        assert!(!logs[0].sql.contains("password"));
        assert!(!logs[0].sql.contains("secret"));
        assert!(logs[0].sql.contains("******"));

        let _ = store.clear();
    }

    #[test]
    fn test_file_audit_log_store_clear_removes_entries() {
        let path = test_data_dir().join("sz_orm_audit_store_clear.jsonl");
        let path_str = path.to_str().unwrap();
        let store = FileAuditLogStore::new(path_str);
        let _ = store.clear();

        store.append(&ctx("SELECT 1", "u", 1)).unwrap();
        store.append(&ctx("SELECT 2", "u", 2)).unwrap();
        assert_eq!(store.read_all().unwrap().len(), 2);

        store.clear().unwrap();
        // 清空后读取应返回空
        assert_eq!(store.read_all().unwrap().len(), 0);

        let _ = store.clear();
    }

    #[test]
    fn test_file_audit_log_store_clear_nonexistent_is_ok() {
        // 清空不存在的文件不应报错
        let path = test_data_dir().join("sz_orm_audit_store_nonexistent.jsonl");
        let path_str = path.to_str().unwrap();
        let store = FileAuditLogStore::new(path_str);
        // 确保文件不存在
        let _ = std::fs::remove_file(path_str);
        assert!(store.clear().is_ok());
    }

    #[test]
    fn test_file_audit_log_store_read_all_empty_file() {
        let path = test_data_dir().join("sz_orm_audit_store_empty_read.jsonl");
        let path_str = path.to_str().unwrap();
        let store = FileAuditLogStore::new(path_str);
        let _ = store.clear();

        // 未写入任何内容，read_all 应返回空 vec（文件不存在视为空）
        let logs = store.read_all().unwrap();
        assert!(logs.is_empty());

        let _ = store.clear();
    }

    #[test]
    fn test_file_audit_log_store_skips_blank_lines() {
        let path = test_data_dir().join("sz_orm_audit_store_blank_lines.jsonl");
        let path_str = path.to_str().unwrap();
        let store = FileAuditLogStore::new(path_str);
        let _ = store.clear();

        store.append(&ctx("SELECT 1", "u", 1)).unwrap();
        // 手动追加空行模拟人工编辑或异常写入

        let mut file = std::fs::OpenOptions::new()
            .append(true)
            .open(path_str)
            .unwrap();
        use std::io::Write;
        writeln!(file).unwrap();
        writeln!(file, "   ").unwrap();
        drop(file);

        store.append(&ctx("SELECT 2", "u", 2)).unwrap();

        let logs = store.read_all().unwrap();
        // 空行应被跳过，只解析到 2 条有效日志
        assert_eq!(logs.len(), 2);

        let _ = store.clear();
    }

    #[test]
    fn test_file_audit_log_store_path_accessor() {
        let store = FileAuditLogStore::new("/tmp/sz_orm_audit_path_test.jsonl");
        assert_eq!(store.path(), "/tmp/sz_orm_audit_path_test.jsonl");
    }

    #[test]
    fn test_file_audit_log_store_concurrent_append() {
        use std::sync::Arc;
        let path = test_data_dir().join("sz_orm_audit_store_concurrent.jsonl");
        let path_str = path.to_str().unwrap();
        let store = Arc::new(FileAuditLogStore::new(path_str));
        let _ = store.clear();

        let mut handles = vec![];
        for i in 0..4 {
            let s = Arc::clone(&store);
            handles.push(std::thread::spawn(move || {
                for j in 0..10 {
                    s.append(&ctx(&format!("SELECT {}_{}", i, j), "u", j as i64))
                        .unwrap();
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }

        // 4 线程 × 10 条 = 40 条，全部应成功落盘
        let logs = store.read_all().unwrap();
        assert_eq!(logs.len(), 40);

        let _ = store.clear();
    }

    #[test]
    fn test_audit_log_store_trait_object() {
        // 验证 FileAuditLogStore 可作为 trait object 使用
        let path = test_data_dir().join("sz_orm_audit_store_trait.jsonl");
        let path_str = path.to_str().unwrap();
        let store: Box<dyn AuditLogStore> = Box::new(FileAuditLogStore::new(path_str));
        let _ = store.clear();

        store.append(&ctx("SELECT 1", "u", 1)).unwrap();
        let logs = store.read_all().unwrap();
        assert_eq!(logs.len(), 1);

        let _ = store.clear();
    }

    // ===== #11 修复：哈希链防篡改测试 =====

    #[test]
    fn test_hash_chain_empty_auditor_verify_ok() {
        let auditor = HashChainAuditor::new();
        assert!(auditor.is_empty());
        assert_eq!(auditor.len(), 0);
        // 空链应通过验证
        assert!(auditor.verify().is_ok());
    }

    #[test]
    fn test_hash_chain_single_entry_genesis() {
        let auditor = HashChainAuditor::new();
        auditor.log(&ctx("SELECT 1", "admin", 1000));
        assert_eq!(auditor.len(), 1);

        let entries = auditor.get_entries();
        // 首条记录的 prev_hash 必须为 GENESIS_HASH
        assert_eq!(entries[0].prev_hash, GENESIS_HASH);
        // current_hash 必须为 64 字符的十六进制串
        assert_eq!(entries[0].current_hash.len(), 64);
        // 验证链完整
        assert!(auditor.verify().is_ok());
    }

    #[test]
    fn test_hash_chain_multiple_entries_linked() {
        let auditor = HashChainAuditor::new();
        auditor.log(&ctx("SELECT 1", "admin", 1000));
        auditor.log(&ctx("SELECT 2", "admin", 1001));
        auditor.log(&ctx("SELECT 3", "admin", 1002));
        assert_eq!(auditor.len(), 3);

        let entries = auditor.get_entries();
        // 验证相邻记录的 prev_hash 链接
        assert_eq!(entries[1].prev_hash, entries[0].current_hash);
        assert_eq!(entries[2].prev_hash, entries[1].current_hash);
        // 验证链完整
        assert!(auditor.verify().is_ok());
    }

    #[test]
    fn test_hash_chain_detects_tampered_sql() {
        let auditor = HashChainAuditor::new();
        auditor.log(&ctx("SELECT 1", "admin", 1000));
        auditor.log(&ctx("SELECT 2", "admin", 1001));

        // 篡改第一条记录的 SQL（模拟攻击者修改历史日志）
        {
            let mut entries = auditor.entries.lock().unwrap();
            entries[0].entry.sql = "DROP TABLE users".to_string();
        }

        // 验证应失败
        let result = auditor.verify();
        assert!(result.is_err());
        let err = result.unwrap_err();
        // 错误信息应包含篡改位置
        assert!(err.contains("index 0"), "error: {}", err);
        assert!(err.contains("hash mismatch"), "error: {}", err);
    }

    #[test]
    fn test_hash_chain_detects_broken_link() {
        let auditor = HashChainAuditor::new();
        auditor.log(&ctx("SELECT 1", "admin", 1000));
        auditor.log(&ctx("SELECT 2", "admin", 1001));

        // 篡改第二条记录的 prev_hash（模拟删除中间记录）
        {
            let mut entries = auditor.entries.lock().unwrap();
            entries[1].prev_hash = "deadbeef".to_string();
        }

        let result = auditor.verify();
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(err.contains("chain broken at index 1"), "error: {}", err);
    }

    #[test]
    fn test_hash_chain_detects_genesis_tamper() {
        let auditor = HashChainAuditor::new();
        auditor.log(&ctx("SELECT 1", "admin", 1000));

        // 篡改首条记录的 prev_hash（模拟裁剪链首）
        {
            let mut entries = auditor.entries.lock().unwrap();
            entries[0].prev_hash = "deadbeef".to_string();
        }

        let result = auditor.verify();
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(err.contains("genesis prev_hash mismatch"), "error: {}", err);
    }

    #[test]
    fn test_hash_chain_masks_sensitive_data() {
        let auditor = HashChainAuditor::new();
        auditor.log(&ctx(
            "SELECT * FROM users WHERE password='secret'",
            "admin",
            1000,
        ));

        let entries = auditor.get_entries();
        // 链中存储的 SQL 应已被脱敏
        assert!(!entries[0].entry.sql.contains("password"));
        assert!(!entries[0].entry.sql.contains("secret"));
        assert!(entries[0].entry.sql.contains("******"));
        // 脱敏后的链仍应通过验证
        assert!(auditor.verify().is_ok());
    }

    #[test]
    fn test_hash_chain_deterministic_hashes() {
        // 相同输入应产生相同哈希（便于跨节点对账）
        let entry = ctx("SELECT 1", "admin", 1000);
        let e1 = HashChainEntry::genesis(entry.clone());
        let e2 = HashChainEntry::genesis(entry);
        assert_eq!(e1.current_hash, e2.current_hash);
        assert_eq!(e1.prev_hash, e2.prev_hash);
    }

    #[test]
    fn test_hash_chain_different_inputs_different_hashes() {
        let e1 = HashChainEntry::genesis(ctx("SELECT 1", "admin", 1000));
        let e2 = HashChainEntry::genesis(ctx("SELECT 2", "admin", 1000));
        assert_ne!(e1.current_hash, e2.current_hash);
    }

    #[test]
    fn test_hash_chain_flush_and_persist() {
        let auditor = HashChainAuditor::new();
        auditor.log(&ctx("SELECT 1", "admin", 1000));
        auditor.log(&ctx("SELECT 2", "admin", 1001));

        let path = test_data_dir().join("sz_orm_audit_hash_chain.json");
        let path_str = path.to_str().unwrap();
        let count = auditor.flush(path_str).unwrap();
        assert_eq!(count, 2);

        // 验证文件存在且非空
        let content = std::fs::read_to_string(path_str).unwrap();
        assert!(!content.is_empty());
        assert!(content.contains("current_hash"));

        let _ = std::fs::remove_file(path_str);
    }

    #[test]
    fn test_hash_chain_concurrent_log_thread_safe() {
        use std::sync::Arc;
        use std::thread;

        let auditor = Arc::new(HashChainAuditor::new());
        let mut handles = vec![];
        for i in 0..4 {
            let a = Arc::clone(&auditor);
            handles.push(thread::spawn(move || {
                for j in 0..25 {
                    a.log(&ctx(&format!("SELECT {}_{}", i, j), "u", j as i64));
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }

        // 4 线程 × 25 条 = 100 条
        assert_eq!(auditor.len(), 100);
        // 并发写入后链仍应完整
        assert!(auditor.verify().is_ok());
    }

    #[test]
    fn test_genesis_hash_constant_is_64_zeros() {
        // 验证 GENESIS_HASH 为 64 字符全零（SHA-256 输出长度）
        assert_eq!(GENESIS_HASH.len(), 64);
        assert!(GENESIS_HASH.chars().all(|c| c == '0'));
    }
}