feat(history): add query history filters

crates/dbx-core/src/history.rs

@@ -27,6 +27,64 @@ pub struct HistoryEntry {
     pub details_json: Option<String>,
 }
 
+/// Matches current entries by connection ID and legacy entries by connection name.
+#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct HistoryConnectionFilter {
+    #[serde(default)]
+    pub connection_id: String,
+    #[serde(default)]
+    pub connection_name: String,
+}
+
+/// Includes connection identity so same-named databases do not match across connections.
+#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct HistoryDatabaseFilter {
+    #[serde(default)]
+    pub connection_id: String,
+    #[serde(default)]
+    pub connection_name: String,
+    #[serde(default)]
+    pub database: String,
+}
+
+/// Mirrors the descending (executed_at, id) order to paginate equal timestamps safely.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct HistoryCursor {
+    pub executed_at: String,
+    pub id: String,
+}
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct HistorySearchRequest {
+    #[serde(default)]
+    pub search_text: String,
+    #[serde(default)]
+    pub connections: Vec<HistoryConnectionFilter>,
+    #[serde(default)]
+    pub databases: Vec<HistoryDatabaseFilter>,
+    pub activity_kind: Option<String>,
+    pub success: Option<bool>,
+    pub started_at: Option<String>,
+    pub ended_at: Option<String>,
+    pub cursor: Option<HistoryCursor>,
+    #[serde(default)]
+    pub limit: usize,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct HistorySearchResult {
+    pub entries: Vec<HistoryEntry>,
+    pub next_cursor: Option<HistoryCursor>,
+    pub total: usize,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct HistoryConnectionOption {
+    pub connection_id: String,
+    pub connection_name: String,
+    pub databases: Vec<String>,
+}
+
 pub const MAX_HISTORY: usize = 1000;
 
 fn default_activity_kind() -> String {

crates/dbx-core/src/storage.rs

@@ -3,7 +3,7 @@ use std::path::Path;
 use std::time::{SystemTime, UNIX_EPOCH};
 
 use log::warn;
-use rusqlite::{params, params_from_iter, Connection, DatabaseName, OpenFlags, OptionalExtension, ToSql};
+use rusqlite::{params, params_from_iter, types::Value, Connection, DatabaseName, OpenFlags, OptionalExtension, ToSql};
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
@@ -14,7 +14,10 @@ use crate::connection_secrets::{
     NACOS_AUTH_SECRET_PREFIX,
 };
 use crate::db::sqlite::{connect_path_create_if_missing, SqliteHandle};
-use crate::history::{HistoryEntry, MAX_HISTORY};
+use crate::history::{
+    HistoryConnectionFilter, HistoryConnectionOption, HistoryCursor, HistoryDatabaseFilter, HistoryEntry,
+    HistorySearchRequest, HistorySearchResult, MAX_HISTORY,
+};
 use crate::models::connection::{ConnectionConfig, DatabaseConnectionInfo, DatabaseType, TransportLayerConfig};
 use crate::saved_sql::{SavedSqlFile, SavedSqlFolder, SavedSqlLibrary};
 
@@ -668,6 +671,127 @@ fn delete_secret_prefix_in_tx(
 
 // History
 
+fn history_filter_targets_connection(connection: &HistoryConnectionFilter, database: &HistoryDatabaseFilter) -> bool {
+    if !connection.connection_id.is_empty() || !database.connection_id.is_empty() {
+        !connection.connection_id.is_empty() && connection.connection_id == database.connection_id
+    } else {
+        !connection.connection_name.is_empty() && connection.connection_name == database.connection_name
+    }
+}
+
+fn append_history_scope_clause(
+    clauses: &mut Vec<String>,
+    values: &mut Vec<Value>,
+    connections: &[HistoryConnectionFilter],
+    databases: &[HistoryDatabaseFilter],
+) {
+    let mut alternatives = Vec::new();
+    for connection in connections {
+        // Database selections narrow only their owning connection; other selected connections remain whole scopes.
+        if databases
+            .iter()
+            .any(|database| !database.database.is_empty() && history_filter_targets_connection(connection, database))
+        {
+            continue;
+        }
+        if !connection.connection_id.is_empty() {
+            alternatives.push("connection_id = ?".to_string());
+            values.push(Value::Text(connection.connection_id.clone()));
+        } else if !connection.connection_name.is_empty() {
+            // Legacy JSON entries have no connection ID, so name fallback is limited to empty-ID rows.
+            alternatives.push("(connection_id = '' AND connection_name = ?)".to_string());
+            values.push(Value::Text(connection.connection_name.clone()));
+        }
+    }
+    for database in databases.iter().filter(|database| !database.database.is_empty()) {
+        if !database.connection_id.is_empty() {
+            alternatives.push("(connection_id = ? AND database = ?)".to_string());
+            values.push(Value::Text(database.connection_id.clone()));
+            values.push(Value::Text(database.database.clone()));
+        } else if !database.connection_name.is_empty() {
+            alternatives.push("(connection_id = '' AND connection_name = ? AND database = ?)".to_string());
+            values.push(Value::Text(database.connection_name.clone()));
+            values.push(Value::Text(database.database.clone()));
+        }
+    }
+    if !alternatives.is_empty() {
+        clauses.push(format!("({})", alternatives.join(" OR ")));
+    }
+}
+
+fn escape_history_like_pattern(value: &str) -> String {
+    value.replace('\\', "\\\\").replace('%', "\\%").replace('_', "\\_")
+}
+
+// Only fixed SQL fragments are assembled dynamically; every filter value remains parameter-bound.
+fn history_search_predicate(request: &HistorySearchRequest) -> (String, Vec<Value>) {
+    let mut clauses = Vec::new();
+    let mut values = Vec::new();
+    append_history_scope_clause(&mut clauses, &mut values, &request.connections, &request.databases);
+
+    if let Some(kind) = request.activity_kind.as_ref().filter(|kind| !kind.is_empty()) {
+        clauses.push("activity_kind = ?".to_string());
+        values.push(Value::Text(kind.clone()));
+    }
+    if let Some(success) = request.success {
+        clauses.push("success = ?".to_string());
+        values.push(Value::Integer(i64::from(success)));
+    }
+    if let Some(started_at) = request.started_at.as_ref().filter(|value| !value.is_empty()) {
+        clauses.push("julianday(executed_at) >= julianday(?)".to_string());
+        values.push(Value::Text(started_at.clone()));
+    }
+    if let Some(ended_at) = request.ended_at.as_ref().filter(|value| !value.is_empty()) {
+        clauses.push("julianday(executed_at) <= julianday(?)".to_string());
+        values.push(Value::Text(ended_at.clone()));
+    }
+
+    let search_text = request.search_text.trim();
+    if !search_text.is_empty() {
+        let pattern = format!("%{}%", escape_history_like_pattern(search_text));
+        let fields = ["sql_text", "connection_name", "database", "operation", "target"];
+        clauses.push(format!(
+            "({})",
+            fields
+                .iter()
+                .map(|field| format!("{field} LIKE ? ESCAPE '\\' COLLATE NOCASE"))
+                .collect::<Vec<_>>()
+                .join(" OR ")
+        ));
+        values.extend(fields.iter().map(|_| Value::Text(pattern.clone())));
+    }
+
+    let predicate = if clauses.is_empty() { String::new() } else { format!(" WHERE {}", clauses.join(" AND ")) };
+    (predicate, values)
+}
+
+fn map_history_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<HistoryEntry> {
+    Ok(HistoryEntry {
+        id: row.get(0)?,
+        connection_name: row.get(1)?,
+        database: row.get(2)?,
+        sql: row.get(3)?,
+        executed_at: row.get(4)?,
+        execution_time_ms: row.get::<_, i64>(5)? as u128,
+        success: row.get(6)?,
+        error: row.get(7)?,
+        activity_kind: {
+            let value: String = row.get(8)?;
+            if value.is_empty() {
+                "query".to_string()
+            } else {
+                value
+            }
+        },
+        connection_id: row.get(9)?,
+        operation: row.get(10)?,
+        target: row.get(11)?,
+        affected_rows: row.get(12)?,
+        rollback_sql: row.get(13)?,
+        details_json: row.get(14)?,
+    })
+}
+
 impl Storage {
     pub async fn save_history_entry(&self, entry: &HistoryEntry) -> Result<(), String> {
         let entry = entry.clone();
@@ -767,6 +891,96 @@ impl Storage {
         .await
     }
 
+    pub async fn search_history_entries(&self, request: HistorySearchRequest) -> Result<HistorySearchResult, String> {
+        self.with_conn(move |conn| {
+            let (predicate, values) = history_search_predicate(&request);
+            // Count before applying the cursor so the UI keeps the full filtered total.
+            let count_sql = format!("SELECT COUNT(*) FROM history{predicate}");
+            let total = conn
+                .query_row(&count_sql, params_from_iter(values.iter()), |row| row.get::<_, i64>(0))
+                .map_err(|error| error.to_string())? as usize;
+
+            let mut page_predicate = predicate;
+            let mut page_values = values;
+            // Keep this predicate aligned with ORDER BY to avoid skipping equal-timestamp rows.
+            if let Some(cursor) = &request.cursor {
+                let cursor_clause = "(executed_at < ? OR (executed_at = ? AND id < ?))";
+                if page_predicate.is_empty() {
+                    page_predicate = format!(" WHERE {cursor_clause}");
+                } else {
+                    page_predicate.push_str(" AND ");
+                    page_predicate.push_str(cursor_clause);
+                }
+                page_values.push(Value::Text(cursor.executed_at.clone()));
+                page_values.push(Value::Text(cursor.executed_at.clone()));
+                page_values.push(Value::Text(cursor.id.clone()));
+            }
+
+            let limit = if request.limit == 0 { 100 } else { request.limit.clamp(1, 200) };
+            let sql = format!(
+                "SELECT id, connection_name, database, sql_text, executed_at, execution_time_ms, success, \
+                 error, activity_kind, connection_id, operation, target, affected_rows, rollback_sql, details_json \
+                 FROM history{page_predicate} ORDER BY executed_at DESC, id DESC LIMIT ?"
+            );
+            page_values.push(Value::Integer((limit + 1) as i64));
+            let mut stmt = conn.prepare(&sql).map_err(|error| error.to_string())?;
+            let rows = stmt
+                .query_map(params_from_iter(page_values.iter()), map_history_row)
+                .map_err(|error| error.to_string())?;
+            let mut entries = rows.collect::<Result<Vec<_>, _>>().map_err(|error| error.to_string())?;
+            let has_more = entries.len() > limit;
+            entries.truncate(limit);
+            let next_cursor = if has_more {
+                entries
+                    .last()
+                    .map(|entry| HistoryCursor { executed_at: entry.executed_at.clone(), id: entry.id.clone() })
+            } else {
+                None
+            };
+
+            Ok(HistorySearchResult { entries, next_cursor, total })
+        })
+        .await
+    }
+
+    pub async fn load_history_connection_options(&self) -> Result<Vec<HistoryConnectionOption>, String> {
+        self.with_conn(|conn| {
+            let mut stmt = conn
+                .prepare(
+                    "SELECT connection_id, connection_name, database \
+                     FROM history ORDER BY executed_at DESC, id DESC",
+                )
+                .map_err(|error| error.to_string())?;
+            let rows = stmt
+                .query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?)))
+                .map_err(|error| error.to_string())?;
+
+            let mut options = Vec::<HistoryConnectionOption>::new();
+            let mut indexes = HashMap::<String, usize>::new();
+            for row in rows {
+                let (connection_id, connection_name, database) = row.map_err(|error| error.to_string())?;
+                let key = if connection_id.is_empty() {
+                    format!("legacy:{connection_name}")
+                } else {
+                    format!("id:{connection_id}")
+                };
+                let index = if let Some(index) = indexes.get(&key) {
+                    *index
+                } else {
+                    let index = options.len();
+                    indexes.insert(key, index);
+                    options.push(HistoryConnectionOption { connection_id, connection_name, databases: Vec::new() });
+                    index
+                };
+                if !database.is_empty() && !options[index].databases.contains(&database) {
+                    options[index].databases.push(database);
+                }
+            }
+            Ok(options)
+        })
+        .await
+    }
+
     pub async fn clear_history(&self) -> Result<(), String> {
         self.with_conn(|conn| conn.execute("DELETE FROM history", []).map(|_| ()).map_err(|e| e.to_string())).await
     }
@@ -3044,6 +3258,7 @@ mod tests {
     use crate::connection_secrets::{
         MQ_AUTH_PASSWORD_KEY, MQ_AUTH_TOKEN_KEY, MQ_TOKEN_SIGNING_KEY, NACOS_AUTH_PASSWORD_KEY,
     };
+    use crate::history::{HistoryConnectionFilter, HistoryDatabaseFilter, HistoryEntry, HistorySearchRequest};
     use crate::models::connection::{
         ConnectionConfig, DatabaseConnectionInfo, DatabaseType, SshTunnelConfig, TransportLayerConfig,
     };
@@ -3061,6 +3276,195 @@ mod tests {
         std::env::temp_dir().join(format!("dbx-storage-{name}-{}-{stamp}", std::process::id()))
     }
 
+    fn history_entry(
+        id: &str,
+        connection_id: &str,
+        connection_name: &str,
+        database: &str,
+        sql: &str,
+        executed_at: &str,
+        success: bool,
+    ) -> HistoryEntry {
+        HistoryEntry {
+            id: id.to_string(),
+            connection_id: connection_id.to_string(),
+            connection_name: connection_name.to_string(),
+            database: database.to_string(),
+            sql: sql.to_string(),
+            executed_at: executed_at.to_string(),
+            execution_time_ms: 10,
+            success,
+            error: (!success).then(|| "query failed".to_string()),
+            activity_kind: "query".to_string(),
+            operation: "SELECT".to_string(),
+            target: "orders".to_string(),
+            affected_rows: None,
+            rollback_sql: None,
+            details_json: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn history_search_filters_connection_database_and_legacy_entries() {
+        let path = temp_db_path("history-search-scope");
+        let storage = Storage::open(&path).await.unwrap();
+        let entries = [
+            history_entry("1", "conn-a", "Primary", "sales", "select 1", "2026-07-18T01:00:00Z", true),
+            history_entry("2", "conn-b", "Replica", "sales", "select 2", "2026-07-18T02:00:00Z", true),
+            history_entry("3", "", "Legacy", "archive", "select 3", "2026-07-18T03:00:00Z", true),
+        ];
+        for entry in &entries {
+            storage.save_history_entry(entry).await.unwrap();
+        }
+
+        let result = storage
+            .search_history_entries(HistorySearchRequest {
+                connections: vec![HistoryConnectionFilter {
+                    connection_id: "conn-a".to_string(),
+                    connection_name: "Primary".to_string(),
+                }],
+                databases: vec![HistoryDatabaseFilter {
+                    connection_id: "conn-a".to_string(),
+                    connection_name: "Primary".to_string(),
+                    database: "sales".to_string(),
+                }],
+                limit: 100,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        assert_eq!(result.entries.iter().map(|entry| entry.id.as_str()).collect::<Vec<_>>(), vec!["1"]);
+        assert_eq!(result.total, 1);
+
+        let legacy = storage
+            .search_history_entries(HistorySearchRequest {
+                connections: vec![HistoryConnectionFilter {
+                    connection_id: String::new(),
+                    connection_name: "Legacy".to_string(),
+                }],
+                limit: 100,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        assert_eq!(legacy.entries[0].id, "3");
+
+        let options = storage.load_history_connection_options().await.unwrap();
+        assert_eq!(options.len(), 3);
+        assert!(options.iter().any(|option| option.connection_id == "conn-a" && option.databases == ["sales"]));
+        assert!(options.iter().any(|option| option.connection_id.is_empty() && option.connection_name == "Legacy"));
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn history_search_combines_whole_connections_with_narrowed_database_scopes() {
+        let path = temp_db_path("history-search-hierarchical-scope");
+        let storage = Storage::open(&path).await.unwrap();
+        let entries = [
+            history_entry("a-sales", "conn-a", "Primary", "sales", "select 1", "2026-07-18T01:00:00Z", true),
+            history_entry("a-archive", "conn-a", "Primary", "archive", "select 2", "2026-07-18T02:00:00Z", true),
+            history_entry("b-sales", "conn-b", "Replica", "sales", "select 3", "2026-07-18T03:00:00Z", true),
+            history_entry("b-archive", "conn-b", "Replica", "archive", "select 4", "2026-07-18T04:00:00Z", true),
+        ];
+        for entry in &entries {
+            storage.save_history_entry(entry).await.unwrap();
+        }
+
+        let result = storage
+            .search_history_entries(HistorySearchRequest {
+                connections: vec![
+                    HistoryConnectionFilter {
+                        connection_id: "conn-a".to_string(),
+                        connection_name: "Primary".to_string(),
+                    },
+                    HistoryConnectionFilter {
+                        connection_id: "conn-b".to_string(),
+                        connection_name: "Replica".to_string(),
+                    },
+                ],
+                databases: vec![HistoryDatabaseFilter {
+                    connection_id: "conn-b".to_string(),
+                    connection_name: "Replica".to_string(),
+                    database: "sales".to_string(),
+                }],
+                limit: 100,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(
+            result.entries.iter().map(|entry| entry.id.as_str()).collect::<Vec<_>>(),
+            vec!["b-sales", "a-archive", "a-sales"]
+        );
+        assert_eq!(result.total, 3);
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn history_search_combines_text_status_and_time_filters() {
+        let path = temp_db_path("history-search-fields");
+        let storage = Storage::open(&path).await.unwrap();
+        let entries = [
+            history_entry("1", "conn", "Main", "app", "select 100% from orders", "2026-07-17T23:59:59Z", false),
+            history_entry("2", "conn", "Main", "app", "select 1000 from orders", "2026-07-18T12:00:00Z", false),
+            history_entry("3", "conn", "Main", "app", "select 100% from orders", "2026-07-18T12:00:00Z", true),
+        ];
+        for entry in &entries {
+            storage.save_history_entry(entry).await.unwrap();
+        }
+
+        let result = storage
+            .search_history_entries(HistorySearchRequest {
+                search_text: "100%".to_string(),
+                success: Some(false),
+                started_at: Some("2026-07-18T00:00:00Z".to_string()),
+                ended_at: Some("2026-07-18T23:59:59Z".to_string()),
+                limit: 100,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        assert!(result.entries.is_empty());
+
+        let successful = storage
+            .search_history_entries(HistorySearchRequest {
+                search_text: "100%".to_string(),
+                success: Some(true),
+                started_at: Some("2026-07-18T00:00:00Z".to_string()),
+                ended_at: Some("2026-07-18T23:59:59Z".to_string()),
+                limit: 100,
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        assert_eq!(successful.entries[0].id, "3");
+        let _ = std::fs::remove_file(path);
+    }
+
+    #[tokio::test]
+    async fn history_search_cursor_is_stable_for_equal_timestamps() {
+        let path = temp_db_path("history-search-cursor");
+        let storage = Storage::open(&path).await.unwrap();
+        for id in ["a", "b", "c"] {
+            storage
+                .save_history_entry(&history_entry(id, "conn", "Main", "app", "select 1", "2026-07-18T12:00:00Z", true))
+                .await
+                .unwrap();
+        }
+
+        let first =
+            storage.search_history_entries(HistorySearchRequest { limit: 2, ..Default::default() }).await.unwrap();
+        assert_eq!(first.entries.iter().map(|entry| entry.id.as_str()).collect::<Vec<_>>(), vec!["c", "b"]);
+        let second = storage
+            .search_history_entries(HistorySearchRequest { cursor: first.next_cursor, limit: 2, ..Default::default() })
+            .await
+            .unwrap();
+        assert_eq!(second.entries.iter().map(|entry| entry.id.as_str()).collect::<Vec<_>>(), vec!["a"]);
+        assert!(second.next_cursor.is_none());
+        let _ = std::fs::remove_file(path);
+    }
+
     fn ssh_profile(id: &str, password: &str) -> TransportLayerConfig {
         TransportLayerConfig::Ssh(SshTunnelConfig {
             id: id.to_string(),