From c56bcb8e2d6fa9e733e08111441753ab755706cb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BE=90=E6=96=87=E5=BD=AC?= <1473157698@qq.com> Date: Thu, 28 May 2026 18:58:26 +0800 Subject: [PATCH] =?UTF-8?q?feat(mongodb):=20=E6=94=AF=E6=8C=81=E8=81=9A?= =?UTF-8?q?=E5=90=88=E6=9F=A5=E8=AF=A2=E3=80=81=E6=89=B9=E9=87=8F=E5=86=99?= =?UTF-8?q?=E5=85=A5=E5=92=8C=20MCP=20=E9=9B=86=E6=88=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 桌面端：解析 db.<集合>.aggregate([...]) shell 语法，新增 mongoAggregateDocuments 接口（Tauri + HTTP）；mongo 入口统一补 ensureConnected；ObjectId(...) 自动 转换为 {"\$oid":"..."} 扩展 JSON 表示。 - dbx-core：新增 aggregate_documents、insert/update/delete_documents bulk 操作；扩展 JSON 解析识别 \$oid；schema list_tables_core 补 MongoDB/Elasticsearch 分支返回 COLLECTION/INDEX。 - dbx-web / Tauri 命令 / MCP bridge：注册 4 个新 mongo 路由 (aggregate/insert/update/delete documents)；MCP bridge 新增 mongo HTTP 入口 和 lazy 连接复用。 - node-core (MCP 后端)：mongodb 走专属 executeQuery / listTables / describeTable；read-only 默认 + DBX_MCP_ALLOW_WRITES / DBX_MCP_ALLOW_DANGEROUS_SQL 两级开关；aggregate 含 \$out/\$merge 走同样开关； describeTable 抽样 20 条文档推断字段类型，_id 自动标记主键。 - mcp-server：mongodb 跳过 SQL safety evaluator（mongo shell 与 SQL 语法 不兼容），改由 backend 做命令级 gating；附注释。 - AiAssistant 同时 emit replaceSql + executeSql 让编辑器和结果保持一致； App.vue ensureQueryTab 在无上下文时回退到连接默认库；aiMessageRender 识别 mongodb / mongo 代码块。 - Review 修复：合并两份重复的 evaluateMongo*Safety 实现（同时修复 web-backend 用字符串比较导致空 filter 检查可被空格绕过的 bug）；ObjectId 正则兼容 单/双引号；mongo_ops 错误文案 "MCP" 改为 "bulk"；补 aggregate 解析测试用例。 测试：pnpm test 911 通过；vue-tsc 通过；cargo check -p dbx-core 通过。 Co-Authored-By: Claude Opus 4.7 --- apps/desktop/src/App.vue | 2 +- .../src/components/editor/AiAssistant.vue | 1 + apps/desktop/src/lib/aiMessageRender.ts | 2 + apps/desktop/src/lib/api.ts | 1 + apps/desktop/src/lib/http.ts | 9 + apps/desktop/src/lib/mongoShellCommand.ts | 35 +- apps/desktop/src/lib/tauri.ts | 9 + apps/desktop/src/stores/queryStore.ts | 34 ++ crates/dbx-core/src/db/mongo_driver.rs | 133 +++++- crates/dbx-core/src/mongo_ops.rs | 69 +++ crates/dbx-core/src/schema.rs | 19 + crates/dbx-web/src/main.rs | 4 + crates/dbx-web/src/routes/mongo.rs | 106 +++++ packages/app-tests/mongoShellCommand.test.ts | 33 ++ packages/mcp-server/src/index.ts | 17 +- packages/mcp-server/tests/server.test.ts | 64 +++ packages/node-core/src/database.ts | 436 ++++++++++++++++++ packages/node-core/src/web-backend.ts | 133 ++++++ packages/node-core/tests/mongo-query.test.ts | 152 ++++++ src-tauri/src/commands/mcp_bridge.rs | 230 +++++++++ src-tauri/src/commands/mongo_cmd.rs | 12 + src-tauri/src/lib.rs | 1 + 22 files changed, 1492 insertions(+), 10 deletions(-) create mode 100644 packages/node-core/tests/mongo-query.test.ts diff --git a/apps/desktop/src/App.vue b/apps/desktop/src/App.vue index 31cc105cc4..ef35ddff23 100644 --- a/apps/desktop/src/App.vue +++ b/apps/desktop/src/App.vue @@ -622,7 +622,7 @@ function ensureQueryTab(): string { const tab = activeTab.value; if (tab && tab.mode === "query") return tab.id; const connId = connectionStore.activeConnectionId || connectionStore.connections[0]?.id || ""; - const db = tab?.database || ""; + const db = tab?.database || connectionStore.getConfig(connId)?.database || ""; return queryStore.createTab(connId, db, undefined, "query"); } diff --git a/apps/desktop/src/components/editor/AiAssistant.vue b/apps/desktop/src/components/editor/AiAssistant.vue index ec9db21be0..37725fcbd8 100644 --- a/apps/desktop/src/components/editor/AiAssistant.vue +++ b/apps/desktop/src/components/editor/AiAssistant.vue @@ -556,6 +556,7 @@ function applySql(code: string) { } function executeSql(code: string) { + emit("replaceSql", code); emit("executeSql", code); } diff --git a/apps/desktop/src/lib/aiMessageRender.ts b/apps/desktop/src/lib/aiMessageRender.ts index 7ab17c16f2..c7adb3155a 100644 --- a/apps/desktop/src/lib/aiMessageRender.ts +++ b/apps/desktop/src/lib/aiMessageRender.ts @@ -35,6 +35,8 @@ const SQL_LANGUAGES = new Map([ ["sqlite", "SQLITE"], ["tsql", "TSQL"], ["clickhouse", "CLICKHOUSE"], + ["mongodb", "MONGODB"], + ["mongo", "MONGODB"], ]); const SHELL_LANGUAGES = new Map([ ["bash", "BASH"], diff --git a/apps/desktop/src/lib/api.ts b/apps/desktop/src/lib/api.ts index f88c7f7ca5..6ac2af5396 100644 --- a/apps/desktop/src/lib/api.ts +++ b/apps/desktop/src/lib/api.ts @@ -208,6 +208,7 @@ export const redisLoadMore = forward("redisLoadMore"); export const mongoListDatabases = forward("mongoListDatabases"); export const mongoListCollections = forward("mongoListCollections"); export const mongoFindDocuments = forward("mongoFindDocuments"); +export const mongoAggregateDocuments = forward("mongoAggregateDocuments"); export const mongoInsertDocument = forward("mongoInsertDocument"); export const mongoUpdateDocument = forward("mongoUpdateDocument"); export const mongoDeleteDocument = forward("mongoDeleteDocument"); diff --git a/apps/desktop/src/lib/http.ts b/apps/desktop/src/lib/http.ts index 656adeec00..1f1436f195 100644 --- a/apps/desktop/src/lib/http.ts +++ b/apps/desktop/src/lib/http.ts @@ -1267,6 +1267,15 @@ export async function mongoFindDocuments( return post("/api/mongo/find-documents", { connectionId, database, collection, skip, limit, filter, sort }); } +export async function mongoAggregateDocuments( + connectionId: string, + database: string, + collection: string, + pipelineJson: string, +): Promise { + return post("/api/mongo/aggregate-documents", { connectionId, database, collection, pipelineJson }); +} + export async function mongoInsertDocument( connectionId: string, database: string, diff --git a/apps/desktop/src/lib/mongoShellCommand.ts b/apps/desktop/src/lib/mongoShellCommand.ts index c722adb24d..8fa8060e80 100644 --- a/apps/desktop/src/lib/mongoShellCommand.ts +++ b/apps/desktop/src/lib/mongoShellCommand.ts @@ -13,6 +13,11 @@ export interface MongoCountDocumentsCommand { filter: string; } +export interface MongoAggregateCommand { + collection: string; + pipeline: string; +} + const DEFAULT_LIMIT = 100; export function parseMongoFindCommand(input: string): MongoFindCommand | null { @@ -72,6 +77,31 @@ export function parseMongoCountDocumentsCommand(input: string): MongoCountDocume }; } +export function parseMongoAggregateCommand(input: string): MongoAggregateCommand | null { + const source = input.trim().replace(/;$/, "").trim(); + const target = parseCollectionMethodTarget(source, "aggregate"); + if (!target) return null; + + const openIndex = source.indexOf("(", target.methodCallIndex); + const closeIndex = findMatchingParen(source, openIndex); + if (closeIndex < 0 || source.slice(closeIndex + 1).trim()) return null; + + const args = splitTopLevel(source.slice(openIndex + 1, closeIndex)); + if (args.length !== 1) return null; + const pipeline = normalizeJsonArgument(args[0]); + if (!pipeline) return null; + try { + if (!Array.isArray(JSON.parse(pipeline))) return null; + } catch { + return null; + } + + return { + collection: target.collection, + pipeline, + }; +} + export function mongoDocumentsToQueryResult(documents: unknown[], executionTimeMs: number, total: number): QueryResult { const columns: string[] = []; @@ -146,9 +176,10 @@ function parseCollectionMethodTarget( function normalizeJsonArgument(value: string): string | null { const trimmed = value.trim(); if (!trimmed) return "{}"; + const preprocessed = trimmed.replace(/ObjectId\s*\(\s*["']([^"']+)["']\s*\)/g, '{"$oid":"$1"}'); try { - JSON.parse(trimmed); - return trimmed; + JSON.parse(preprocessed); + return preprocessed; } catch { return null; } diff --git a/apps/desktop/src/lib/tauri.ts b/apps/desktop/src/lib/tauri.ts index 7ebdbf9698..f3e0ed9051 100644 --- a/apps/desktop/src/lib/tauri.ts +++ b/apps/desktop/src/lib/tauri.ts @@ -1098,6 +1098,15 @@ export async function mongoFindDocuments( return invoke("mongo_find_documents", { connectionId, database, collection, skip, limit, filter, sort }); } +export async function mongoAggregateDocuments( + connectionId: string, + database: string, + collection: string, + pipelineJson: string, +): Promise { + return invoke("mongo_aggregate_documents", { connectionId, database, collection, pipelineJson }); +} + export async function mongoInsertDocument( connectionId: string, database: string, diff --git a/apps/desktop/src/stores/queryStore.ts b/apps/desktop/src/stores/queryStore.ts index 35c9c0f14a..f028ba1b8e 100644 --- a/apps/desktop/src/stores/queryStore.ts +++ b/apps/desktop/src/stores/queryStore.ts @@ -12,6 +12,7 @@ import { restoreOpenTabsState, serializeOpenTabs } from "@/lib/openTabsPersisten import { mongoCountToQueryResult, mongoDocumentsToQueryResult, + parseMongoAggregateCommand, parseMongoCountDocumentsCommand, parseMongoFindCommand, } from "@/lib/mongoShellCommand"; @@ -589,6 +590,7 @@ export const useQueryStore = defineStore("query", () => { } const mongoFind = conn?.db_type === "mongodb" ? parseMongoFindCommand(sql) : null; if (mongoFind) { + await connStore.ensureConnected(tab.connectionId); console.info("[DBX][executeTabSql:mongo-find:start]", { traceId, collection: mongoFind.collection }); const result = await api.mongoFindDocuments( tab.connectionId, @@ -621,6 +623,7 @@ export const useQueryStore = defineStore("query", () => { } const mongoCount = conn?.db_type === "mongodb" ? parseMongoCountDocumentsCommand(sql) : null; if (mongoCount) { + await connStore.ensureConnected(tab.connectionId); console.info("[DBX][executeTabSql:mongo-count:start]", { traceId, collection: mongoCount.collection }); const result = await api.mongoFindDocuments( tab.connectionId, @@ -650,6 +653,37 @@ export const useQueryStore = defineStore("query", () => { return; } + const mongoAggregate = conn?.db_type === "mongodb" ? parseMongoAggregateCommand(sql) : null; + if (mongoAggregate) { + await connStore.ensureConnected(tab.connectionId); + console.info("[DBX][executeTabSql:mongo-aggregate:start]", { traceId, collection: mongoAggregate.collection }); + const result = await api.mongoAggregateDocuments( + tab.connectionId, + tab.database, + mongoAggregate.collection, + mongoAggregate.pipeline, + ); + console.info("[DBX][executeTabSql:mongo-aggregate:done]", { + traceId, + rowCount: result.documents.length, + total: result.total, + elapsed: elapsed(), + }); + const current = tabs.value.find((t) => t.id === id); + if (current?.executionId === executionId) { + current.results = undefined; + current.activeResultIndex = undefined; + current.result = mongoDocumentsToQueryResult(result.documents, performance.now() - startedAt, result.total); + current.queryAnalysis = undefined; + current.querySourceColumns = undefined; + current.queryEditabilityReason = undefined; + current.tableMeta = undefined; + current.resultBaseSql = options?.resultBaseSql ?? sql; + current.resultSortedSql = options?.resultSortedSql; + } + return; + } + console.info("[DBX][executeTabSql:execute-multi:start]", { traceId, elapsed: elapsed() }); const executionOptions = { ...(typeof pageLimit === "number" diff --git a/crates/dbx-core/src/db/mongo_driver.rs b/crates/dbx-core/src/db/mongo_driver.rs index c8200d531e..bde3241b8f 100644 --- a/crates/dbx-core/src/db/mongo_driver.rs +++ b/crates/dbx-core/src/db/mongo_driver.rs @@ -1,5 +1,5 @@ use mongodb::{ - bson::{doc, Bson, Document}, + bson::{doc, oid::ObjectId, Bson, Document}, Client, }; use serde::{Deserialize, Serialize}; @@ -50,7 +50,7 @@ pub async fn find_documents( let filter_doc: Document = match filter { Some(f) if !f.trim().is_empty() => { let json: serde_json::Value = serde_json::from_str(f).map_err(|e| format!("Invalid filter JSON: {e}"))?; - mongodb::bson::to_document(&json).map_err(|e| format!("Invalid filter: {e}"))? + json_object_to_document(&json)? } _ => doc! {}, }; @@ -61,7 +61,7 @@ pub async fn find_documents( if let Some(s) = sort { if !s.trim().is_empty() { let json: serde_json::Value = serde_json::from_str(s).map_err(|e| format!("Invalid sort JSON: {e}"))?; - let sort_doc = mongodb::bson::to_document(&json).map_err(|e| format!("Invalid sort: {e}"))?; + let sort_doc = json_object_to_document(&json).map_err(|e| format!("Invalid sort: {e}"))?; find = find.sort(sort_doc); } } @@ -78,6 +78,30 @@ pub async fn find_documents( Ok(MongoDocumentResult { documents, total }) } +pub async fn aggregate_documents( + client: &Client, + database: &str, + collection: &str, + pipeline_json: &str, +) -> Result { + let json: serde_json::Value = + serde_json::from_str(pipeline_json).map_err(|e| format!("Invalid pipeline JSON: {e}"))?; + let pipeline_values = json.as_array().ok_or_else(|| "Aggregate pipeline must be a JSON array".to_string())?; + let pipeline = pipeline_values + .iter() + .map(|value| json_object_to_document(value).map_err(|e| format!("Invalid pipeline stage: {e}"))) + .collect::, String>>()?; + let col = client.database(database).collection::(collection); + let mut cursor = col.aggregate(pipeline).await.map_err(|e| e.to_string())?; + let mut documents = Vec::new(); + while cursor.advance().await.map_err(|e| e.to_string())? { + let doc = cursor.deserialize_current().map_err(|e| e.to_string())?; + documents.push(bson_to_json(&Bson::Document(doc))); + } + let total = documents.len() as u64; + Ok(MongoDocumentResult { documents, total }) +} + pub async fn insert_document( client: &Client, database: &str, @@ -90,6 +114,28 @@ pub async fn insert_document( Ok(format!("{}", result.inserted_id)) } +pub async fn insert_documents( + client: &Client, + database: &str, + collection: &str, + docs_json: &str, +) -> Result { + let json: serde_json::Value = serde_json::from_str(docs_json).map_err(|e| format!("Invalid JSON: {e}"))?; + let docs = match json { + serde_json::Value::Array(values) => values + .into_iter() + .map(|value| json_object_to_document(&value).map_err(|e| format!("Invalid document: {e}"))) + .collect::, String>>()?, + value => vec![json_object_to_document(&value).map_err(|e| format!("Invalid document: {e}"))?], + }; + if docs.is_empty() { + return Ok(0); + } + let col = client.database(database).collection::(collection); + let result = col.insert_many(docs).await.map_err(|e| e.to_string())?; + Ok(result.inserted_ids.len() as u64) +} + pub async fn update_document( client: &Client, database: &str, @@ -105,6 +151,29 @@ pub async fn update_document( Ok(result.modified_count) } +pub async fn update_documents( + client: &Client, + database: &str, + collection: &str, + filter_json: &str, + update_json: &str, + many: bool, +) -> Result { + let filter_value: serde_json::Value = + serde_json::from_str(filter_json).map_err(|e| format!("Invalid filter JSON: {e}"))?; + let update_value: serde_json::Value = + serde_json::from_str(update_json).map_err(|e| format!("Invalid update JSON: {e}"))?; + let filter = json_object_to_document(&filter_value).map_err(|e| format!("Invalid filter: {e}"))?; + let update = json_object_to_document(&update_value).map_err(|e| format!("Invalid update: {e}"))?; + let col = client.database(database).collection::(collection); + let result = if many { + col.update_many(filter, update).await.map_err(|e| e.to_string())? + } else { + col.update_one(filter, update).await.map_err(|e| e.to_string())? + }; + Ok(result.modified_count) +} + pub async fn delete_document(client: &Client, database: &str, collection: &str, id: &str) -> Result { let oid = mongodb::bson::oid::ObjectId::parse_str(id).map_err(|e| format!("Invalid ObjectId: {e}"))?; let col = client.database(database).collection::(collection); @@ -112,6 +181,25 @@ pub async fn delete_document(client: &Client, database: &str, collection: &str, Ok(result.deleted_count) } +pub async fn delete_documents( + client: &Client, + database: &str, + collection: &str, + filter_json: &str, + many: bool, +) -> Result { + let filter_value: serde_json::Value = + serde_json::from_str(filter_json).map_err(|e| format!("Invalid filter JSON: {e}"))?; + let filter = json_object_to_document(&filter_value).map_err(|e| format!("Invalid filter: {e}"))?; + let col = client.database(database).collection::(collection); + let result = if many { + col.delete_many(filter).await.map_err(|e| e.to_string())? + } else { + col.delete_one(filter).await.map_err(|e| e.to_string())? + }; + Ok(result.deleted_count) +} + fn bson_to_json(bson: &Bson) -> serde_json::Value { match bson { Bson::Double(v) => serde_json::json!(v), @@ -133,3 +221,42 @@ fn bson_to_json(bson: &Bson) -> serde_json::Value { _ => serde_json::Value::String(format!("{bson}")), } } + +/// Convert a `serde_json::Value` (JSON object) to a BSON `Document`, +/// handling MongoDB extended JSON conventions such as `{"$oid":"..."}`. +pub fn json_object_to_document(value: &serde_json::Value) -> Result { + match json_value_to_bson(value) { + Bson::Document(doc) => Ok(doc), + other => Err(format!("Expected a JSON object, got {other:?}")), + } +} + +fn json_value_to_bson(value: &serde_json::Value) -> Bson { + match value { + serde_json::Value::Null => Bson::Null, + serde_json::Value::Bool(b) => Bson::Boolean(*b), + serde_json::Value::Number(n) => { + if let Some(i) = n.as_i64() { + Bson::Int64(i) + } else if let Some(f) = n.as_f64() { + Bson::Double(f) + } else { + Bson::Null + } + } + serde_json::Value::String(s) => Bson::String(s.clone()), + serde_json::Value::Array(arr) => Bson::Array(arr.iter().map(json_value_to_bson).collect()), + serde_json::Value::Object(obj) => { + // Extended JSON: {"$oid":"..."} → BSON ObjectId + if obj.len() == 1 { + if let Some(serde_json::Value::String(hex)) = obj.get("$oid") { + if let Ok(oid) = ObjectId::parse_str(hex) { + return Bson::ObjectId(oid); + } + } + } + let doc: Document = obj.iter().map(|(k, v)| (k.clone(), json_value_to_bson(v))).collect(); + Bson::Document(doc) + } + } +} diff --git a/crates/dbx-core/src/mongo_ops.rs b/crates/dbx-core/src/mongo_ops.rs index 7afdc0bd4f..17c417afe6 100644 --- a/crates/dbx-core/src/mongo_ops.rs +++ b/crates/dbx-core/src/mongo_ops.rs @@ -71,6 +71,23 @@ pub async fn mongo_find_documents_core( } } +pub async fn mongo_aggregate_documents_core( + state: &AppState, + connection_id: &str, + database: &str, + collection: &str, + pipeline_json: &str, +) -> Result { + let connections = state.connections.read().await; + match connections.get(connection_id).ok_or("Not found")? { + PoolKind::MongoDb(client) => { + mongo_driver::aggregate_documents(client, database, collection, pipeline_json).await + } + PoolKind::Agent(_) => Err("MongoDB legacy agent does not support aggregate".to_string()), + _ => Err("Not a MongoDB connection".to_string()), + } +} + pub async fn mongo_insert_document_core( state: &AppState, connection_id: &str, @@ -101,6 +118,21 @@ pub async fn mongo_insert_document_core( } } +pub async fn mongo_insert_documents_core( + state: &AppState, + connection_id: &str, + database: &str, + collection: &str, + docs_json: &str, +) -> Result { + let connections = state.connections.read().await; + match connections.get(connection_id).ok_or("Not found")? { + PoolKind::MongoDb(client) => mongo_driver::insert_documents(client, database, collection, docs_json).await, + PoolKind::Agent(_) => Err("MongoDB legacy agent does not support bulk insertMany/insertOne writes".to_string()), + _ => Err("Not a MongoDB connection".to_string()), + } +} + pub async fn mongo_update_document_core( state: &AppState, connection_id: &str, @@ -133,6 +165,25 @@ pub async fn mongo_update_document_core( } } +pub async fn mongo_update_documents_core( + state: &AppState, + connection_id: &str, + database: &str, + collection: &str, + filter_json: &str, + update_json: &str, + many: bool, +) -> Result { + let connections = state.connections.read().await; + match connections.get(connection_id).ok_or("Not found")? { + PoolKind::MongoDb(client) => { + mongo_driver::update_documents(client, database, collection, filter_json, update_json, many).await + } + PoolKind::Agent(_) => Err("MongoDB legacy agent does not support bulk updateOne/updateMany writes".to_string()), + _ => Err("Not a MongoDB connection".to_string()), + } +} + pub async fn mongo_delete_document_core( state: &AppState, connection_id: &str, @@ -157,3 +208,21 @@ pub async fn mongo_delete_document_core( _ => Err("Not a MongoDB/Elasticsearch connection".to_string()), } } + +pub async fn mongo_delete_documents_core( + state: &AppState, + connection_id: &str, + database: &str, + collection: &str, + filter_json: &str, + many: bool, +) -> Result { + let connections = state.connections.read().await; + match connections.get(connection_id).ok_or("Not found")? { + PoolKind::MongoDb(client) => { + mongo_driver::delete_documents(client, database, collection, filter_json, many).await + } + PoolKind::Agent(_) => Err("MongoDB legacy agent does not support bulk deleteOne/deleteMany writes".to_string()), + _ => Err("Not a MongoDB connection".to_string()), + } +} diff --git a/crates/dbx-core/src/schema.rs b/crates/dbx-core/src/schema.rs index 8f614b89af..5f4cfdc1d2 100644 --- a/crates/dbx-core/src/schema.rs +++ b/crates/dbx-core/src/schema.rs @@ -386,10 +386,29 @@ pub async fn list_tables_core( PoolKind::Sqlite(p) => { db::sqlite::list_tables(p, schema).await.map(|tables| filter_table_infos(tables, filter, limit)) } + PoolKind::MongoDb(client) => db::mongo_driver::list_collections(client, database) + .await + .map(|names| collection_names_to_tables(names, "COLLECTION")) + .map(|tables| filter_table_infos(tables, filter, limit)), + PoolKind::Elasticsearch(client) => db::elasticsearch_driver::list_indices(client) + .await + .map(|names| collection_names_to_tables(names, "INDEX")) + .map(|tables| filter_table_infos(tables, filter, limit)), _ => Ok(vec![]), } } +fn collection_names_to_tables(names: Vec, table_type: &str) -> Vec { + names + .into_iter() + .map(|name| db::TableInfo { + name, + table_type: table_type.to_string(), + comment: None, + }) + .collect() +} + fn filter_table_infos(tables: Vec, filter: Option<&str>, limit: Option) -> Vec { let filter = filter.unwrap_or("").to_lowercase(); let limit = limit.unwrap_or(usize::MAX); diff --git a/crates/dbx-web/src/main.rs b/crates/dbx-web/src/main.rs index 7f9719d131..11e479150b 100644 --- a/crates/dbx-web/src/main.rs +++ b/crates/dbx-web/src/main.rs @@ -220,9 +220,13 @@ async fn main() { .route("/mongo/list-databases", post(routes::mongo::list_databases)) .route("/mongo/list-collections", post(routes::mongo::list_collections)) .route("/mongo/find-documents", post(routes::mongo::find_documents)) + .route("/mongo/aggregate-documents", post(routes::mongo::aggregate_documents)) .route("/mongo/insert-document", post(routes::mongo::insert_document)) + .route("/mongo/insert-documents", post(routes::mongo::insert_documents)) .route("/mongo/update-document", post(routes::mongo::update_document)) + .route("/mongo/update-documents", post(routes::mongo::update_documents)) .route("/mongo/delete-document", post(routes::mongo::delete_document)) + .route("/mongo/delete-documents", post(routes::mongo::delete_documents)) // History .route("/history", get(routes::history::load_history).delete(routes::history::clear_history)) .route("/history/save", post(routes::history::save_history)) diff --git a/crates/dbx-web/src/routes/mongo.rs b/crates/dbx-web/src/routes/mongo.rs index 1c169a0705..ad87a40aa4 100644 --- a/crates/dbx-web/src/routes/mongo.rs +++ b/crates/dbx-web/src/routes/mongo.rs @@ -32,6 +32,15 @@ pub struct MongoFindRequest { pub sort: Option, } +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MongoAggregateRequest { + pub connection_id: String, + pub database: String, + pub collection: String, + pub pipeline_json: String, +} + #[derive(Deserialize)] #[serde(rename_all = "camelCase")] pub struct MongoInsertRequest { @@ -41,6 +50,36 @@ pub struct MongoInsertRequest { pub doc_json: String, } +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MongoInsertDocumentsRequest { + pub connection_id: String, + pub database: String, + pub collection: String, + pub docs_json: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MongoUpdateDocumentsRequest { + pub connection_id: String, + pub database: String, + pub collection: String, + pub filter_json: String, + pub update_json: String, + pub many: bool, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MongoDeleteDocumentsRequest { + pub connection_id: String, + pub database: String, + pub collection: String, + pub filter_json: String, + pub many: bool, +} + #[derive(Deserialize)] #[serde(rename_all = "camelCase")] pub struct MongoUpdateRequest { @@ -98,6 +137,22 @@ pub async fn find_documents( Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) } +pub async fn aggregate_documents( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::mongo_ops::mongo_aggregate_documents_core( + &state.app, + &req.connection_id, + &req.database, + &req.collection, + &req.pipeline_json, + ) + .await + .map_err(AppError)?; + Ok(Json(serde_json::to_value(result).map_err(|e| AppError(e.to_string()))?)) +} + pub async fn insert_document( State(state): State>, Json(req): Json, @@ -114,6 +169,22 @@ pub async fn insert_document( Ok(Json(result)) } +pub async fn insert_documents( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::mongo_ops::mongo_insert_documents_core( + &state.app, + &req.connection_id, + &req.database, + &req.collection, + &req.docs_json, + ) + .await + .map_err(AppError)?; + Ok(Json(serde_json::json!({ "affected_rows": result }))) +} + pub async fn update_document( State(state): State>, Json(req): Json, @@ -131,6 +202,24 @@ pub async fn update_document( Ok(Json(result)) } +pub async fn update_documents( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::mongo_ops::mongo_update_documents_core( + &state.app, + &req.connection_id, + &req.database, + &req.collection, + &req.filter_json, + &req.update_json, + req.many, + ) + .await + .map_err(AppError)?; + Ok(Json(serde_json::json!({ "affected_rows": result }))) +} + pub async fn delete_document( State(state): State>, Json(req): Json, @@ -146,3 +235,20 @@ pub async fn delete_document( .map_err(AppError)?; Ok(Json(result)) } + +pub async fn delete_documents( + State(state): State>, + Json(req): Json, +) -> Result, AppError> { + let result = dbx_core::mongo_ops::mongo_delete_documents_core( + &state.app, + &req.connection_id, + &req.database, + &req.collection, + &req.filter_json, + req.many, + ) + .await + .map_err(AppError)?; + Ok(Json(serde_json::json!({ "affected_rows": result }))) +} diff --git a/packages/app-tests/mongoShellCommand.test.ts b/packages/app-tests/mongoShellCommand.test.ts index 673c4e3ada..d26aa77d4c 100644 --- a/packages/app-tests/mongoShellCommand.test.ts +++ b/packages/app-tests/mongoShellCommand.test.ts @@ -3,6 +3,7 @@ import test from "node:test"; import { mongoCountToQueryResult, mongoDocumentsToQueryResult, + parseMongoAggregateCommand, parseMongoCountDocumentsCommand, parseMongoFindCommand, } from "../../apps/desktop/src/lib/mongoShellCommand.ts"; @@ -43,6 +44,38 @@ test("parseMongoCountDocumentsCommand parses db collection countDocuments", () = }); }); +test("parseMongoAggregateCommand parses db collection aggregate", () => { + assert.deepEqual(parseMongoAggregateCommand('db.products.aggregate([{"$match":{"active":true}},{"$count":"total"}])'), { + collection: "products", + pipeline: '[{"$match":{"active":true}},{"$count":"total"}]', + }); +}); + +test("parseMongoAggregateCommand accepts an empty pipeline", () => { + assert.deepEqual(parseMongoAggregateCommand("db.products.aggregate([])"), { + collection: "products", + pipeline: "[]", + }); +}); + +test("parseMongoAggregateCommand rejects non-array pipelines and extra arguments", () => { + assert.equal(parseMongoAggregateCommand('db.products.aggregate({"$match":{}})'), null); + assert.equal(parseMongoAggregateCommand("db.products.aggregate([], {})"), null); + assert.equal(parseMongoAggregateCommand("db.products.aggregate([]).limit(10)"), null); +}); + +test("parseMongoAggregateCommand normalises ObjectId arguments with either quote style", () => { + const oid = "507f1f77bcf86cd799439011"; + for (const quote of ["\"", "'"]) { + const command = parseMongoAggregateCommand( + `db.orders.aggregate([{"$match":{"_id":ObjectId(${quote}${oid}${quote})}}])`, + ); + assert.ok(command, `quote=${quote} should parse`); + assert.equal(command.collection, "orders"); + assert.deepEqual(JSON.parse(command.pipeline), [{ "$match": { "_id": { "$oid": oid } } }]); + } +}); + test("mongoCountToQueryResult returns a single count row", () => { assert.deepEqual(mongoCountToQueryResult(42, 5), { columns: ["count"], diff --git a/packages/mcp-server/src/index.ts b/packages/mcp-server/src/index.ts index 6d613b04bc..cc556f8f48 100644 --- a/packages/mcp-server/src/index.ts +++ b/packages/mcp-server/src/index.ts @@ -102,8 +102,12 @@ export function createDbxMcpServer(backend: Backend, options: { isWebMode?: bool async ({ connection_name, database, sql }) => { const config = await backend.findConnection(connection_name); if (!config) return text(`Connection "${connection_name}" not found`); - const safety = evaluateSqlSafety(sql, sqlSafetyFromEnv()); - if (!safety.allowed) return text(`Query blocked: ${safety.reason}`); + if (config.db_type !== "mongodb") { + const safety = evaluateSqlSafety(sql, sqlSafetyFromEnv()); + if (!safety.allowed) return text(`Query blocked: ${safety.reason}`); + } + // MongoDB shell commands don't fit the SQL safety evaluator; the backend + // (node-core executeQuery) applies command-aware read/write gating. try { const result = await backend.executeQuery(withDatabase(config, database), sql); if (result.columns.length === 0) return text(`Query executed. ${result.row_count} row(s) affected.`); @@ -211,8 +215,13 @@ export function createDbxMcpServer(backend: Backend, options: { isWebMode?: bool database: z.string().optional().describe("Database name"), }, async ({ connection_name, sql, database }) => { - const safety = evaluateSqlSafety(sql, sqlSafetyFromEnv()); - if (!safety.allowed) return text(`Query blocked: ${safety.reason}`); + const config = await backend.findConnection(connection_name); + if (config?.db_type !== "mongodb") { + const safety = evaluateSqlSafety(sql, sqlSafetyFromEnv()); + if (!safety.allowed) return text(`Query blocked: ${safety.reason}`); + } + // MongoDB shell commands bypass the SQL safety evaluator; the desktop + // app's executor applies command-aware read/write gating. return bridgeRequest("/execute-query", { connection_name, sql, database }, "Query sent to DBX"); }, ); diff --git a/packages/mcp-server/tests/server.test.ts b/packages/mcp-server/tests/server.test.ts index b34beed888..90b7a1e1f9 100644 --- a/packages/mcp-server/tests/server.test.ts +++ b/packages/mcp-server/tests/server.test.ts @@ -53,3 +53,67 @@ test("execute query scopes the connection to the requested database", async () = assert.equal(usedDatabase, "stores_demo"); }); + +test("mongodb list tables returns collections from the selected database", async () => { + let usedDatabase = ""; + const mongoConnection: ConnectionConfig = { ...connection, db_type: "mongodb", database: "admin" }; + const scopedBackend: Backend = { + ...backend, + findConnection: async () => mongoConnection, + listTables: async (config) => { + usedDatabase = config.database || ""; + return [{ name: "projects", type: "COLLECTION" }]; + }, + }; + const server = createDbxMcpServer(scopedBackend, { isWebMode: true }); + + const result = await (server as any)._registeredTools.dbx_list_tables.handler({ + connection_name: "local", + database: "pystrument", + }); + + assert.equal(usedDatabase, "pystrument"); + assert.match(result.content[0].text, /projects/); + assert.match(result.content[0].text, /COLLECTION/); +}); + +test("mongodb describe table returns inferred document fields", async () => { + const mongoConnection: ConnectionConfig = { ...connection, db_type: "mongodb" }; + const scopedBackend: Backend = { + ...backend, + findConnection: async () => mongoConnection, + describeTable: async () => [ + { name: "_id", data_type: "object", is_nullable: false, column_default: null, is_primary_key: true, comment: null }, + { name: "name", data_type: "string", is_nullable: false, column_default: null, is_primary_key: false, comment: null }, + ], + }; + const server = createDbxMcpServer(scopedBackend, { isWebMode: true }); + + const result = await (server as any)._registeredTools.dbx_describe_table.handler({ + connection_name: "local", + database: "pystrument", + table: "projects", + }); + + assert.match(result.content[0].text, /_id \(PK\)/); + assert.match(result.content[0].text, /name/); +}); + +test("mongodb execute query formats shell-style find results", async () => { + const mongoConnection: ConnectionConfig = { ...connection, db_type: "mongodb" }; + const scopedBackend: Backend = { + ...backend, + findConnection: async () => mongoConnection, + executeQuery: async () => ({ columns: ["_id", "name"], rows: [{ _id: "1", name: "demo" }], row_count: 1 }), + }; + const server = createDbxMcpServer(scopedBackend, { isWebMode: true }); + + const result = await (server as any)._registeredTools.dbx_execute_query.handler({ + connection_name: "local", + database: "pystrument", + sql: "db.projects.find({}).limit(1)", + }); + + assert.match(result.content[0].text, /demo/); + assert.match(result.content[0].text, /1 row\(s\)/); +}); diff --git a/packages/node-core/src/database.ts b/packages/node-core/src/database.ts index 8c559e1b37..b1c0443231 100644 --- a/packages/node-core/src/database.ts +++ b/packages/node-core/src/database.ts @@ -272,6 +272,11 @@ interface BridgeColumnInfo { comment: string | null; } +interface MongoDocumentResult { + documents: unknown[]; + total: number; +} + function bridgeAppDataDir(): string { const home = homedir(); switch (platform()) { @@ -407,6 +412,35 @@ function sqliteQuery(config: ConnectionConfig, sql: string, options?: QueryOptio } export async function executeQuery(config: ConnectionConfig, sql: string, options?: QueryOptions): Promise { + if (config.db_type === "mongodb") { + const find = parseMongoFindCommand(sql); + if (find) { + const result = await withTimeout(mongoFindDocuments(config, find.collection, find.skip, find.limit, find.filter, find.sort), resolveTimeoutMs(options)); + return mongoDocumentsToQueryResult(result.documents.slice(0, resolveMaxRows(options)), result.total); + } + const count = parseMongoCountDocumentsCommand(sql); + if (count) { + const result = await withTimeout(mongoFindDocuments(config, count.collection, 0, 1, count.filter), resolveTimeoutMs(options)); + return { columns: ["count"], rows: [{ count: result.total }], row_count: 1 }; + } + const aggregate = parseMongoAggregateCommand(sql); + if (aggregate) { + const safety = evaluateMongoAggregateSafety(aggregate, sqlSafetyFromEnv()); + if (!safety.allowed) throw new Error(safety.reason); + const result = await withTimeout(mongoAggregateDocuments(config, aggregate.collection, aggregate.pipeline), resolveTimeoutMs(options)); + return mongoDocumentsToQueryResult(result.documents.slice(0, resolveMaxRows(options)), result.total); + } + const write = parseMongoWriteCommand(sql); + if (write) { + const safety = evaluateMongoWriteSafety(write, sqlSafetyFromEnv()); + if (!safety.allowed) throw new Error(safety.reason); + const affected = await withTimeout(executeMongoWrite(config, write), resolveTimeoutMs(options)); + return { columns: [], rows: [], row_count: affected }; + } + throw new Error( + "Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.projects.countDocuments({}), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})", + ); + } if (isDirectType(config.db_type)) { return query(config, sql, undefined, options); } @@ -419,6 +453,14 @@ export async function executeQuery(config: ConnectionConfig, sql: string, option } export async function listTables(config: ConnectionConfig, schema?: string): Promise { + if (config.db_type === "mongodb") { + const collections = await bridgeDataRequest("/data/mongo/list-collections", { + connection_name: config.name, + database: config.database || "", + schema: schema || "", + }); + return collections.map((name) => ({ name, type: "COLLECTION" })); + } if (config.db_type === "sqlite") { const result = await query( config, @@ -448,6 +490,10 @@ export async function listTables(config: ConnectionConfig, schema?: string): Pro } export async function describeTable(config: ConnectionConfig, table: string, schema?: string): Promise { + if (config.db_type === "mongodb") { + const result = await mongoFindDocuments(config, table, 0, 20, "{}"); + return inferMongoColumns(result.documents); + } if (config.db_type === "sqlite") { const result = await query(config, `PRAGMA table_info(${quoteSqliteIdentifier(table)})`); return result.rows.map((r) => ({ @@ -498,3 +544,393 @@ export async function describeTable(config: ConnectionConfig, table: string, sch comment: r.comment != null ? String(r.comment) : null, })); } + +async function mongoFindDocuments( + config: ConnectionConfig, + collection: string, + skip: number, + limit: number, + filter: string, + sort?: string, +): Promise { + return bridgeDataRequest("/data/mongo/find-documents", { + connection_name: config.name, + database: config.database || "", + collection, + skip, + limit, + filter, + sort, + }); +} + +async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCommand): Promise { + if (command.kind === "insert") { + const result = await bridgeDataRequest<{ affected_rows: number }>("/data/mongo/insert-documents", { + connection_name: config.name, + database: config.database || "", + collection: command.collection, + docs_json: command.docsJson, + }); + return result.affected_rows; + } + if (command.kind === "update") { + const result = await bridgeDataRequest<{ affected_rows: number }>("/data/mongo/update-documents", { + connection_name: config.name, + database: config.database || "", + collection: command.collection, + filter_json: command.filter, + update_json: command.update, + many: command.many, + }); + return result.affected_rows; + } + const result = await bridgeDataRequest<{ affected_rows: number }>("/data/mongo/delete-documents", { + connection_name: config.name, + database: config.database || "", + collection: command.collection, + filter_json: command.filter, + many: command.many, + }); + return result.affected_rows; +} + +async function mongoAggregateDocuments( + config: ConnectionConfig, + collection: string, + pipelineJson: string, +): Promise { + return bridgeDataRequest("/data/mongo/aggregate-documents", { + connection_name: config.name, + database: config.database || "", + collection, + pipeline_json: pipelineJson, + }); +} + +export function mongoDocumentsToQueryResult(documents: unknown[], total: number): QueryResult { + const columns: string[] = []; + for (const doc of documents) { + if (isRecord(doc)) { + for (const key of Object.keys(doc)) { + if (!columns.includes(key)) columns.push(key); + } + } else if (!columns.includes("value")) { + columns.push("value"); + } + } + const rows = documents.map((doc) => { + const row: Record = {}; + for (const column of columns) { + row[column] = isRecord(doc) ? toCellValue(doc[column]) : column === "value" ? toCellValue(doc) : null; + } + return row; + }); + return { columns, rows, row_count: rows.length }; +} + +export function inferMongoColumns(documents: unknown[]): ColumnInfo[] { + const columns = new Map; nullable: boolean }>(); + for (const doc of documents) { + if (!isRecord(doc)) { + const entry = columns.get("value") ?? { types: new Set(), nullable: false }; + entry.types.add(mongoTypeName(doc)); + columns.set("value", entry); + continue; + } + for (const [name, value] of Object.entries(doc)) { + const entry = columns.get(name) ?? { types: new Set(), nullable: false }; + entry.types.add(mongoTypeName(value)); + if (value === null || value === undefined) entry.nullable = true; + columns.set(name, entry); + } + } + return Array.from(columns.entries()).map(([name, entry]) => ({ + name, + data_type: Array.from(entry.types).sort().join(" | ") || "unknown", + is_nullable: entry.nullable, + column_default: null, + is_primary_key: name === "_id", + comment: null, + })); +} + +interface MongoFindCommand { + collection: string; + filter: string; + skip: number; + limit: number; + sort?: string; +} + +interface MongoCountDocumentsCommand { + collection: string; + filter: string; +} + +interface MongoAggregateCommand { + collection: string; + pipeline: string; +} + +export type MongoWriteCommand = + | { kind: "insert"; collection: string; docsJson: string } + | { kind: "update"; collection: string; filter: string; update: string; many: boolean } + | { kind: "delete"; collection: string; filter: string; many: boolean }; + +export function parseMongoFindCommand(input: string): MongoFindCommand | null { + const source = input.trim().replace(/;$/, "").trim(); + const target = parseCollectionMethodTarget(source, "find"); + if (!target) return null; + const findOpenIndex = source.indexOf("(", target.methodCallIndex); + const findCloseIndex = findMatchingParen(source, findOpenIndex); + if (findCloseIndex < 0) return null; + const findArgs = splitTopLevel(source.slice(findOpenIndex + 1, findCloseIndex)); + const filter = normalizeJsonArgument(findArgs[0] || "{}"); + if (!filter) return null; + const chain = source.slice(findCloseIndex + 1).trim(); + if (chain && !chain.startsWith(".")) return null; + const sortArg = readChainedCallArgument(chain, "sort"); + let sort: string | undefined; + if (sortArg !== undefined) { + const parsedSort = normalizeJsonArgument(sortArg); + if (!parsedSort) return null; + sort = parsedSort; + } + const skip = readChainedIntegerArgument(chain, "skip", 0); + const limit = readChainedIntegerArgument(chain, "limit", MAX_ROWS); + if (skip === null || limit === null) return null; + return { collection: target.collection, filter, skip, limit, sort }; +} + +export function parseMongoCountDocumentsCommand(input: string): MongoCountDocumentsCommand | null { + const source = input.trim().replace(/;$/, "").trim(); + const target = parseCollectionMethodTarget(source, "countDocuments"); + if (!target) return null; + const openIndex = source.indexOf("(", target.methodCallIndex); + const closeIndex = findMatchingParen(source, openIndex); + if (closeIndex < 0 || source.slice(closeIndex + 1).trim()) return null; + const args = splitTopLevel(source.slice(openIndex + 1, closeIndex)); + if (args.length > 1 && args.slice(1).some((arg) => arg.trim())) return null; + const filter = normalizeJsonArgument(args[0] || "{}"); + return filter ? { collection: target.collection, filter } : null; +} + +export function parseMongoAggregateCommand(input: string): MongoAggregateCommand | null { + const source = input.trim().replace(/;$/, "").trim(); + const target = parseCollectionMethodTarget(source, "aggregate"); + if (!target) return null; + const args = parseMethodArgs(source, target.methodCallIndex); + if (!args || args.length !== 1) return null; + const pipeline = normalizeJsonArgument(args[0]); + if (!pipeline) return null; + return Array.isArray(JSON.parse(pipeline)) ? { collection: target.collection, pipeline } : null; +} + +export function mongoAggregateWriteStage(pipelineJson: string): "$out" | "$merge" | null { + try { + const pipeline = JSON.parse(pipelineJson); + if (!Array.isArray(pipeline)) return null; + for (const stage of pipeline) { + if (!isRecord(stage)) continue; + if (Object.prototype.hasOwnProperty.call(stage, "$out")) return "$out"; + if (Object.prototype.hasOwnProperty.call(stage, "$merge")) return "$merge"; + } + return null; + } catch { + return null; + } +} + +export function parseMongoWriteCommand(input: string): MongoWriteCommand | null { + const source = input.trim().replace(/;$/, "").trim(); + const insertOne = parseCollectionMethodTarget(source, "insertOne"); + if (insertOne) { + const args = parseMethodArgs(source, insertOne.methodCallIndex); + if (!args || args.length !== 1) return null; + const doc = normalizeJsonArgument(args[0]); + return doc ? { kind: "insert", collection: insertOne.collection, docsJson: doc } : null; + } + + const insertMany = parseCollectionMethodTarget(source, "insertMany"); + if (insertMany) { + const args = parseMethodArgs(source, insertMany.methodCallIndex); + if (!args || args.length !== 1) return null; + const docs = normalizeJsonArgument(args[0]); + if (!docs) return null; + return Array.isArray(JSON.parse(docs)) ? { kind: "insert", collection: insertMany.collection, docsJson: docs } : null; + } + + for (const method of ["updateOne", "updateMany"] as const) { + const target = parseCollectionMethodTarget(source, method); + if (!target) continue; + const args = parseMethodArgs(source, target.methodCallIndex); + if (!args || args.length !== 2) return null; + const filter = normalizeJsonArgument(args[0]); + const update = normalizeJsonArgument(args[1]); + if (!filter || !update) return null; + return { kind: "update", collection: target.collection, filter, update, many: method === "updateMany" }; + } + + for (const method of ["deleteOne", "deleteMany"] as const) { + const target = parseCollectionMethodTarget(source, method); + if (!target) continue; + const args = parseMethodArgs(source, target.methodCallIndex); + if (!args || args.length !== 1) return null; + const filter = normalizeJsonArgument(args[0]); + if (!filter) return null; + return { kind: "delete", collection: target.collection, filter, many: method === "deleteMany" }; + } + + return null; +} + +export function evaluateMongoWriteSafety( + command: MongoWriteCommand, + options: { allowWrites?: boolean; allowDangerous?: boolean }, +): { allowed: boolean; reason?: string } { + if (!options.allowWrites) { + return { + allowed: false, + reason: "MCP MongoDB execution is read-only by default. Set DBX_MCP_ALLOW_WRITES=1 to allow write commands.", + }; + } + if (!options.allowDangerous && (command.kind === "update" || command.kind === "delete") && isEmptyJsonObject(command.filter)) { + return { + allowed: false, + reason: "MongoDB update/delete commands must include a non-empty filter unless DBX_MCP_ALLOW_DANGEROUS_SQL=1 is set.", + }; + } + return { allowed: true }; +} + +export function evaluateMongoAggregateSafety( + command: MongoAggregateCommand, + options: { allowWrites?: boolean; allowDangerous?: boolean }, +): { allowed: boolean; reason?: string } { + const writeStage = mongoAggregateWriteStage(command.pipeline); + if (!writeStage) return { allowed: true }; + if (!options.allowWrites) { + return { + allowed: false, + reason: `MongoDB aggregate stage "${writeStage}" writes data. Set DBX_MCP_ALLOW_WRITES=1 to allow write commands.`, + }; + } + if (!options.allowDangerous) { + return { + allowed: false, + reason: `MongoDB aggregate stage "${writeStage}" is dangerous. Set DBX_MCP_ALLOW_DANGEROUS_SQL=1 to allow it.`, + }; + } + return { allowed: true }; +} + +function parseCollectionMethodTarget(source: string, method: string): { collection: string; methodCallIndex: number } | null { + const direct = new RegExp(`^db\\.([A-Za-z_$][\\w$]*)\\.${method}\\s*\\(`).exec(source); + if (direct) return { collection: direct[1], methodCallIndex: source.indexOf(`.${method}`) }; + const quoted = new RegExp(`^db\\.getCollection\\(\\s*(['"])([^'"]+)\\1\\s*\\)\\.${method}\\s*\\(`).exec(source); + if (quoted) return { collection: quoted[2], methodCallIndex: source.indexOf(`.${method}`) }; + return null; +} + +function parseMethodArgs(source: string, methodCallIndex: number): string[] | null { + const openIndex = source.indexOf("(", methodCallIndex); + const closeIndex = findMatchingParen(source, openIndex); + if (closeIndex < 0 || source.slice(closeIndex + 1).trim()) return null; + return splitTopLevel(source.slice(openIndex + 1, closeIndex)); +} + +function readChainedCallArgument(chain: string, method: string): string | undefined { + const pattern = new RegExp(`\\.${method}\\s*\\(`, "g"); + const match = pattern.exec(chain); + if (!match) return undefined; + const openIndex = match.index + match[0].lastIndexOf("("); + const closeIndex = findMatchingParen(chain, openIndex); + return closeIndex < 0 ? undefined : chain.slice(openIndex + 1, closeIndex); +} + +function readChainedIntegerArgument(chain: string, method: string, fallback: number): number | null { + const arg = readChainedCallArgument(chain, method); + if (arg === undefined) return fallback; + if (!/^\d+$/.test(arg.trim())) return null; + return Number(arg.trim()); +} + +function normalizeJsonArgument(arg: string): string | null { + const value = (arg.trim() || "{}").replace(/ObjectId\s*\(\s*["']([^"']+)["']\s*\)/g, '{"$oid":"$1"}'); + try { + JSON.parse(value); + return value; + } catch { + return null; + } +} + +function isEmptyJsonObject(json: string): boolean { + try { + const parsed = JSON.parse(json); + return isRecord(parsed) && Object.keys(parsed).length === 0; + } catch { + return false; + } +} + +function splitTopLevel(source: string): string[] { + const parts: string[] = []; + let depth = 0; + let start = 0; + let quote: string | null = null; + for (let i = 0; i < source.length; i += 1) { + const ch = source[i]; + if (quote) { + if (ch === "\\" && i + 1 < source.length) i += 1; + else if (ch === quote) quote = null; + continue; + } + if (ch === "'" || ch === '"') quote = ch; + else if (ch === "{" || ch === "[" || ch === "(") depth += 1; + else if (ch === "}" || ch === "]" || ch === ")") depth -= 1; + else if (ch === "," && depth === 0) { + parts.push(source.slice(start, i)); + start = i + 1; + } + } + parts.push(source.slice(start)); + return parts; +} + +function findMatchingParen(source: string, openIndex: number): number { + if (openIndex < 0 || source[openIndex] !== "(") return -1; + let depth = 0; + let quote: string | null = null; + for (let i = openIndex; i < source.length; i += 1) { + const ch = source[i]; + if (quote) { + if (ch === "\\" && i + 1 < source.length) i += 1; + else if (ch === quote) quote = null; + continue; + } + if (ch === "'" || ch === '"') quote = ch; + else if (ch === "(") depth += 1; + else if (ch === ")") { + depth -= 1; + if (depth === 0) return i; + } + } + return -1; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function mongoTypeName(value: unknown): string { + if (value === null || value === undefined) return "null"; + if (Array.isArray(value)) return "array"; + if (isRecord(value)) return "object"; + return typeof value; +} + +function toCellValue(value: unknown): unknown { + return typeof value === "object" && value !== null ? JSON.stringify(value) : value; +} diff --git a/packages/node-core/src/web-backend.ts b/packages/node-core/src/web-backend.ts index d2eacf0b3d..7f48dd4427 100644 --- a/packages/node-core/src/web-backend.ts +++ b/packages/node-core/src/web-backend.ts @@ -1,5 +1,17 @@ import type { ConnectionConfig } from "./connections.js"; import type { TableInfo, ColumnInfo, QueryOptions, QueryResult } from "./database.js"; +import { + evaluateMongoAggregateSafety, + evaluateMongoWriteSafety, + inferMongoColumns, + mongoDocumentsToQueryResult, + parseMongoAggregateCommand, + parseMongoCountDocumentsCommand, + parseMongoFindCommand, + parseMongoWriteCommand, + type MongoWriteCommand, +} from "./database.js"; +import { sqlSafetyFromEnv } from "./sql-safety.js"; const baseUrl = process.env.DBX_WEB_URL!.replace(/\/+$/, ""); const password = process.env.DBX_WEB_PASSWORD || ""; @@ -86,6 +98,14 @@ async function ensureConnected(config: ConnectionConfig): Promise { export async function listTables(config: ConnectionConfig, schema?: string): Promise { await ensureConnected(config); + if (config.db_type === "mongodb") { + const res = await apiFetch("/api/mongo/list-collections", { + method: "POST", + body: JSON.stringify({ connectionId: config.id, database: config.database || "" }), + }); + const collections = (await res.json()) as string[]; + return collections.map((name) => ({ name, type: "COLLECTION" })); + } const params = new URLSearchParams({ connection_id: config.id, database: config.database || "", @@ -97,6 +117,14 @@ export async function listTables(config: ConnectionConfig, schema?: string): Pro export async function describeTable(config: ConnectionConfig, table: string, schema?: string): Promise { await ensureConnected(config); + if (config.db_type === "mongodb") { + const res = await apiFetch("/api/mongo/find-documents", { + method: "POST", + body: JSON.stringify({ connectionId: config.id, database: config.database || "", collection: table, skip: 0, limit: 20, filter: "{}" }), + }); + const result = (await res.json()) as { documents: unknown[]; total: number }; + return inferMongoColumns(result.documents); + } const params = new URLSearchParams({ connection_id: config.id, database: config.database || "", @@ -109,6 +137,67 @@ export async function describeTable(config: ConnectionConfig, table: string, sch export async function executeQuery(config: ConnectionConfig, sql: string, options?: QueryOptions): Promise { await ensureConnected(config); + if (config.db_type === "mongodb") { + const find = parseMongoFindCommand(sql); + if (find) { + const res = await apiFetch("/api/mongo/find-documents", { + method: "POST", + body: JSON.stringify({ + connectionId: config.id, + database: config.database || "", + collection: find.collection, + skip: find.skip, + limit: find.limit, + filter: find.filter, + sort: find.sort, + }), + }); + const result = (await res.json()) as { documents: unknown[]; total: number }; + return mongoDocumentsToQueryResult(result.documents.slice(0, options?.maxRows ?? result.documents.length), result.total); + } + const count = parseMongoCountDocumentsCommand(sql); + if (count) { + const res = await apiFetch("/api/mongo/find-documents", { + method: "POST", + body: JSON.stringify({ + connectionId: config.id, + database: config.database || "", + collection: count.collection, + skip: 0, + limit: 1, + filter: count.filter, + }), + }); + const result = (await res.json()) as { documents: unknown[]; total: number }; + return { columns: ["count"], rows: [{ count: result.total }], row_count: 1 }; + } + const aggregate = parseMongoAggregateCommand(sql); + if (aggregate) { + const safety = evaluateMongoAggregateSafety(aggregate, sqlSafetyFromEnv()); + if (!safety.allowed) throw new Error(safety.reason); + const res = await apiFetch("/api/mongo/aggregate-documents", { + method: "POST", + body: JSON.stringify({ + connectionId: config.id, + database: config.database || "", + collection: aggregate.collection, + pipelineJson: aggregate.pipeline, + }), + }); + const result = (await res.json()) as { documents: unknown[]; total: number }; + return mongoDocumentsToQueryResult(result.documents.slice(0, options?.maxRows ?? result.documents.length), result.total); + } + const write = parseMongoWriteCommand(sql); + if (write) { + const safety = evaluateMongoWriteSafety(write, sqlSafetyFromEnv()); + if (!safety.allowed) throw new Error(safety.reason); + const affected = await executeMongoWrite(config, write); + return { columns: [], rows: [], row_count: affected }; + } + throw new Error( + "Use MongoDB shell-style commands, for example: db.projects.find({}).limit(100), db.projects.countDocuments({}), db.projects.insertOne({...}), db.projects.updateOne({...}, {$set: {...}}), or db.projects.deleteOne({...})", + ); + } const res = await apiFetch("/api/query/execute", { method: "POST", body: JSON.stringify({ @@ -128,3 +217,47 @@ export async function executeQuery(config: ConnectionConfig, sql: string, option const limitedRows = rows.slice(0, options?.maxRows ?? rows.length); return { columns: data.columns, rows: limitedRows, row_count: limitedRows.length }; } + +async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCommand): Promise { + if (command.kind === "insert") { + const res = await apiFetch("/api/mongo/insert-documents", { + method: "POST", + body: JSON.stringify({ + connectionId: config.id, + database: config.database || "", + collection: command.collection, + docsJson: command.docsJson, + }), + }); + const result = (await res.json()) as { affected_rows: number }; + return result.affected_rows; + } + if (command.kind === "update") { + const res = await apiFetch("/api/mongo/update-documents", { + method: "POST", + body: JSON.stringify({ + connectionId: config.id, + database: config.database || "", + collection: command.collection, + filterJson: command.filter, + updateJson: command.update, + many: command.many, + }), + }); + const result = (await res.json()) as { affected_rows: number }; + return result.affected_rows; + } + const res = await apiFetch("/api/mongo/delete-documents", { + method: "POST", + body: JSON.stringify({ + connectionId: config.id, + database: config.database || "", + collection: command.collection, + filterJson: command.filter, + many: command.many, + }), + }); + const result = (await res.json()) as { affected_rows: number }; + return result.affected_rows; +} + diff --git a/packages/node-core/tests/mongo-query.test.ts b/packages/node-core/tests/mongo-query.test.ts new file mode 100644 index 0000000000..a43fa08b3b --- /dev/null +++ b/packages/node-core/tests/mongo-query.test.ts @@ -0,0 +1,152 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { + executeQuery, + inferMongoColumns, + mongoAggregateWriteStage, + mongoDocumentsToQueryResult, + parseMongoAggregateCommand, + parseMongoCountDocumentsCommand, + parseMongoFindCommand, + parseMongoWriteCommand, +} from "../src/database.js"; + +test("parseMongoFindCommand accepts shell-style find commands", () => { + assert.deepEqual(parseMongoFindCommand('db.getCollection("operation_logs").find({"level":"info"}).sort({"ts":-1}).skip(5).limit(10)'), { + collection: "operation_logs", + filter: '{"level":"info"}', + skip: 5, + limit: 10, + sort: '{"ts":-1}', + }); +}); + +test("parseMongoCountDocumentsCommand accepts shell-style count commands", () => { + assert.deepEqual(parseMongoCountDocumentsCommand('db.projects.countDocuments({"active":true})'), { + collection: "projects", + filter: '{"active":true}', + }); +}); + +test("parseMongoAggregateCommand accepts aggregate pipelines", () => { + assert.deepEqual(parseMongoAggregateCommand('db.projects.aggregate([{"$match":{"active":true}},{"$group":{"_id":"$owner","total":{"$sum":1}}}])'), { + collection: "projects", + pipeline: '[{"$match":{"active":true}},{"$group":{"_id":"$owner","total":{"$sum":1}}}]', + }); +}); + +test("mongoAggregateWriteStage detects write stages", () => { + assert.equal(mongoAggregateWriteStage('[{"$match":{"active":true}}]'), null); + assert.equal(mongoAggregateWriteStage('[{"$match":{}},{"$out":"projects_dump"}]'), "$out"); + assert.equal(mongoAggregateWriteStage('[{"$merge":{"into":"projects_dump"}}]'), "$merge"); +}); + +test("mongodb executeQuery blocks aggregate write stages without both env flags", async () => { + const oldAllowWrites = process.env.DBX_MCP_ALLOW_WRITES; + const oldAllowDangerous = process.env.DBX_MCP_ALLOW_DANGEROUS_SQL; + delete process.env.DBX_MCP_ALLOW_WRITES; + delete process.env.DBX_MCP_ALLOW_DANGEROUS_SQL; + const config = { + id: "mongo", + name: "mongo", + db_type: "mongodb", + host: "127.0.0.1", + port: 27017, + username: "", + password: "", + database: "app", + ssh_enabled: false, + ssl: false, + } as const; + + await assert.rejects(executeQuery(config, 'db.projects.aggregate([{"$out":"projects_dump"}])'), /DBX_MCP_ALLOW_WRITES=1/); + + process.env.DBX_MCP_ALLOW_WRITES = "1"; + await assert.rejects( + executeQuery(config, 'db.projects.aggregate([{"$merge":{"into":"projects_dump"}}])'), + /DBX_MCP_ALLOW_DANGEROUS_SQL=1/, + ); + + if (oldAllowWrites === undefined) delete process.env.DBX_MCP_ALLOW_WRITES; + else process.env.DBX_MCP_ALLOW_WRITES = oldAllowWrites; + if (oldAllowDangerous === undefined) delete process.env.DBX_MCP_ALLOW_DANGEROUS_SQL; + else process.env.DBX_MCP_ALLOW_DANGEROUS_SQL = oldAllowDangerous; +}); + +test("parseMongoWriteCommand accepts supported write commands", () => { + assert.deepEqual(parseMongoWriteCommand('db.projects.insertOne({"name":"demo"})'), { + kind: "insert", + collection: "projects", + docsJson: '{"name":"demo"}', + }); + assert.deepEqual(parseMongoWriteCommand('db.projects.updateOne({"_id":"1"},{"$set":{"name":"next"}})'), { + kind: "update", + collection: "projects", + filter: '{"_id":"1"}', + update: '{"$set":{"name":"next"}}', + many: false, + }); + assert.deepEqual(parseMongoWriteCommand('db.projects.deleteMany({"stale":true})'), { + kind: "delete", + collection: "projects", + filter: '{"stale":true}', + many: true, + }); +}); + +test("mongodb executeQuery blocks writes without the write env flag", async () => { + const oldAllowWrites = process.env.DBX_MCP_ALLOW_WRITES; + delete process.env.DBX_MCP_ALLOW_WRITES; + await assert.rejects( + executeQuery( + { + id: "mongo", + name: "mongo", + db_type: "mongodb", + host: "127.0.0.1", + port: 27017, + username: "", + password: "", + database: "app", + ssh_enabled: false, + ssl: false, + }, + 'db.projects.insertOne({"name":"demo"})', + ), + /DBX_MCP_ALLOW_WRITES=1/, + ); + if (oldAllowWrites === undefined) delete process.env.DBX_MCP_ALLOW_WRITES; + else process.env.DBX_MCP_ALLOW_WRITES = oldAllowWrites; +}); + +test("mongoDocumentsToQueryResult turns documents into rows", () => { + assert.deepEqual(mongoDocumentsToQueryResult([{ _id: "1", nested: { ok: true } }, { _id: "2", name: "demo" }], 2), { + columns: ["_id", "nested", "name"], + rows: [ + { _id: "1", nested: '{"ok":true}', name: undefined }, + { _id: "2", nested: undefined, name: "demo" }, + ], + row_count: 2, + }); +}); + +test("inferMongoColumns marks _id as primary and reports observed types", () => { + assert.deepEqual(inferMongoColumns([{ _id: "1", active: true }, { _id: "2", active: null }]), [ + { + name: "_id", + data_type: "string", + is_nullable: false, + column_default: null, + is_primary_key: true, + comment: null, + }, + { + name: "active", + data_type: "boolean | null", + is_nullable: true, + column_default: null, + is_primary_key: false, + comment: null, + }, + ]); +}); diff --git a/src-tauri/src/commands/mcp_bridge.rs b/src-tauri/src/commands/mcp_bridge.rs index cdc2f3de30..c01035e2b3 100644 --- a/src-tauri/src/commands/mcp_bridge.rs +++ b/src-tauri/src/commands/mcp_bridge.rs @@ -39,6 +39,52 @@ struct DescribeTableRequest { table: String, } +#[derive(Deserialize)] +struct MongoFindDocumentsRequest { + connection_name: String, + database: Option, + collection: String, + skip: Option, + limit: Option, + filter: Option, + sort: Option, +} + +#[derive(Deserialize)] +struct MongoAggregateDocumentsRequest { + connection_name: String, + database: Option, + collection: String, + pipeline_json: String, +} + +#[derive(Deserialize)] +struct MongoInsertDocumentsRequest { + connection_name: String, + database: Option, + collection: String, + docs_json: String, +} + +#[derive(Deserialize)] +struct MongoUpdateDocumentsRequest { + connection_name: String, + database: Option, + collection: String, + filter_json: String, + update_json: String, + many: bool, +} + +#[derive(Deserialize)] +struct MongoDeleteDocumentsRequest { + connection_name: String, + database: Option, + collection: String, + filter_json: String, + many: bool, +} + #[derive(Clone, Serialize)] pub struct McpOpenTableEvent { pub connection_id: String, @@ -92,6 +138,18 @@ pub fn start(app_handle: AppHandle, state: Arc) { handle_list_tables_data(&st, body, &mut stream).await; } else if first_line.starts_with("POST /data/describe-table") { handle_describe_table_data(&st, body, &mut stream).await; + } else if first_line.starts_with("POST /data/mongo/list-collections") { + handle_mongo_list_collections_data(&st, body, &mut stream).await; + } else if first_line.starts_with("POST /data/mongo/find-documents") { + handle_mongo_find_documents_data(&st, body, &mut stream).await; + } else if first_line.starts_with("POST /data/mongo/aggregate-documents") { + handle_mongo_aggregate_documents_data(&st, body, &mut stream).await; + } else if first_line.starts_with("POST /data/mongo/insert-documents") { + handle_mongo_insert_documents_data(&st, body, &mut stream).await; + } else if first_line.starts_with("POST /data/mongo/update-documents") { + handle_mongo_update_documents_data(&st, body, &mut stream).await; + } else if first_line.starts_with("POST /data/mongo/delete-documents") { + handle_mongo_delete_documents_data(&st, body, &mut stream).await; } else if first_line.starts_with("POST /data/execute-query") { handle_execute_query_data(&st, body, &mut stream).await; } else if first_line.starts_with("POST /execute-query") { @@ -148,6 +206,30 @@ async fn resolve_connection( Ok(config) } +async fn resolve_mongo_pool_key( + state: &Arc, + connection_name: &str, + database: Option, + stream: &mut tokio::net::TcpStream, +) -> Option<(String, String)> { + let config = match resolve_connection(state, connection_name).await { + Ok(c) => c, + Err(e) => { + respond_error(stream, "404 Not Found", &e).await; + return None; + } + }; + let database = database.unwrap_or_else(|| config.database.clone().unwrap_or_default()); + let pool_key = match state.get_or_create_pool(&config.id, Some(&database)).await { + Ok(key) => key, + Err(e) => { + respond_error(stream, "500 Internal Server Error", &e).await; + return None; + } + }; + Some((pool_key, database)) +} + async fn handle_open_table(app: &AppHandle, state: &Arc, body: &str, stream: &mut tokio::net::TcpStream) { let req: OpenTableRequest = match serde_json::from_str(body) { Ok(r) => r, @@ -251,6 +333,154 @@ async fn handle_describe_table_data(state: &Arc, body: &str, stream: & } } +async fn handle_mongo_list_collections_data(state: &Arc, body: &str, stream: &mut tokio::net::TcpStream) { + let req: ListTablesRequest = match serde_json::from_str(body) { + Ok(r) => r, + Err(_) => { + respond_error(stream, "400 Bad Request", "Invalid JSON").await; + return; + } + }; + let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await + else { + return; + }; + match dbx_core::mongo_ops::mongo_list_collections_core(state, &pool_key, &database).await { + Ok(collections) => respond_json(stream, &collections).await, + Err(e) => respond_error(stream, "500 Internal Server Error", &e).await, + } +} + +async fn handle_mongo_find_documents_data(state: &Arc, body: &str, stream: &mut tokio::net::TcpStream) { + let req: MongoFindDocumentsRequest = match serde_json::from_str(body) { + Ok(r) => r, + Err(_) => { + respond_error(stream, "400 Bad Request", "Invalid JSON").await; + return; + } + }; + let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await + else { + return; + }; + match dbx_core::mongo_ops::mongo_find_documents_core( + state, + &pool_key, + &database, + &req.collection, + req.skip.unwrap_or(0), + req.limit.unwrap_or(100), + req.filter.as_deref(), + req.sort.as_deref(), + ) + .await + { + Ok(result) => respond_json(stream, &result).await, + Err(e) => respond_error(stream, "500 Internal Server Error", &e).await, + } +} + +async fn handle_mongo_aggregate_documents_data(state: &Arc, body: &str, stream: &mut tokio::net::TcpStream) { + let req: MongoAggregateDocumentsRequest = match serde_json::from_str(body) { + Ok(r) => r, + Err(_) => { + respond_error(stream, "400 Bad Request", "Invalid JSON").await; + return; + } + }; + let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await + else { + return; + }; + match dbx_core::mongo_ops::mongo_aggregate_documents_core( + state, + &pool_key, + &database, + &req.collection, + &req.pipeline_json, + ) + .await + { + Ok(result) => respond_json(stream, &result).await, + Err(e) => respond_error(stream, "500 Internal Server Error", &e).await, + } +} + +async fn handle_mongo_insert_documents_data(state: &Arc, body: &str, stream: &mut tokio::net::TcpStream) { + let req: MongoInsertDocumentsRequest = match serde_json::from_str(body) { + Ok(r) => r, + Err(_) => { + respond_error(stream, "400 Bad Request", "Invalid JSON").await; + return; + } + }; + let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await + else { + return; + }; + match dbx_core::mongo_ops::mongo_insert_documents_core(state, &pool_key, &database, &req.collection, &req.docs_json) + .await + { + Ok(inserted) => respond_json(stream, &serde_json::json!({ "affected_rows": inserted })).await, + Err(e) => respond_error(stream, "500 Internal Server Error", &e).await, + } +} + +async fn handle_mongo_update_documents_data(state: &Arc, body: &str, stream: &mut tokio::net::TcpStream) { + let req: MongoUpdateDocumentsRequest = match serde_json::from_str(body) { + Ok(r) => r, + Err(_) => { + respond_error(stream, "400 Bad Request", "Invalid JSON").await; + return; + } + }; + let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await + else { + return; + }; + match dbx_core::mongo_ops::mongo_update_documents_core( + state, + &pool_key, + &database, + &req.collection, + &req.filter_json, + &req.update_json, + req.many, + ) + .await + { + Ok(modified) => respond_json(stream, &serde_json::json!({ "affected_rows": modified })).await, + Err(e) => respond_error(stream, "500 Internal Server Error", &e).await, + } +} + +async fn handle_mongo_delete_documents_data(state: &Arc, body: &str, stream: &mut tokio::net::TcpStream) { + let req: MongoDeleteDocumentsRequest = match serde_json::from_str(body) { + Ok(r) => r, + Err(_) => { + respond_error(stream, "400 Bad Request", "Invalid JSON").await; + return; + } + }; + let Some((pool_key, database)) = resolve_mongo_pool_key(state, &req.connection_name, req.database, stream).await + else { + return; + }; + match dbx_core::mongo_ops::mongo_delete_documents_core( + state, + &pool_key, + &database, + &req.collection, + &req.filter_json, + req.many, + ) + .await + { + Ok(deleted) => respond_json(stream, &serde_json::json!({ "affected_rows": deleted })).await, + Err(e) => respond_error(stream, "500 Internal Server Error", &e).await, + } +} + async fn handle_execute_query_data(state: &Arc, body: &str, stream: &mut tokio::net::TcpStream) { let req: ExecuteQueryRequest = match serde_json::from_str(body) { Ok(r) => r, diff --git a/src-tauri/src/commands/mongo_cmd.rs b/src-tauri/src/commands/mongo_cmd.rs index a79b144b9a..48e28e423a 100644 --- a/src-tauri/src/commands/mongo_cmd.rs +++ b/src-tauri/src/commands/mongo_cmd.rs @@ -45,6 +45,18 @@ pub async fn mongo_find_documents( .await } +#[tauri::command] +pub async fn mongo_aggregate_documents( + state: State<'_, Arc>, + connection_id: String, + database: String, + collection: String, + pipeline_json: String, +) -> Result { + dbx_core::mongo_ops::mongo_aggregate_documents_core(&state, &connection_id, &database, &collection, &pipeline_json) + .await +} + #[tauri::command] pub async fn mongo_insert_document( state: State<'_, Arc>, diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 58d5e02f91..4cb1ef85a4 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -370,6 +370,7 @@ pub fn run() { commands::mongo_cmd::mongo_list_databases, commands::mongo_cmd::mongo_list_collections, commands::mongo_cmd::mongo_find_documents, + commands::mongo_cmd::mongo_aggregate_documents, commands::mongo_cmd::mongo_insert_document, commands::mongo_cmd::mongo_update_document, commands::mongo_cmd::mongo_delete_document,