feat(mongodb): support replaceOne

This commit is contained in:
Loc Nguyen authored and GitHub committed 2026-09-16 12:18:54 +08:00
1 parent 72be4ad9eb
commit 9eafac706e
24 files changed
+496 -16

No files matched your search

@@ -2,7 +2,7 @@
"protocolVersion": 1,
"handshakeMethod": "handshake",
"handshakeResponseFields": ["protocolVersion", "agentProtocolVersion", "capabilities"],
"allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents"],
"allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "mongo_replace_document"],
"capabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl"],
"defaultSqlCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl"],
"commonMethods": [
@@ -42,6 +42,6 @@
"disconnect",
"shutdown"
],
"mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "delete_document", "delete_documents", "run_command"],
"mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "replace_document", "delete_document", "delete_documents", "run_command"],
"kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status", "etcd_compact", "etcd_defrag", "etcd_watch_start", "etcd_watch_poll", "etcd_watch_stop", "etcd_lease_list", "etcd_lease_get", "etcd_lease_grant", "etcd_lease_keepalive_once", "etcd_lease_revoke", "etcd_auth_user_list", "etcd_auth_user_get", "etcd_auth_user_add", "etcd_auth_user_delete", "etcd_auth_user_change_password", "etcd_auth_user_grant_role", "etcd_auth_user_revoke_role", "etcd_auth_role_list", "etcd_auth_role_get", "etcd_auth_role_add", "etcd_auth_role_delete", "etcd_auth_role_grant_permission", "etcd_auth_role_revoke_permission"]
}
@@ -2,7 +2,7 @@
"protocolVersion": 2,
"handshakeMethod": "handshake",
"handshakeResponseFields": ["protocolVersion", "agentProtocolVersion", "capabilities"],
"allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "multi_session", "structured_error_v1"],
"allCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "kv", "kv_ttl", "kv_cas", "kv_list_values", "kv_status", "kv_history", "etcd_compaction", "etcd_defrag", "etcd_watch", "etcd_lease", "etcd_auth", "mongo_drop_database", "mongo_clone_collection", "mongo_run_command", "mongo_insert_documents", "mongo_replace_document", "multi_session", "structured_error_v1"],
"capabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "multi_session"],
"defaultSqlCapabilities": ["connect", "test_connection", "metadata", "query", "paged_query", "transaction", "ddl", "multi_session", "structured_error_v1"],
"commonMethods": [
@@ -46,7 +46,7 @@
"disconnect",
"shutdown"
],
"mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "delete_document", "delete_documents", "run_command"],
"mongoLegacyMethods": ["list_databases", "list_collections", "find_documents", "find_one", "explain_find", "aggregate_documents", "find_documents_extended_json", "count_documents", "server_version", "create_index", "create_user", "drop_indexes", "drop_collection", "clone_collection", "drop_database", "insert_document", "insert_documents", "update_document", "update_documents", "replace_document", "delete_document", "delete_documents", "run_command"],
"kvMethods": ["kv_list_prefix", "kv_get", "kv_put", "kv_delete", "kv_rename", "kv_history", "kv_status", "etcd_compact", "etcd_defrag", "etcd_watch_start", "etcd_watch_poll", "etcd_watch_stop", "etcd_lease_list", "etcd_lease_get", "etcd_lease_grant", "etcd_lease_keepalive_once", "etcd_lease_revoke", "etcd_auth_user_list", "etcd_auth_user_get", "etcd_auth_user_add", "etcd_auth_user_delete", "etcd_auth_user_change_password", "etcd_auth_user_grant_role", "etcd_auth_user_revoke_role", "etcd_auth_role_list", "etcd_auth_role_get", "etcd_auth_role_add", "etcd_auth_role_delete", "etcd_auth_role_grant_permission", "etcd_auth_role_revoke_permission"],
"sessionField": "agentSessionId",
"cursorSessionField": "sessionId"
+19 -3
View File
@@ -1041,6 +1041,7 @@ pub enum AgentCapability {
MongoCloneCollection,
MongoRunCommand,
MongoInsertDocuments,
MongoReplaceDocument,
MultiSession,
StructuredErrorV1,
}
@@ -1123,7 +1124,7 @@ fn parse_agent_rpc_error_header(header: &str) -> (Option<i64>, String) {
}
impl AgentCapability {
pub const ALL: [Self; 24] = [
pub const ALL: [Self; 25] = [
Self::Connect,
Self::TestConnection,
Self::Metadata,
@@ -1146,6 +1147,7 @@ impl AgentCapability {
Self::MongoCloneCollection,
Self::MongoRunCommand,
Self::MongoInsertDocuments,
Self::MongoReplaceDocument,
Self::MultiSession,
Self::StructuredErrorV1,
];
@@ -1174,6 +1176,7 @@ impl AgentCapability {
Self::MongoCloneCollection => "mongo_clone_collection",
Self::MongoRunCommand => "mongo_run_command",
Self::MongoInsertDocuments => "mongo_insert_documents",
Self::MongoReplaceDocument => "mongo_replace_document",
Self::MultiSession => "multi_session",
Self::StructuredErrorV1 => "structured_error_v1",
}
@@ -1376,13 +1379,14 @@ pub enum MongoAgentMethod {
InsertDocuments,
UpdateDocument,
UpdateDocuments,
ReplaceDocument,
DeleteDocument,
DeleteDocuments,
RunCommand,
}
impl MongoAgentMethod {
pub const ALL: [Self; 22] = [
pub const ALL: [Self; 23] = [
Self::ListDatabases,
Self::ListCollections,
Self::FindDocuments,
@@ -1402,6 +1406,7 @@ impl MongoAgentMethod {
Self::InsertDocuments,
Self::UpdateDocument,
Self::UpdateDocuments,
Self::ReplaceDocument,
Self::DeleteDocument,
Self::DeleteDocuments,
Self::RunCommand,
@@ -1428,6 +1433,7 @@ impl MongoAgentMethod {
Self::InsertDocuments => "insert_documents",
Self::UpdateDocument => "update_document",
Self::UpdateDocuments => "update_documents",
Self::ReplaceDocument => "replace_document",
Self::DeleteDocument => "delete_document",
Self::DeleteDocuments => "delete_documents",
Self::RunCommand => "run_command",
@@ -2803,6 +2809,13 @@ impl AgentDriverClient {
self.call_mongo_method(MongoAgentMethod::UpdateDocuments, params).await
}
pub async fn mongo_replace_document<T: DeserializeOwned + Send + 'static>(
&mut self,
params: Value,
) -> Result<T, String> {
self.call_mongo_method(MongoAgentMethod::ReplaceDocument, params).await
}
pub async fn mongo_delete_document<T: DeserializeOwned + Send + 'static>(
&mut self,
params: Value,
@@ -2958,6 +2971,7 @@ pub fn agent_supports_capability(handshake: Option<&AgentHandshake>, capability:
| AgentCapability::MongoCloneCollection
| AgentCapability::MongoRunCommand
| AgentCapability::MongoInsertDocuments
| AgentCapability::MongoReplaceDocument
) {
return handshake.map(|value| value.supports(capability)).unwrap_or(false);
}
@@ -4991,9 +5005,10 @@ for line in sys.stdin:
assert_eq!(AgentCapability::MongoCloneCollection.as_str(), "mongo_clone_collection");
assert_eq!(AgentCapability::MongoRunCommand.as_str(), "mongo_run_command");
assert_eq!(AgentCapability::MongoInsertDocuments.as_str(), "mongo_insert_documents");
assert_eq!(AgentCapability::MongoReplaceDocument.as_str(), "mongo_replace_document");
assert_eq!(AgentCapability::MultiSession.as_str(), "multi_session");
assert_eq!(AgentCapability::StructuredErrorV1.as_str(), "structured_error_v1");
assert_eq!(AgentCapability::ALL.len(), 24);
assert_eq!(AgentCapability::ALL.len(), 25);
}
#[test]
@@ -5055,6 +5070,7 @@ for line in sys.stdin:
assert_eq!(MongoAgentMethod::InsertDocuments.as_str(), "insert_documents");
assert_eq!(MongoAgentMethod::UpdateDocument.as_str(), "update_document");
assert_eq!(MongoAgentMethod::UpdateDocuments.as_str(), "update_documents");
assert_eq!(MongoAgentMethod::ReplaceDocument.as_str(), "replace_document");
assert_eq!(MongoAgentMethod::DeleteDocument.as_str(), "delete_document");
assert_eq!(MongoAgentMethod::DeleteDocuments.as_str(), "delete_documents");
}
+44
View File
@@ -2120,6 +2120,50 @@ pub async fn update_documents(
Ok(result.modified_count)
}
/// replaceOne(filter, replacement[, { upsert }]): the whole document is swapped, so
/// unlike update there are no array filters to apply.
pub async fn replace_document(
client: &Client,
database: &str,
collection: &str,
filter_json: &str,
replacement_json: &str,
options_json: Option<&str>,
) -> Result<u64, String> {
let filter_value: serde_json::Value =
serde_json::from_str(filter_json).map_err(|e| format!("Invalid filter JSON: {e}"))?;
let replacement_value: serde_json::Value =
serde_json::from_str(replacement_json).map_err(|e| format!("Invalid replacement JSON: {e}"))?;
let filter = json_filter_to_document(&filter_value).map_err(|e| format!("Invalid filter: {e}"))?;
let replacement = json_object_to_document(&replacement_value).map_err(|e| format!("Invalid replacement: {e}"))?;
if let Some(operator) = replacement.keys().find(|key| key.starts_with('$')) {
return Err(format!("Replacement document must not contain update operators such as {operator}"));
}
let upsert = parse_replace_options(options_json)?;
let col = client.database(database).collection::<Document>(collection);
let mut action = col.replace_one(filter, replacement);
if let Some(upsert) = upsert {
action = action.upsert(upsert);
}
let result = action.await.map_err(|e| e.to_string())?;
Ok(result.modified_count)
}
#[derive(Default, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct MongoReplaceOptions {
upsert: Option<bool>,
}
fn parse_replace_options(options_json: Option<&str>) -> Result<Option<bool>, String> {
let Some(raw) = options_json.filter(|value| !value.trim().is_empty()) else {
return Ok(None);
};
let options: MongoReplaceOptions =
serde_json::from_str(raw).map_err(|e| format!("Invalid replace options: {e}"))?;
Ok(options.upsert)
}
#[derive(Default, Deserialize)]
#[serde(rename_all = "camelCase")]
struct MongoUpdateOptions {
+52
View File
@@ -804,6 +804,45 @@ pub async fn mongo_update_documents_core(
}
}
pub async fn mongo_replace_document_core(
state: &AppState,
connection_id: &str,
database: &str,
collection: &str,
filter_json: &str,
replacement_json: &str,
options_json: Option<&str>,
) -> Result<u64, String> {
ensure_document_pool(state, connection_id).await?;
let pool = state.pool_handle(connection_id).await.ok_or("Not found")?;
match &pool {
PoolKind::MongoDb(client) => {
mongo_driver::replace_document(client, database, collection, filter_json, replacement_json, options_json)
.await
}
PoolKind::Agent(client) => {
let mut client = client.lock().await;
if !client.supports_capability(AgentCapability::MongoReplaceDocument) {
return Err(
"MongoDB Legacy Agent does not support replaceOne; upgrade or reinstall the MongoDB Legacy driver"
.to_string(),
);
}
let result: serde_json::Value = client
.mongo_replace_document(serde_json::json!({
"database": database,
"collection": collection,
"filter_json": filter_json,
"replacement_json": replacement_json,
"options_json": options_json,
}))
.await?;
Ok(result.get("modified_count").and_then(|v| v.as_u64()).unwrap_or(0))
}
_ => Err("Not a MongoDB connection".to_string()),
}
}
pub async fn mongo_delete_document_core(
state: &AppState,
connection_id: &str,
@@ -1031,6 +1070,19 @@ pub async fn execute_mongo_command_core(
let affected = mongo_insert_documents_core(state, connection_id, database, collection, documents).await?;
Ok(affected_query_result(affected))
}
MongoCommand::Replace { collection, filter, replacement, options } => {
let affected = mongo_replace_document_core(
state,
connection_id,
database,
collection,
filter,
replacement,
options.as_deref(),
)
.await?;
Ok(affected_query_result(affected))
}
MongoCommand::Update { collection, filter, update, options, many } => {
let affected = mongo_update_documents_core(
state,
+79
View File
@@ -66,6 +66,8 @@ pub enum MongoCommand {
},
#[serde(rename = "update")]
Update { collection: String, filter: String, update: String, options: Option<String>, many: bool },
#[serde(rename = "replace")]
Replace { collection: String, filter: String, replacement: String, options: Option<String> },
#[serde(rename = "delete")]
Delete { collection: String, filter: String, many: bool },
#[serde(rename = "createIndex")]
@@ -98,6 +100,7 @@ impl MongoCommand {
| Self::CreateUser { .. }
| Self::Insert { .. }
| Self::Update { .. }
| Self::Replace { .. }
| Self::Delete { .. }
| Self::CreateIndex { .. }
| Self::DropIndexes { .. }
@@ -117,6 +120,7 @@ impl MongoCommand {
pub fn has_empty_filter(&self) -> bool {
match self {
Self::Update { filter, .. }
| Self::Replace { filter, .. }
| Self::Delete { filter, .. }
| Self::FindOneAndUpdate { filter, .. }
| Self::FindOneAndReplace { filter, .. }
@@ -128,6 +132,7 @@ impl MongoCommand {
pub fn has_effectively_unbounded_filter(&self) -> bool {
match self {
Self::Update { filter, .. }
| Self::Replace { filter, .. }
| Self::Delete { filter, .. }
| Self::FindOneAndUpdate { filter, .. }
| Self::FindOneAndReplace { filter, .. }
@@ -609,6 +614,22 @@ pub fn parse(input: &str) -> Result<MongoCommand, String> {
return Ok(MongoCommand::Insert { collection, documents });
}
if let Some((args, tail)) = method_call(source, prefix_end, "replaceOne") {
if !tail.is_empty() || !(2..=3).contains(&args.len()) {
return Err(
"MongoDB replaceOne() requires a filter, a replacement document, and optional options.".to_string()
);
}
let replacement = normalized_json(&args[1])?;
require_replacement_document(&replacement)?;
return Ok(MongoCommand::Replace {
collection,
filter: normalized_json(&args[0])?,
replacement,
options: args.get(2).filter(|arg| !arg.trim().is_empty()).map(|arg| normalized_json(arg)).transpose()?,
});
}
for (method, many) in [("updateOne", false), ("updateMany", true)] {
if let Some((args, tail)) = method_call(source, prefix_end, method) {
if !tail.is_empty() || !(2..=3).contains(&args.len()) {
@@ -1348,6 +1369,19 @@ fn parse_use_database(source: &str) -> Option<String> {
Some(database.to_string())
}
/// A replacement is a whole document; `{$set: ...}` here almost always means updateOne() was intended.
fn require_replacement_document(replacement: &str) -> Result<(), String> {
let Some(Value::Object(document)) = parse_json_value(replacement) else {
return Err("MongoDB replaceOne() replacement must be a document.".to_string());
};
match document.keys().find(|key| key.starts_with('$')) {
Some(operator) => Err(format!(
"MongoDB replaceOne() replacement must not contain update operators such as {operator}; use updateOne() to modify fields."
)),
None => Ok(()),
}
}
fn is_empty_object(value: &str) -> bool {
parse_json_value(value).is_some_and(|value| value.as_object().is_some_and(|object| object.is_empty()))
}
@@ -1865,6 +1899,51 @@ mod tests {
assert!(matches!(parse("db.orders.stats()").unwrap(), MongoCommand::CollectionStats { .. }));
}
#[test]
fn parses_replace_one_as_a_filtered_write() {
let command = parse(
r#"db.orders.replaceOne({_id: ObjectId("507f1f77bcf86cd799439011")}, {name: "new", tags: []}, {upsert: true})"#,
)
.unwrap();
assert_eq!(
command,
MongoCommand::Replace {
collection: "orders".to_string(),
filter: r#"{"_id":{"$oid":"507f1f77bcf86cd799439011"}}"#.to_string(),
replacement: r#"{"name":"new","tags":[]}"#.to_string(),
options: Some(r#"{"upsert":true}"#.to_string()),
}
);
assert!(command.is_mutating());
assert!(!command.is_dangerous());
assert!(!command.has_empty_filter());
assert_eq!(validate_safety(&command, true, false, false), Ok(()));
assert_eq!(validate_safety(&command, false, false, false), Err(MongoSafetyError::WritesDisabled));
let without_options = parse("db.orders.replaceOne({a: 1}, {b: 2})").unwrap();
assert!(matches!(without_options, MongoCommand::Replace { options: None, .. }));
// An empty filter replaces an arbitrary document, so it is guarded like an update.
let unbounded = parse("db.orders.replaceOne({}, {b: 2})").unwrap();
assert!(unbounded.has_empty_filter());
assert_eq!(validate_safety(&unbounded, true, false, false), Err(MongoSafetyError::EmptyFilter));
}
#[test]
fn rejects_replace_one_with_operators_or_the_wrong_shape() {
let error = parse("db.orders.replaceOne({a: 1}, {$set: {b: 2}})").unwrap_err();
assert!(error.contains("$set") && error.contains("updateOne"), "{error}");
for source in [
"db.orders.replaceOne({a: 1})",
"db.orders.replaceOne({a: 1}, {b: 2}, {upsert: true}, 4)",
"db.orders.replaceOne({a: 1}, [{b: 2}])",
"db.orders.replaceOne({a: 1}, {b: 2}).limit(1)",
] {
assert!(parse(source).is_err(), "{source}");
}
}
#[test]
fn parses_show_databases_aliases_as_read_only() {
for source in [
+20
View File
@@ -1860,6 +1860,26 @@ impl DbxBackend for WebBackend {
.map_err(|error| format!("Invalid MongoDB insert response: {error}"))?;
Ok(affected_query_result(affected_rows_from_value(&value)))
}
MongoCommand::Replace { collection, filter, replacement, options } => {
let value: Value = self
.request(
reqwest::Method::POST,
"/api/mongo/replace-document",
Some(json!({
"connectionId": connection_id,
"database": database,
"collection": collection,
"filterJson": filter,
"replacementJson": replacement,
"optionsJson": options,
})),
)
.await?
.json()
.await
.map_err(|error| format!("Invalid MongoDB replace response: {error}"))?;
Ok(affected_query_result(affected_rows_from_value(&value)))
}
MongoCommand::Update { collection, filter, update, options, many } => {
let value: Value = self
.request(
+1
View File
@@ -967,6 +967,7 @@ async fn main() {
.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/replace-document", post(routes::mongo::replace_document))
.route("/mongo/delete-document", post(routes::mongo::delete_document))
.route("/mongo/delete-documents", post(routes::mongo::delete_documents))
.route("/mongo/find-one-and-update", post(routes::mongo::find_one_and_update))
+37
View File
@@ -258,6 +258,17 @@ pub struct MongoUpdateDocumentsRequest {
pub options_json: Option<String>,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct MongoReplaceDocumentRequest {
pub connection_id: String,
pub database: String,
pub collection: String,
pub filter_json: String,
pub replacement_json: String,
pub options_json: Option<String>,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct MongoDeleteDocumentsRequest {
@@ -809,6 +820,32 @@ pub async fn update_documents(
Ok(Json(serde_json::json!({ "affected_rows": result })))
}
pub async fn replace_document(
State(state): State<Arc<WebState>>,
headers: HeaderMap,
Json(req): Json<MongoReplaceDocumentRequest>,
) -> Result<Json<serde_json::Value>, AppError> {
let database = super::mcp_policy::resolve_database(&state, &headers, &req.connection_id, &req.database).await?;
if super::mcp_policy::mongo_filter_is_effectively_unbounded(&req.filter_json) {
super::mcp_policy::ensure_dangerous_write(&state, &headers, &req.connection_id, &database, "Replace").await?;
} else {
super::mcp_policy::ensure_write(&state, &headers, &req.connection_id, &database, "Replace").await?;
}
ensure_writable(&state.app, &req.connection_id, "Replace").await?;
let result = dbx_core::mongo_ops::mongo_replace_document_core(
&state.app,
&req.connection_id,
&database,
&req.collection,
&req.filter_json,
&req.replacement_json,
req.options_json.as_deref(),
)
.await
.map_err(AppError::from)?;
Ok(Json(serde_json::json!({ "affected_rows": result })))
}
pub async fn find_one_and_update(
State(state): State<Arc<WebState>>,
headers: HeaderMap,