diff --git a/agents/drivers/mongodb/src/main/java/com/dbx/agent/mongodb/MongoAgent.java b/agents/drivers/mongodb/src/main/java/com/dbx/agent/mongodb/MongoAgent.java index eee2af05ca..b5b883f90c 100644 --- a/agents/drivers/mongodb/src/main/java/com/dbx/agent/mongodb/MongoAgent.java +++ b/agents/drivers/mongodb/src/main/java/com/dbx/agent/mongodb/MongoAgent.java @@ -13,6 +13,7 @@ import com.mongodb.MongoClientSettings; import com.mongodb.ServerAddress; import com.mongodb.client.MongoClient; import com.mongodb.client.MongoClients; +import com.mongodb.client.model.UpdateOptions; import java.io.BufferedReader; import java.io.FileInputStream; import java.io.InputStream; @@ -625,15 +626,47 @@ public final class MongoAgent { String filterJson = params.get("filter_json").getAsString(); String updateJson = params.get("update_json").getAsString(); boolean many = params.get("many").getAsBoolean(); + String optionsJson = params.has("options_json") && !params.get("options_json").isJsonNull() + ? params.get("options_json").getAsString() + : null; var col = c.getDatabase(database).getCollection(collection); Document filter = documentForWrite(filterJson); Document update = documentForWrite(updateJson); requireBulkUpdateOperatorDocument(update); - var result = many ? col.updateMany(filter, update) : col.updateOne(filter, update); + UpdateOptions options = updateOptionsForWrite(optionsJson); + var result = many ? col.updateMany(filter, update, options) : col.updateOne(filter, update, options); return Collections.singletonMap("modified_count", result.getModifiedCount()); } + static UpdateOptions updateOptionsForWrite(String optionsJson) { + UpdateOptions result = new UpdateOptions(); + if (optionsJson == null || optionsJson.trim().isEmpty()) { + return result; + } + Document options = Document.parse(optionsJson); + for (String key : options.keySet()) { + if (!"arrayFilters".equals(key)) { + throw new IllegalArgumentException("Unsupported update option: " + key); + } + } + Object rawFilters = options.get("arrayFilters"); + if (rawFilters == null) { + return result; + } + if (!(rawFilters instanceof List)) { + throw new IllegalArgumentException("arrayFilters must be an array"); + } + List filters = new ArrayList<>(); + for (Object filter : (List) rawFilters) { + if (!(filter instanceof Document)) { + throw new IllegalArgumentException("Each arrayFilters entry must be an object"); + } + filters.add((Document) filter); + } + return result.arrayFilters(filters); + } + static Document documentForWrite(String docJson) { Document doc = Document.parse(docJson); convertMongoShellDates(doc); diff --git a/agents/drivers/mongodb/src/test/java/com/dbx/agent/mongodb/MongoAgentTest.java b/agents/drivers/mongodb/src/test/java/com/dbx/agent/mongodb/MongoAgentTest.java index 17d9f3fc9b..a90406cf71 100644 --- a/agents/drivers/mongodb/src/test/java/com/dbx/agent/mongodb/MongoAgentTest.java +++ b/agents/drivers/mongodb/src/test/java/com/dbx/agent/mongodb/MongoAgentTest.java @@ -12,6 +12,7 @@ import com.google.gson.JsonArray; import com.google.gson.JsonObject; import com.google.gson.JsonParser; import com.mongodb.MongoClientSettings; +import com.mongodb.client.model.UpdateOptions; import java.io.FileInputStream; import java.nio.file.Files; import java.nio.file.Path; @@ -202,6 +203,25 @@ class MongoAgentTest { assertFalse(json.getAsJsonObject("error").get("message").getAsString().contains("Unknown method")); } + @Test + void parsesArrayFiltersUpdateOption() { + UpdateOptions options = MongoAgent.updateOptionsForWrite( + "{\"arrayFilters\":[{\"item.id\":322678}]}" + ); + + assertEquals(1, options.getArrayFilters().size()); + assertEquals(322678, ((Document) options.getArrayFilters().get(0)).getInteger("item.id")); + } + + @Test + void rejectsUnsupportedUpdateOptions() { + IllegalArgumentException error = assertThrows( + IllegalArgumentException.class, + () -> MongoAgent.updateOptionsForWrite("{\"upsert\":true}") + ); + assertEquals("Unsupported update option: upsert", error.getMessage()); + } + @Test void deleteDocumentsMethodIsRecognizedOverJsonRpc() { String response = MongoAgent.handleRequest( diff --git a/apps/desktop/src/lib/backend/http.ts b/apps/desktop/src/lib/backend/http.ts index 4d15663cdc..2253f9702b 100644 --- a/apps/desktop/src/lib/backend/http.ts +++ b/apps/desktop/src/lib/backend/http.ts @@ -2098,8 +2098,8 @@ export async function documentUpdateDocument(connectionId: string, database: str return post("/api/document-store/update-document", { connectionId, database, collection, id, docJson, routing }); } -export async function mongoUpdateDocuments(connectionId: string, database: string, collection: string, filterJson: string, updateJson: string, many: boolean): Promise<{ affected_rows: number }> { - return post("/api/mongo/update-documents", { connectionId, database, collection, filterJson, updateJson, many }); +export async function mongoUpdateDocuments(connectionId: string, database: string, collection: string, filterJson: string, updateJson: string, many: boolean, optionsJson?: string): Promise<{ affected_rows: number }> { + return post("/api/mongo/update-documents", { connectionId, database, collection, filterJson, updateJson, many, optionsJson }); } export async function mongoDeleteDocument(connectionId: string, database: string, collection: string, id: string, routing?: string): Promise { diff --git a/apps/desktop/src/lib/backend/tauri.ts b/apps/desktop/src/lib/backend/tauri.ts index 40afa6e72b..bc11bc31cb 100644 --- a/apps/desktop/src/lib/backend/tauri.ts +++ b/apps/desktop/src/lib/backend/tauri.ts @@ -1869,7 +1869,7 @@ export async function documentUpdateDocument(connectionId: string, database: str return invoke("document_update_document", { connectionId, database, collection, id, docJson, routing }); } -export async function mongoUpdateDocuments(connectionId: string, database: string, collection: string, filterJson: string, updateJson: string, many: boolean): Promise<{ affected_rows: number }> { +export async function mongoUpdateDocuments(connectionId: string, database: string, collection: string, filterJson: string, updateJson: string, many: boolean, optionsJson?: string): Promise<{ affected_rows: number }> { const affectedRows = await invoke("mongo_update_documents", { connectionId, database, @@ -1877,6 +1877,7 @@ export async function mongoUpdateDocuments(connectionId: string, database: strin filterJson, updateJson, many, + optionsJson, }); return { affected_rows: affectedRows }; } diff --git a/apps/desktop/src/lib/mongo/mongoShellCommand.ts b/apps/desktop/src/lib/mongo/mongoShellCommand.ts index 9404d290ef..449411386f 100644 --- a/apps/desktop/src/lib/mongo/mongoShellCommand.ts +++ b/apps/desktop/src/lib/mongo/mongoShellCommand.ts @@ -50,7 +50,7 @@ export type MongoCommand = | ({ kind: "collectionStats" } & MongoCollectionStatsCommand) | ({ kind: "use" } & MongoUseCommand) | { kind: "insert"; collection: string; docsJson: string } - | { kind: "update"; collection: string; filter: string; update: string; many: boolean } + | { kind: "update"; collection: string; filter: string; update: string; options?: string; many: boolean } | { kind: "delete"; collection: string; filter: string; many: boolean } | { kind: "createIndex"; collection: string; keys: string; options?: string } | { kind: "dropIndex"; collection: string; index: string } @@ -293,11 +293,13 @@ export function parseMongoWriteCommand(input: string): MongoWriteCommand | null const target = parseCollectionMethodTarget(source, method); if (!target) continue; const args = parseMethodArgs(source, target.methodCallIndex); - if (!args || args.length !== 2) return null; + if (!args || args.length < 2 || args.length > 3) 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" }; + const options = args[2]?.trim() ? normalizeJsonArgument(args[2]) : undefined; + if (args[2]?.trim() && !options) return null; + return { kind: "update", collection: target.collection, filter, update, ...(options ? { options } : {}), many: method === "updateMany" }; } for (const method of ["deleteOne", "deleteMany"] as const) { diff --git a/apps/desktop/src/stores/queryStore.ts b/apps/desktop/src/stores/queryStore.ts index 71fb51deb2..a32896f35b 100644 --- a/apps/desktop/src/stores/queryStore.ts +++ b/apps/desktop/src/stores/queryStore.ts @@ -2732,7 +2732,7 @@ export const useQueryStore = defineStore("query", () => { const result = await api.mongoInsertDocuments(tab.connectionId, currentDatabase, mongoCommand.collection, mongoCommand.docsJson); allResults.push(markQueryResultRowsRaw(annotateMongoResult(mongoWriteToQueryResult(result.affected_rows, performance.now() - commandStartedAt)))); } else if (mongoCommand.kind === "update") { - const result = await api.mongoUpdateDocuments(tab.connectionId, currentDatabase, mongoCommand.collection, mongoCommand.filter, mongoCommand.update, mongoCommand.many); + const result = await api.mongoUpdateDocuments(tab.connectionId, currentDatabase, mongoCommand.collection, mongoCommand.filter, mongoCommand.update, mongoCommand.many, mongoCommand.options); allResults.push(markQueryResultRowsRaw(annotateMongoResult(mongoWriteToQueryResult(result.affected_rows, performance.now() - commandStartedAt)))); } else if (mongoCommand.kind === "createIndex") { const result = await api.mongoCreateIndex(tab.connectionId, currentDatabase, mongoCommand.collection, mongoCommand.keys, mongoCommand.options); diff --git a/crates/dbx-core/src/db/mongo_driver.rs b/crates/dbx-core/src/db/mongo_driver.rs index a3290ff1fc..474bf7bbe2 100644 --- a/crates/dbx-core/src/db/mongo_driver.rs +++ b/crates/dbx-core/src/db/mongo_driver.rs @@ -905,6 +905,7 @@ pub async fn update_documents( filter_json: &str, update_json: &str, many: bool, + options_json: Option<&str>, ) -> Result { let filter_value: serde_json::Value = serde_json::from_str(filter_json).map_err(|e| format!("Invalid filter JSON: {e}"))?; @@ -912,15 +913,47 @@ pub async fn update_documents( serde_json::from_str(update_json).map_err(|e| format!("Invalid update JSON: {e}"))?; let filter = json_filter_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 array_filters = parse_update_array_filters(options_json)?; let col = client.database(database).collection::(collection); let result = if many { - col.update_many(filter, update).await.map_err(|e| e.to_string())? + let mut action = col.update_many(filter, update); + if let Some(filters) = array_filters { + action = action.array_filters(filters); + } + action.await.map_err(|e| e.to_string())? } else { - col.update_one(filter, update).await.map_err(|e| e.to_string())? + let mut action = col.update_one(filter, update); + if let Some(filters) = array_filters { + action = action.array_filters(filters); + } + action.await.map_err(|e| e.to_string())? }; Ok(result.modified_count) } +#[derive(Default, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +struct MongoUpdateOptions { + array_filters: Option>, +} + +fn parse_update_array_filters(options_json: Option<&str>) -> Result>, String> { + let Some(raw) = options_json.filter(|value| !value.trim().is_empty()) else { + return Ok(None); + }; + let options: MongoUpdateOptions = serde_json::from_str(raw).map_err(|e| format!("Invalid update options: {e}"))?; + options + .array_filters + .map(|filters| { + filters + .iter() + .map(json_filter_to_document) + .collect::, _>>() + .map_err(|e| format!("Invalid arrayFilters: {e}")) + }) + .transpose() +} + pub async fn delete_document(client: &Client, database: &str, collection: &str, id: &str) -> Result { let col = client.database(database).collection::(collection); for filter in document_id_filters(id) { @@ -1248,6 +1281,23 @@ fn expand_object_id_string_array(items: &[serde_json::Value]) -> Bson { mod tests { use super::*; + #[test] + fn update_options_parse_array_filters() { + let filters = parse_update_array_filters(Some(r#"{"arrayFilters":[{"item.id":322678},{"item.active":true}]}"#)) + .unwrap() + .unwrap(); + + assert_eq!(filters, vec![doc! { "item.id": 322678_i64 }, doc! { "item.active": true }]); + } + + #[test] + fn update_options_reject_unsupported_fields() { + let error = parse_update_array_filters(Some(r#"{"upsert":true}"#)).unwrap_err(); + + assert!(error.starts_with("Invalid update options:")); + assert!(error.contains("unknown field `upsert`")); + } + #[test] fn multi_seed_uri_removes_direct_connection_true_before_driver_parse() { let uri = diff --git a/crates/dbx-core/src/mongo_ops.rs b/crates/dbx-core/src/mongo_ops.rs index e6ecea5194..28c3643308 100644 --- a/crates/dbx-core/src/mongo_ops.rs +++ b/crates/dbx-core/src/mongo_ops.rs @@ -315,12 +315,14 @@ pub async fn mongo_update_documents_core( filter_json: &str, update_json: &str, many: bool, + options_json: Option<&str>, ) -> Result { ensure_document_pool(state, connection_id).await?; 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 + mongo_driver::update_documents(client, database, collection, filter_json, update_json, many, options_json) + .await } PoolKind::Agent(client) => { let mut client = client.lock().await; @@ -331,6 +333,7 @@ pub async fn mongo_update_documents_core( "filter_json": filter_json, "update_json": update_json, "many": many, + "options_json": options_json, })) .await?; Ok(result.get("modified_count").and_then(|v| v.as_u64()).unwrap_or(0)) diff --git a/crates/dbx-web/src/routes/mongo.rs b/crates/dbx-web/src/routes/mongo.rs index bb09cfff06..5dd25c0cdb 100644 --- a/crates/dbx-web/src/routes/mongo.rs +++ b/crates/dbx-web/src/routes/mongo.rs @@ -154,6 +154,7 @@ pub struct MongoUpdateDocumentsRequest { pub filter_json: String, pub update_json: String, pub many: bool, + pub options_json: Option, } #[derive(Deserialize)] @@ -439,6 +440,7 @@ pub async fn update_documents( &req.filter_json, &req.update_json, req.many, + req.options_json.as_deref(), ) .await .map_err(AppError)?; diff --git a/packages/app-tests/mongoShellCommand.test.ts b/packages/app-tests/mongoShellCommand.test.ts index 878f1501a8..91f116639f 100644 --- a/packages/app-tests/mongoShellCommand.test.ts +++ b/packages/app-tests/mongoShellCommand.test.ts @@ -143,6 +143,27 @@ test("parseMongoWriteCommand accepts unquoted insert and update commands", () => }); }); +test("parseMongoWriteCommand accepts updateMany arrayFilters options", () => { + assert.deepEqual( + parseMongoWriteCommand(`db.issue_3231.updateMany( + { msgType: 3, "order.orderId": { $in: [12345] } }, + { $set: { "order.$[orderElem].bcorderproducts.$[prodElem].pankouType": "双双2" } }, + { arrayFilters: [ + { "orderElem.orderId": { $in: [12345] } }, + { "prodElem.id": 322678 } + ] } + )`), + { + kind: "update", + collection: "issue_3231", + filter: '{ "msgType": 3, "order.orderId": { "$in": [12345] } }', + update: '{ "$set": { "order.$[orderElem].bcorderproducts.$[prodElem].pankouType": "双双2" } }', + options: '{ "arrayFilters": [\n { "orderElem.orderId": { "$in": [12345] } },\n { "prodElem.id": 322678 }\n ] }', + many: true, + }, + ); +}); + test("parseMongoWriteCommand parses createIndex with optional options", () => { assert.deepEqual(parseMongoWriteCommand("db.users.createIndex({email: 1}, {unique: true, name: 'users_email_unique'})"), { kind: "createIndex", diff --git a/packages/node-core/src/database.ts b/packages/node-core/src/database.ts index d35ed107f8..55fac5ff24 100644 --- a/packages/node-core/src/database.ts +++ b/packages/node-core/src/database.ts @@ -1163,6 +1163,7 @@ async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCo filter_json: command.filter, update_json: command.update, many: command.many, + options_json: command.options, }); return { affectedRows: result.affected_rows }; } @@ -1316,7 +1317,7 @@ interface MongoCollectionStatsCommand { export type MongoWriteCommand = | { kind: "insert"; collection: string; docsJson: string } - | { kind: "update"; collection: string; filter: string; update: string; many: boolean } + | { kind: "update"; collection: string; filter: string; update: string; options?: string; many: boolean } | { kind: "delete"; collection: string; filter: string; many: boolean } | { kind: "createIndex"; collection: string; keys: string; options?: string } | { kind: "dropIndex"; collection: string; index: string } @@ -1465,11 +1466,13 @@ export function parseMongoWriteCommand(input: string): MongoWriteCommand | null const target = parseCollectionMethodTarget(source, method); if (!target) continue; const args = parseMethodArgs(source, target.methodCallIndex); - if (!args || args.length !== 2) return null; + if (!args || args.length < 2 || args.length > 3) 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" }; + const options = args[2]?.trim() ? normalizeJsonArgument(args[2]) : undefined; + if (args[2]?.trim() && !options) return null; + return { kind: "update", collection: target.collection, filter, update, ...(options ? { options } : {}), many: method === "updateMany" }; } for (const method of ["deleteOne", "deleteMany"] as const) { diff --git a/packages/node-core/src/web-backend.ts b/packages/node-core/src/web-backend.ts index 9a6c775368..e3968ee584 100644 --- a/packages/node-core/src/web-backend.ts +++ b/packages/node-core/src/web-backend.ts @@ -375,6 +375,7 @@ async function executeMongoWrite(config: ConnectionConfig, command: MongoWriteCo filterJson: command.filter, updateJson: command.update, many: command.many, + optionsJson: command.options, }), }); const result = (await res.json()) as { affected_rows: number }; diff --git a/packages/node-core/tests/mongo-query.test.ts b/packages/node-core/tests/mongo-query.test.ts index ee24ca63dc..a7d0cce049 100644 --- a/packages/node-core/tests/mongo-query.test.ts +++ b/packages/node-core/tests/mongo-query.test.ts @@ -73,6 +73,22 @@ test("parseMongoWriteCommand accepts unquoted update operator keys", () => { }); }); +test("parseMongoWriteCommand accepts updateMany arrayFilters options", () => { + assert.deepEqual( + parseMongoWriteCommand( + 'db.orders.updateMany({status: "open"}, {$set: {"items.$[item].status": "done"}}, {arrayFilters: [{"item.id": 7}]})', + ), + { + kind: "update", + collection: "orders", + filter: '{"status": "open"}', + update: '{"$set": {"items.$[item].status": "done"}}', + options: '{"arrayFilters": [{"item.id": 7}]}', + many: true, + }, + ); +}); + test("parseMongoCountDocumentsCommand accepts shell-style count commands", () => { assert.deepEqual(parseMongoCountDocumentsCommand('db.projects.countDocuments({"active":true})'), { collection: "projects", diff --git a/src-tauri/src/commands/mcp_bridge.rs b/src-tauri/src/commands/mcp_bridge.rs index 651668998e..ad885308f9 100644 --- a/src-tauri/src/commands/mcp_bridge.rs +++ b/src-tauri/src/commands/mcp_bridge.rs @@ -134,6 +134,7 @@ struct MongoUpdateDocumentsRequest { filter_json: String, update_json: String, many: bool, + options_json: Option, } #[derive(Deserialize)] @@ -722,6 +723,7 @@ async fn handle_mongo_update_documents_data(state: &Arc, body: &str, s &req.filter_json, &req.update_json, req.many, + req.options_json.as_deref(), ) .await { diff --git a/src-tauri/src/commands/mongo_cmd.rs b/src-tauri/src/commands/mongo_cmd.rs index 7ff560f0f4..b3f29a004a 100644 --- a/src-tauri/src/commands/mongo_cmd.rs +++ b/src-tauri/src/commands/mongo_cmd.rs @@ -263,6 +263,7 @@ pub async fn mongo_update_documents( filter_json: String, update_json: String, many: bool, + options_json: Option, ) -> Result { ensure_connection_writable(&state, &connection_id, "Update").await?; dbx_core::mongo_ops::mongo_update_documents_core( @@ -273,6 +274,7 @@ pub async fn mongo_update_documents( &filter_json, &update_json, many, + options_json.as_deref(), ) .await }