fix(mongodb): support createUser shell commands

Closes #5639

crates/dbx-core/src/db/agent_driver.rs

@@ -1279,6 +1279,7 @@ pub enum MongoAgentMethod {
     CountDocuments,
     ServerVersion,
     CreateIndex,
+    CreateUser,
     DropIndexes,
     DropCollection,
     DropDatabase,
@@ -1290,7 +1291,7 @@ pub enum MongoAgentMethod {
 }
 
 impl MongoAgentMethod {
-    pub const ALL: [Self; 18] = [
+    pub const ALL: [Self; 19] = [
         Self::ListDatabases,
         Self::ListCollections,
         Self::FindDocuments,
@@ -1301,6 +1302,7 @@ impl MongoAgentMethod {
         Self::CountDocuments,
         Self::ServerVersion,
         Self::CreateIndex,
+        Self::CreateUser,
         Self::DropIndexes,
         Self::DropCollection,
         Self::DropDatabase,
@@ -1323,6 +1325,7 @@ impl MongoAgentMethod {
             Self::CountDocuments => "count_documents",
             Self::ServerVersion => "server_version",
             Self::CreateIndex => "create_index",
+            Self::CreateUser => "create_user",
             Self::DropIndexes => "drop_indexes",
             Self::DropCollection => "drop_collection",
             Self::DropDatabase => "drop_database",
@@ -2520,6 +2523,13 @@ impl AgentDriverClient {
         self.call_mongo_method(MongoAgentMethod::CreateIndex, params).await
     }
 
+    pub async fn mongo_create_user<T: DeserializeOwned + Send + 'static>(
+        &mut self,
+        params: Value,
+    ) -> Result<T, String> {
+        self.call_mongo_method(MongoAgentMethod::CreateUser, params).await
+    }
+
     pub async fn mongo_drop_indexes<T: DeserializeOwned + Send + 'static>(
         &mut self,
         params: Value,
@@ -4348,6 +4358,7 @@ for line in sys.stdin:
         assert_eq!(MongoAgentMethod::CountDocuments.as_str(), "count_documents");
         assert_eq!(MongoAgentMethod::ServerVersion.as_str(), "server_version");
         assert_eq!(MongoAgentMethod::CreateIndex.as_str(), "create_index");
+        assert_eq!(MongoAgentMethod::CreateUser.as_str(), "create_user");
         assert_eq!(MongoAgentMethod::DropIndexes.as_str(), "drop_indexes");
         assert_eq!(MongoAgentMethod::DropCollection.as_str(), "drop_collection");
         assert_eq!(MongoAgentMethod::DropDatabase.as_str(), "drop_database");

crates/dbx-core/src/db/mongo_driver.rs

@@ -1231,6 +1231,48 @@ pub async fn create_index(
     Ok(name)
 }
 
+pub async fn create_user(
+    client: &Client,
+    database: &str,
+    user_json: &str,
+    write_concern_json: Option<&str>,
+) -> Result<(), String> {
+    let database = validate_mongo_namespace_name(database, "Database")?;
+    let command = create_user_command(user_json, write_concern_json)?;
+    client.database(database).run_command(command).await.map_err(|error| error.kind.to_string())?;
+    Ok(())
+}
+
+pub fn validate_create_user_request(user_json: &str, write_concern_json: Option<&str>) -> Result<(), String> {
+    create_user_command(user_json, write_concern_json).map(|_| ())
+}
+
+fn create_user_command(user_json: &str, write_concern_json: Option<&str>) -> Result<Document, String> {
+    let user_value: serde_json::Value =
+        serde_json::from_str(user_json).map_err(|error| format!("Invalid MongoDB user JSON: {error}"))?;
+    let mut user = json_object_to_document_extended_json(&user_value)
+        .map_err(|error| format!("Invalid MongoDB user document: {error}"))?;
+    let username = user
+        .remove("user")
+        .and_then(|value| value.as_str().map(str::to_string))
+        .filter(|value| !value.trim().is_empty())
+        .ok_or("MongoDB createUser requires a non-empty user name")?;
+    if user.contains_key("createUser") || user.contains_key("writeConcern") {
+        return Err("MongoDB createUser user document contains reserved command fields".to_string());
+    }
+
+    let mut command = doc! { "createUser": username };
+    command.extend(user);
+    if let Some(write_concern_json) = write_concern_json.filter(|value| !value.trim().is_empty()) {
+        let write_concern_value: serde_json::Value = serde_json::from_str(write_concern_json)
+            .map_err(|error| format!("Invalid MongoDB write concern JSON: {error}"))?;
+        let write_concern = json_object_to_document_extended_json(&write_concern_value)
+            .map_err(|error| format!("Invalid MongoDB write concern: {error}"))?;
+        command.insert("writeConcern", write_concern);
+    }
+    Ok(command)
+}
+
 /// Validate an index request before it reaches either the native driver or the
 /// Legacy Agent. In particular, `key` belongs to the createIndexes command
 /// itself and must not be smuggled through the options document.
@@ -3013,6 +3055,35 @@ mod tests {
         assert_eq!(error, "Collection name is required");
     }
 
+    #[test]
+    fn create_user_command_preserves_roles_and_write_concern() {
+        let command = create_user_command(
+            r#"{"user":"test-db","pwd":"test-password","roles":[{"role":"readWrite","db":"db1"}]}"#,
+            Some(r#"{"w":"majority","wtimeout":5000}"#),
+        )
+        .unwrap();
+
+        assert_eq!(command.keys().next().map(String::as_str), Some("createUser"));
+        assert_eq!(command.get_str("createUser").unwrap(), "test-db");
+        assert_eq!(command.get_str("pwd").unwrap(), "test-password");
+        let roles = command.get_array("roles").unwrap();
+        assert_eq!(roles[0].as_document().unwrap(), &doc! { "role": "readWrite", "db": "db1" });
+        assert_eq!(command.get_document("writeConcern").unwrap(), &doc! { "w": "majority", "wtimeout": 5000_i32 });
+    }
+
+    #[test]
+    fn create_user_command_rejects_missing_names_and_reserved_fields() {
+        for user in [
+            r#"{"pwd":"secret","roles":[]}"#,
+            r#"{"user":"","pwd":"secret","roles":[]}"#,
+            r#"{"user":"app","createUser":"other","pwd":"secret","roles":[]}"#,
+            r#"{"user":"app","writeConcern":{"w":1},"pwd":"secret","roles":[]}"#,
+        ] {
+            assert!(create_user_command(user, None).is_err(), "{user}");
+        }
+        assert!(create_user_command(r#"{"user":"app","pwd":"secret","roles":[]}"#, Some("true")).is_err());
+    }
+
     #[test]
     fn create_indexes_command_keeps_raw_options_and_generates_a_name() {
         let (command, name) = create_indexes_command(

crates/dbx-core/src/mongo_ops.rs

@@ -466,6 +466,37 @@ pub async fn mongo_create_index_core(
     }
 }
 
+pub async fn mongo_create_user_core(
+    state: &AppState,
+    connection_id: &str,
+    database: &str,
+    user_json: &str,
+    write_concern_json: Option<&str>,
+) -> Result<u64, String> {
+    mongo_driver::validate_mongo_namespace_name(database, "Database")?;
+    mongo_driver::validate_create_user_request(user_json, write_concern_json)?;
+    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::create_user(client, database, user_json, write_concern_json).await?;
+            Ok(1)
+        }
+        PoolKind::Agent(client) => {
+            let mut client = client.lock().await;
+            let result: serde_json::Value = client
+                .mongo_create_user(serde_json::json!({
+                    "database": database,
+                    "user_json": user_json,
+                    "write_concern_json": write_concern_json,
+                }))
+                .await?;
+            Ok(result.get("affected_rows").and_then(serde_json::Value::as_u64).unwrap_or(1))
+        }
+        _ => Err("Not a MongoDB connection".to_string()),
+    }
+}
+
 pub async fn mongo_drop_indexes_core(
     state: &AppState,
     connection_id: &str,
@@ -875,6 +906,12 @@ pub async fn execute_mongo_command_core(
                 mongo_create_index_core(state, connection_id, database, collection, keys, options.as_deref()).await?;
             Ok(scalar_query_result("name", Value::String(name)))
         }
+        MongoCommand::CreateUser { user_json, write_concern_json } => {
+            let affected =
+                mongo_create_user_core(state, connection_id, database, user_json, write_concern_json.as_deref())
+                    .await?;
+            Ok(affected_query_result(affected))
+        }
         MongoCommand::DropIndexes { collection, indexes, single } => {
             let result =
                 mongo_drop_indexes_core(state, connection_id, database, collection, indexes.as_deref(), *single)

crates/dbx-core/src/mongo_shell.rs

@@ -8,6 +8,13 @@ pub enum MongoCommand {
     Version,
     #[serde(rename = "use")]
     Use { database: String },
+    #[serde(rename = "createUser")]
+    CreateUser {
+        #[serde(rename = "userJson")]
+        user_json: String,
+        #[serde(rename = "writeConcernJson")]
+        write_concern_json: Option<String>,
+    },
     #[serde(rename = "find")]
     Find {
         collection: String,
@@ -77,7 +84,8 @@ impl MongoCommand {
     pub fn is_mutating(&self) -> bool {
         matches!(
             self,
-            Self::Insert { .. }
+            Self::CreateUser { .. }
+                | Self::Insert { .. }
                 | Self::Update { .. }
                 | Self::Delete { .. }
                 | Self::CreateIndex { .. }
@@ -90,7 +98,7 @@ impl MongoCommand {
     }
 
     pub fn is_dangerous(&self) -> bool {
-        matches!(self, Self::DropCollection { .. })
+        matches!(self, Self::CreateUser { .. } | Self::DropCollection { .. })
             || matches!(self, Self::DropIndexes { indexes: None, single: false, .. })
             || matches!(self, Self::Aggregate { pipeline, .. } if aggregate_writes(pipeline))
     }
@@ -348,6 +356,23 @@ pub fn parse(input: &str) -> Result<MongoCommand, String> {
     if let Some(database) = parse_use_database(source) {
         return Ok(MongoCommand::Use { database });
     }
+    if let Some((args, tail)) = database_method_call(source, "createUser") {
+        if !tail.is_empty() || !(1..=2).contains(&args.len()) {
+            return Err("MongoDB createUser() requires a user document and optional write concern.".to_string());
+        }
+        let user_json = normalized_json(&args[0])?;
+        let user = parse_json_value(&user_json)
+            .and_then(|value| value.as_object().cloned())
+            .ok_or("MongoDB createUser() requires a user document.")?;
+        if user.get("user").and_then(Value::as_str).is_none_or(|user| user.trim().is_empty()) {
+            return Err("MongoDB createUser() requires a non-empty user name.".to_string());
+        }
+        let write_concern_json = optional_json_argument(args.get(1))?;
+        if write_concern_json.as_deref().and_then(parse_json_value).is_some_and(|value| !value.is_object()) {
+            return Err("MongoDB createUser() write concern must be a document.".to_string());
+        }
+        return Ok(MongoCommand::CreateUser { user_json, write_concern_json });
+    }
     let (collection, prefix_end) = parse_collection_prefix(source)?;
 
     if let Some((args, tail)) = method_call(source, prefix_end, "find") {
@@ -651,6 +676,24 @@ fn method_call(source: &str, prefix_end: usize, method: &str) -> Option<(Vec<Str
     Some((split_top_level(&source[open + 1..close]), source[close + 1..].trim().to_string()))
 }
 
+fn database_method_call(source: &str, method: &str) -> Option<(Vec<String>, String)> {
+    if !source.get(..2).is_some_and(|prefix| prefix.eq_ignore_ascii_case("db")) {
+        return None;
+    }
+    let after_db = source.get(2..)?.trim_start();
+    let after_dot = after_db.strip_prefix('.')?.trim_start();
+    if !after_dot.get(..method.len()).is_some_and(|name| name.eq_ignore_ascii_case(method)) {
+        return None;
+    }
+    let after_method = after_dot.get(method.len()..)?.trim_start();
+    if !after_method.starts_with('(') {
+        return None;
+    }
+    let open = source.len() - after_method.len();
+    let close = matching_paren(source, open)?;
+    Some((split_top_level(&source[open + 1..close]), source[close + 1..].trim().to_string()))
+}
+
 fn chained_calls(chain: &str) -> Result<Vec<(String, Vec<String>)>, String> {
     let mut rest = chain.trim();
     let mut calls = Vec::new();
@@ -957,6 +1000,28 @@ mod tests {
         assert_eq!(validate_safety(&legacy_update, true, false, false), Err(MongoSafetyError::EmptyFilter));
     }
 
+    #[test]
+    fn parses_create_user_as_a_dangerous_write() {
+        let command = parse(
+            r#"db . createUser({user: "test-db", pwd: "test-password", roles: [{role: "readWrite", db: "db1"}]}, {w: "majority"})"#,
+        )
+        .unwrap();
+        assert_eq!(
+            command,
+            MongoCommand::CreateUser {
+                user_json: r#"{"user":"test-db","pwd":"test-password","roles":[{"role":"readWrite","db":"db1"}]}"#
+                    .to_string(),
+                write_concern_json: Some(r#"{"w":"majority"}"#.to_string()),
+            }
+        );
+        assert!(command.is_mutating());
+        assert!(command.is_dangerous());
+        assert_eq!(validate_safety(&command, true, false, false), Err(MongoSafetyError::Dangerous));
+        assert_eq!(validate_safety(&command, true, true, false), Ok(()));
+        assert!(parse(r#"db.createUser({pwd: "missing-user", roles: []})"#).is_err());
+        assert!(parse(r#"db.createUser({user: "test"}, "majority")"#).is_err());
+    }
+
     #[test]
     fn treats_effectively_unbounded_write_filters_as_dangerous() {
         for command in [
@@ -1102,6 +1167,10 @@ mod tests {
         let count = serde_json::to_value(parse("db.items.count({})").unwrap()).unwrap();
         assert_eq!(count["kind"], "countDocuments");
         assert_eq!(count["accurate"], false);
+        let create_user =
+            serde_json::to_value(parse(r#"db.createUser({user: "app", pwd: "secret", roles: []})"#).unwrap()).unwrap();
+        assert_eq!(create_user["kind"], "createUser");
+        assert_eq!(create_user["userJson"], r#"{"user":"app","pwd":"secret","roles":[]}"#);
     }
 
     #[test]

crates/dbx-core/src/query.rs

@@ -34,7 +34,7 @@ pub const QUERY_CANCELED: &str = "Query canceled";
 /// Fallback when a Mongo connection hits the generic SQL executor instead of the shell path.
 /// Wording must match packages/mongo-shell `MONGO_SHELL_COMMAND_HINT`
 /// (desktop/CLI diagnose first; this is only the Rust SQL-executor backstop).
-const MONGO_SHELL_COMMAND_HINT: &str = "Use MongoDB shell-style commands, for example: db.collection.find({}).limit(100), db.collection.aggregate([]), db.collection.aggregate([], { explain: true }), db.version(), db.collection.countDocuments({}), db.collection.distinct(\"field\"), db.collection.getIndexes(), db.collection.createIndex({...}), or db.collection.insertOne({...}).";
+const MONGO_SHELL_COMMAND_HINT: &str = "Use MongoDB shell-style commands, for example: db.collection.find({}).limit(100), db.collection.aggregate([]), db.collection.aggregate([], { explain: true }), db.version(), db.collection.countDocuments({}), db.collection.distinct(\"field\"), db.collection.getIndexes(), db.collection.createIndex({...}), db.createUser({...}), or db.collection.insertOne({...}).";
 const SQL_OMITTED_ERROR_CONTEXT: &str =
     "SQL text omitted from user-facing error; enable debug SQL diagnostics to inspect the original statement.";