feat(mongodb): improve sidebar index management
This commit is contained in:
1 parent
22055e081f
commit
8e7b74d0ee
56 files changed
+3558
-246
No files matched your search
@@ -971,6 +971,7 @@ pub enum AgentCapability {
|
||||
EtcdWatch,
|
||||
EtcdLease,
|
||||
EtcdAuth,
|
||||
MongoDropDatabase,
|
||||
MultiSession,
|
||||
StructuredErrorV1,
|
||||
}
|
||||
@@ -1053,7 +1054,7 @@ fn parse_agent_rpc_error_header(header: &str) -> (Option<i64>, String) {
|
||||
}
|
||||
|
||||
impl AgentCapability {
|
||||
pub const ALL: [Self; 20] = [
|
||||
pub const ALL: [Self; 21] = [
|
||||
Self::Connect,
|
||||
Self::TestConnection,
|
||||
Self::Metadata,
|
||||
@@ -1072,6 +1073,7 @@ impl AgentCapability {
|
||||
Self::EtcdWatch,
|
||||
Self::EtcdLease,
|
||||
Self::EtcdAuth,
|
||||
Self::MongoDropDatabase,
|
||||
Self::MultiSession,
|
||||
Self::StructuredErrorV1,
|
||||
];
|
||||
@@ -1096,6 +1098,7 @@ impl AgentCapability {
|
||||
Self::EtcdWatch => "etcd_watch",
|
||||
Self::EtcdLease => "etcd_lease",
|
||||
Self::EtcdAuth => "etcd_auth",
|
||||
Self::MongoDropDatabase => "mongo_drop_database",
|
||||
Self::MultiSession => "multi_session",
|
||||
Self::StructuredErrorV1 => "structured_error_v1",
|
||||
}
|
||||
@@ -1275,6 +1278,7 @@ pub enum MongoAgentMethod {
|
||||
CreateIndex,
|
||||
DropIndexes,
|
||||
DropCollection,
|
||||
DropDatabase,
|
||||
InsertDocument,
|
||||
UpdateDocument,
|
||||
UpdateDocuments,
|
||||
@@ -1283,7 +1287,7 @@ pub enum MongoAgentMethod {
|
||||
}
|
||||
|
||||
impl MongoAgentMethod {
|
||||
pub const ALL: [Self; 14] = [
|
||||
pub const ALL: [Self; 15] = [
|
||||
Self::ListDatabases,
|
||||
Self::ListCollections,
|
||||
Self::FindDocuments,
|
||||
@@ -1293,6 +1297,7 @@ impl MongoAgentMethod {
|
||||
Self::CreateIndex,
|
||||
Self::DropIndexes,
|
||||
Self::DropCollection,
|
||||
Self::DropDatabase,
|
||||
Self::InsertDocument,
|
||||
Self::UpdateDocument,
|
||||
Self::UpdateDocuments,
|
||||
@@ -1311,6 +1316,7 @@ impl MongoAgentMethod {
|
||||
Self::CreateIndex => "create_index",
|
||||
Self::DropIndexes => "drop_indexes",
|
||||
Self::DropCollection => "drop_collection",
|
||||
Self::DropDatabase => "drop_database",
|
||||
Self::InsertDocument => "insert_document",
|
||||
Self::UpdateDocument => "update_document",
|
||||
Self::UpdateDocuments => "update_documents",
|
||||
@@ -2438,6 +2444,19 @@ impl AgentDriverClient {
|
||||
self.call_mongo_method(MongoAgentMethod::ListCollections, mongo_database_params(database)).await
|
||||
}
|
||||
|
||||
/// Request collection metadata while accepting old Agents that still
|
||||
/// return the historical string array and ignore `include_types`.
|
||||
pub async fn mongo_list_collection_specs<T: DeserializeOwned + Send + 'static>(
|
||||
&mut self,
|
||||
database: &str,
|
||||
) -> Result<T, String> {
|
||||
self.call_mongo_method(
|
||||
MongoAgentMethod::ListCollections,
|
||||
serde_json::json!({ "database": database, "include_types": true }),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn mongo_find_documents<T: DeserializeOwned + Send + 'static>(
|
||||
&mut self,
|
||||
params: Value,
|
||||
@@ -2488,6 +2507,13 @@ impl AgentDriverClient {
|
||||
self.call_mongo_method(MongoAgentMethod::DropCollection, params).await
|
||||
}
|
||||
|
||||
pub async fn mongo_drop_database<T: DeserializeOwned + Send + 'static>(
|
||||
&mut self,
|
||||
params: Value,
|
||||
) -> Result<T, String> {
|
||||
self.call_mongo_method(MongoAgentMethod::DropDatabase, params).await
|
||||
}
|
||||
|
||||
pub async fn mongo_insert_document<T: DeserializeOwned + Send + 'static>(
|
||||
&mut self,
|
||||
params: Value,
|
||||
@@ -2660,6 +2686,7 @@ pub fn agent_supports_capability(handshake: Option<&AgentHandshake>, capability:
|
||||
| AgentCapability::EtcdWatch
|
||||
| AgentCapability::EtcdLease
|
||||
| AgentCapability::EtcdAuth
|
||||
| AgentCapability::MongoDropDatabase
|
||||
) {
|
||||
return handshake.map(|value| value.supports(capability)).unwrap_or(false);
|
||||
}
|
||||
@@ -4242,9 +4269,10 @@ for line in sys.stdin:
|
||||
assert_eq!(AgentCapability::EtcdWatch.as_str(), "etcd_watch");
|
||||
assert_eq!(AgentCapability::EtcdLease.as_str(), "etcd_lease");
|
||||
assert_eq!(AgentCapability::EtcdAuth.as_str(), "etcd_auth");
|
||||
assert_eq!(AgentCapability::MongoDropDatabase.as_str(), "mongo_drop_database");
|
||||
assert_eq!(AgentCapability::MultiSession.as_str(), "multi_session");
|
||||
assert_eq!(AgentCapability::ALL.len(), 20);
|
||||
assert_eq!(AgentCapability::StructuredErrorV1.as_str(), "structured_error_v1");
|
||||
assert_eq!(AgentCapability::ALL.len(), 21);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -4292,6 +4320,7 @@ for line in sys.stdin:
|
||||
assert_eq!(MongoAgentMethod::CreateIndex.as_str(), "create_index");
|
||||
assert_eq!(MongoAgentMethod::DropIndexes.as_str(), "drop_indexes");
|
||||
assert_eq!(MongoAgentMethod::DropCollection.as_str(), "drop_collection");
|
||||
assert_eq!(MongoAgentMethod::DropDatabase.as_str(), "drop_database");
|
||||
assert_eq!(MongoAgentMethod::InsertDocument.as_str(), "insert_document");
|
||||
assert_eq!(MongoAgentMethod::UpdateDocument.as_str(), "update_document");
|
||||
assert_eq!(MongoAgentMethod::UpdateDocuments.as_str(), "update_documents");
|
||||
@@ -4350,6 +4379,7 @@ for line in sys.stdin:
|
||||
fn exposes_mongo_protocol_wrappers() {
|
||||
let _mongo_list_databases = AgentDriverClient::mongo_list_databases::<serde_json::Value>;
|
||||
let _mongo_list_collections = AgentDriverClient::mongo_list_collections::<serde_json::Value>;
|
||||
let _mongo_list_collection_specs = AgentDriverClient::mongo_list_collection_specs::<serde_json::Value>;
|
||||
let _mongo_find_documents = AgentDriverClient::mongo_find_documents::<serde_json::Value>;
|
||||
let _mongo_find_documents_extended_json =
|
||||
AgentDriverClient::mongo_find_documents_extended_json::<serde_json::Value>;
|
||||
@@ -4357,6 +4387,7 @@ for line in sys.stdin:
|
||||
let _mongo_create_index = AgentDriverClient::mongo_create_index::<serde_json::Value>;
|
||||
let _mongo_drop_indexes = AgentDriverClient::mongo_drop_indexes::<serde_json::Value>;
|
||||
let _mongo_drop_collection = AgentDriverClient::mongo_drop_collection::<serde_json::Value>;
|
||||
let _mongo_drop_database = AgentDriverClient::mongo_drop_database::<serde_json::Value>;
|
||||
let _mongo_insert_document = AgentDriverClient::mongo_insert_document::<serde_json::Value>;
|
||||
let _mongo_update_document = AgentDriverClient::mongo_update_document::<serde_json::Value>;
|
||||
let _mongo_update_documents = AgentDriverClient::mongo_update_documents::<serde_json::Value>;
|
||||
@@ -4587,6 +4618,12 @@ for line in sys.stdin:
|
||||
assert!(!agent_supports_capability(Some(&handshake), AgentCapability::KvStatus));
|
||||
assert!(!agent_supports_capability(None, AgentCapability::KvHistory));
|
||||
assert!(!agent_supports_capability(Some(&handshake), AgentCapability::KvHistory));
|
||||
assert!(!agent_supports_capability(None, AgentCapability::MongoDropDatabase));
|
||||
assert!(!agent_supports_capability(Some(&handshake), AgentCapability::MongoDropDatabase));
|
||||
|
||||
let mongo_handshake =
|
||||
AgentHandshake { capabilities: vec![AgentCapability::MongoDropDatabase.as_str().to_string()], ..handshake };
|
||||
assert!(agent_supports_capability(Some(&mongo_handshake), AgentCapability::MongoDropDatabase));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -20,6 +20,14 @@ pub type MongoDocumentResult = DocumentQueryResult;
|
||||
pub struct MongoDropIndexesResult {
|
||||
pub dropped_names: Vec<String>,
|
||||
pub affected_rows: u64,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub failures: Vec<MongoDropIndexFailure>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct MongoDropIndexFailure {
|
||||
pub name: String,
|
||||
pub message: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||||
@@ -222,6 +230,16 @@ impl MongoCollectionKind {
|
||||
_ => Self::Collection,
|
||||
}
|
||||
}
|
||||
|
||||
/// Convert the Legacy Agent's optional listCollections metadata. Unknown
|
||||
/// values stay compatible with the former name-only response.
|
||||
pub fn from_metadata_kind(kind: Option<&str>) -> Self {
|
||||
match kind.map(str::trim) {
|
||||
Some(kind) if kind.eq_ignore_ascii_case("view") => Self::View,
|
||||
Some(kind) if kind.eq_ignore_ascii_case("timeseries") => Self::Timeseries,
|
||||
_ => Self::Collection,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Name + kind for a MongoDB collection/view/timeseries returned by full `listCollections`.
|
||||
@@ -528,25 +546,26 @@ pub async fn create_database(client: &Client, database: &str) -> Result<(), Stri
|
||||
}
|
||||
|
||||
pub async fn drop_database(client: &Client, database: &str) -> Result<(), String> {
|
||||
let database = database.trim();
|
||||
if database.is_empty() {
|
||||
return Err("Database name is required".to_string());
|
||||
}
|
||||
let database = validate_mongo_namespace_name(database, "Database")?;
|
||||
client.database(database).drop().await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn drop_collection(client: &Client, database: &str, collection: &str) -> Result<(), String> {
|
||||
let database = database.trim();
|
||||
let collection = collection.trim();
|
||||
if database.is_empty() {
|
||||
return Err("Database name is required".to_string());
|
||||
}
|
||||
if collection.is_empty() {
|
||||
return Err("Collection name is required".to_string());
|
||||
}
|
||||
let database = validate_mongo_namespace_name(database, "Database")?;
|
||||
let collection = validate_mongo_namespace_name(collection, "Collection")?;
|
||||
client.database(database).collection::<Document>(collection).drop().await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
/// MongoDB namespace identifiers must reach the server unchanged. In
|
||||
/// particular, a collection name may legitimately contain leading or trailing
|
||||
/// whitespace, so validation only rejects the empty identifier.
|
||||
pub fn validate_mongo_namespace_name<'a>(name: &'a str, kind: &str) -> Result<&'a str, String> {
|
||||
if name.is_empty() {
|
||||
return Err(format!("{kind} name is required"));
|
||||
}
|
||||
Ok(name)
|
||||
}
|
||||
|
||||
/// Build the admin `renameCollection` command document for a same-database rename.
|
||||
///
|
||||
/// Collection (and database) names are identifiers and must be passed exactly as provided —
|
||||
@@ -1020,9 +1039,44 @@ pub async fn create_index(
|
||||
keys_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<String, String> {
|
||||
let database = validate_mongo_namespace_name(database, "Database")?;
|
||||
let collection = validate_mongo_namespace_name(collection, "Collection")?;
|
||||
let (command, name) = create_indexes_command(collection, keys_json, options_json)?;
|
||||
client.database(database).run_command(command).await.map_err(|e| e.kind.to_string())?;
|
||||
Ok(name)
|
||||
}
|
||||
|
||||
/// 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.
|
||||
pub fn validate_create_index_request(keys_json: &str, options_json: Option<&str>) -> Result<(), String> {
|
||||
parse_create_index_spec(keys_json, options_json).map(|_| ())
|
||||
}
|
||||
|
||||
fn create_indexes_command(
|
||||
collection: &str,
|
||||
keys_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<(Document, String), String> {
|
||||
let (keys, options, name) = parse_create_index_spec(keys_json, options_json)?;
|
||||
let mut index = doc! { "key": keys };
|
||||
for (option, value) in options {
|
||||
index.insert(option, value);
|
||||
}
|
||||
|
||||
Ok((doc! { "createIndexes": collection, "indexes": [index] }, name))
|
||||
}
|
||||
|
||||
fn parse_create_index_spec(
|
||||
keys_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<(Document, Document, String), String> {
|
||||
let keys_value: serde_json::Value =
|
||||
serde_json::from_str(keys_json).map_err(|e| format!("Invalid index keys JSON: {e}"))?;
|
||||
let keys = json_object_to_document(&keys_value).map_err(|e| format!("Invalid index keys: {e}"))?;
|
||||
// The internal transport uses JSON rather than a Mongo shell expression.
|
||||
// Keep ordinary strings literal while supporting official Extended JSON
|
||||
// wrappers, matching the Legacy Agent's Document.parse behavior.
|
||||
let keys = json_object_to_document_extended_json(&keys_value).map_err(|e| format!("Invalid index keys: {e}"))?;
|
||||
if keys.is_empty() {
|
||||
return Err("Index keys are required".to_string());
|
||||
}
|
||||
@@ -1031,16 +1085,40 @@ pub async fn create_index(
|
||||
Some(json) => {
|
||||
let value: serde_json::Value =
|
||||
serde_json::from_str(json).map_err(|e| format!("Invalid index options JSON: {e}"))?;
|
||||
let doc = json_object_to_document(&value).map_err(|e| format!("Invalid index options: {e}"))?;
|
||||
Some(mongodb::bson::from_document::<IndexOptions>(doc).map_err(|e| format!("Invalid index options: {e}"))?)
|
||||
json_object_to_document_extended_json(&value).map_err(|e| format!("Invalid index options: {e}"))?
|
||||
}
|
||||
None => None,
|
||||
None => Document::new(),
|
||||
};
|
||||
if options.contains_key("key") {
|
||||
return Err("Index options cannot contain \"key\"; specify index fields in keys JSON".to_string());
|
||||
}
|
||||
let name = match options.get("name") {
|
||||
Some(Bson::String(name)) if !name.trim().is_empty() => name.clone(),
|
||||
Some(_) => return Err("Index option \"name\" must be a non-empty string".to_string()),
|
||||
None => default_index_name(&keys),
|
||||
};
|
||||
|
||||
let col = client.database(database).collection::<Document>(collection);
|
||||
let result =
|
||||
col.create_index(IndexModel::builder().keys(keys).options(options).build()).await.map_err(|e| e.to_string())?;
|
||||
Ok(result.index_name)
|
||||
// Use the raw server command instead of deserializing into the driver's
|
||||
// fixed IndexOptions struct. This keeps native and Legacy Agent requests
|
||||
// equivalent and lets the connected MongoDB version validate new options.
|
||||
let mut options = options;
|
||||
if !options.contains_key("name") {
|
||||
options.insert("name", name.clone());
|
||||
}
|
||||
Ok((keys, options, name))
|
||||
}
|
||||
|
||||
fn default_index_name(keys: &Document) -> String {
|
||||
keys.iter()
|
||||
.map(|(field, value)| {
|
||||
let value = match value {
|
||||
Bson::String(value) => value.clone(),
|
||||
value => value.to_string(),
|
||||
};
|
||||
format!("{field}_{value}")
|
||||
})
|
||||
.collect::<Vec<_>>()
|
||||
.join("_")
|
||||
}
|
||||
|
||||
pub async fn drop_indexes(
|
||||
@@ -1050,14 +1128,8 @@ pub async fn drop_indexes(
|
||||
indexes_json: Option<&str>,
|
||||
single: bool,
|
||||
) -> Result<MongoDropIndexesResult, String> {
|
||||
let database = database.trim();
|
||||
let collection = collection.trim();
|
||||
if database.is_empty() {
|
||||
return Err("Database name is required".to_string());
|
||||
}
|
||||
if collection.is_empty() {
|
||||
return Err("Collection name is required".to_string());
|
||||
}
|
||||
let database = validate_mongo_namespace_name(database, "Database")?;
|
||||
let collection = validate_mongo_namespace_name(collection, "Collection")?;
|
||||
|
||||
let index = parse_drop_indexes_value(indexes_json, single)?;
|
||||
let before = list_indexes(client, database, collection).await?;
|
||||
@@ -1068,7 +1140,57 @@ pub async fn drop_indexes(
|
||||
.map_err(|e| e.to_string())?;
|
||||
let after = list_indexes(client, database, collection).await?;
|
||||
let dropped_names = diff_dropped_index_names(&before, &after);
|
||||
Ok(MongoDropIndexesResult { affected_rows: dropped_names.len() as u64, dropped_names })
|
||||
Ok(MongoDropIndexesResult { affected_rows: dropped_names.len() as u64, dropped_names, failures: Vec::new() })
|
||||
}
|
||||
|
||||
/// Validate a drop-index request before it is sent to either MongoDB driver.
|
||||
/// The Legacy Agent receives the original JSON, so keeping this validation
|
||||
/// public gives both paths the same protection for MongoDB's default index.
|
||||
pub fn validate_drop_indexes_request(indexes_json: Option<&str>, single: bool) -> Result<(), String> {
|
||||
parse_drop_indexes_value(indexes_json, single).map(|_| ())
|
||||
}
|
||||
|
||||
/// MongoDB added array-form `dropIndexes.index` in 4.2. Unknown version
|
||||
/// strings preserve the modern single-command semantics instead of risking a
|
||||
/// partially applied serial fallback.
|
||||
pub fn mongo_server_requires_serial_drop_indexes(version: &str) -> bool {
|
||||
let Some(start) = version.find(|character: char| character.is_ascii_digit()) else {
|
||||
return false;
|
||||
};
|
||||
let mut components = version[start..].split('.');
|
||||
let Some(major) = components.next().and_then(parse_version_component) else {
|
||||
return false;
|
||||
};
|
||||
let Some(minor) = components.next().and_then(parse_version_component) else {
|
||||
return false;
|
||||
};
|
||||
(major, minor) < (4, 2)
|
||||
}
|
||||
|
||||
fn parse_version_component(component: &str) -> Option<u32> {
|
||||
let digits = component.chars().take_while(|character| character.is_ascii_digit()).collect::<String>();
|
||||
(!digits.is_empty()).then(|| digits.parse().ok()).flatten()
|
||||
}
|
||||
|
||||
/// Return names that must be issued as individual dropIndex commands for
|
||||
/// MongoDB 3.4 compatibility. The original request is fully parsed first, so
|
||||
/// an invalid name (including `_id_`) cannot cause a partial batch mutation.
|
||||
pub fn serial_drop_index_names(indexes_json: Option<&str>, single: bool) -> Result<Option<Vec<String>>, String> {
|
||||
let Bson::Array(indexes) = parse_drop_indexes_value(indexes_json, single)? else {
|
||||
return Ok(None);
|
||||
};
|
||||
|
||||
let mut seen = HashSet::new();
|
||||
let mut names = Vec::new();
|
||||
for index in indexes {
|
||||
let Bson::String(name) = index else {
|
||||
return Err("dropIndexes only accepts arrays of string index names".to_string());
|
||||
};
|
||||
if seen.insert(name.clone()) {
|
||||
names.push(name);
|
||||
}
|
||||
}
|
||||
Ok(Some(names))
|
||||
}
|
||||
|
||||
fn diff_dropped_index_names(before: &[IndexInfo], after: &[IndexInfo]) -> Vec<String> {
|
||||
@@ -1086,51 +1208,64 @@ fn parse_drop_indexes_value(indexes_json: Option<&str>, single: bool) -> Result<
|
||||
|
||||
fn parse_drop_indexes_json(json: &str, single: bool) -> Result<Bson, String> {
|
||||
let value: serde_json::Value = serde_json::from_str(json).map_err(|e| format!("Invalid index JSON: {e}"))?;
|
||||
let bson = json_value_to_bson(&value);
|
||||
if single {
|
||||
validate_single_drop_index_value(&value)?;
|
||||
validate_single_drop_index_value(&bson)?;
|
||||
} else {
|
||||
validate_multi_drop_indexes_value(&value)?;
|
||||
validate_multi_drop_indexes_value(&bson)?;
|
||||
}
|
||||
Ok(json_value_to_bson(&value))
|
||||
Ok(bson)
|
||||
}
|
||||
|
||||
fn validate_single_drop_index_value(value: &serde_json::Value) -> Result<(), String> {
|
||||
fn validate_single_drop_index_value(value: &Bson) -> Result<(), String> {
|
||||
match value {
|
||||
serde_json::Value::String(name) => {
|
||||
Bson::String(name) => {
|
||||
if name.trim().is_empty() {
|
||||
Err("Index name is required".to_string())
|
||||
} else if name == "*" {
|
||||
Err(r#"dropIndex does not accept "*"; use dropIndexes() or dropIndexes("*") instead"#.to_string())
|
||||
} else if name == "_id_" {
|
||||
Err("The default MongoDB _id_ index cannot be dropped".to_string())
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
serde_json::Value::Object(doc) if doc.is_empty() => Err("Index specification is required".to_string()),
|
||||
serde_json::Value::Object(_) => Ok(()),
|
||||
serde_json::Value::Array(_) => {
|
||||
Bson::Document(doc) if doc.is_empty() => Err("Index specification is required".to_string()),
|
||||
Bson::Document(doc) if is_default_id_index_specification(doc) => {
|
||||
Err("The default MongoDB _id_ index cannot be dropped".to_string())
|
||||
}
|
||||
Bson::Document(_) => Ok(()),
|
||||
Bson::Array(_) => {
|
||||
Err("dropIndex only accepts a string index name or JSON document; arrays are not supported".to_string())
|
||||
}
|
||||
_ => Err("dropIndex only accepts a string index name or JSON document".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_multi_drop_indexes_value(value: &serde_json::Value) -> Result<(), String> {
|
||||
fn validate_multi_drop_indexes_value(value: &Bson) -> Result<(), String> {
|
||||
match value {
|
||||
serde_json::Value::String(name) => {
|
||||
Bson::String(name) => {
|
||||
if name.trim().is_empty() {
|
||||
Err("Index name is required".to_string())
|
||||
} else if name == "_id_" {
|
||||
Err("The default MongoDB _id_ index cannot be dropped".to_string())
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
serde_json::Value::Object(doc) if doc.is_empty() => Err("Index specification is required".to_string()),
|
||||
serde_json::Value::Object(_) => Ok(()),
|
||||
serde_json::Value::Array(items) if items.is_empty() => {
|
||||
Err("dropIndexes only accepts non-empty string arrays".to_string())
|
||||
Bson::Document(doc) if doc.is_empty() => Err("Index specification is required".to_string()),
|
||||
Bson::Document(doc) if is_default_id_index_specification(doc) => {
|
||||
Err("The default MongoDB _id_ index cannot be dropped".to_string())
|
||||
}
|
||||
serde_json::Value::Array(items) => {
|
||||
if items.iter().all(|item| matches!(item, serde_json::Value::String(name) if !name.trim().is_empty())) {
|
||||
Ok(())
|
||||
Bson::Document(_) => Ok(()),
|
||||
Bson::Array(items) if items.is_empty() => Err("dropIndexes only accepts non-empty string arrays".to_string()),
|
||||
Bson::Array(items) => {
|
||||
if items.iter().all(|item| matches!(item, Bson::String(name) if !name.trim().is_empty())) {
|
||||
if items.iter().any(|item| matches!(item, Bson::String(name) if name == "_id_")) {
|
||||
Err("The default MongoDB _id_ index cannot be dropped".to_string())
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
} else {
|
||||
Err("dropIndexes only accepts arrays of string index names".to_string())
|
||||
}
|
||||
@@ -1139,6 +1274,20 @@ fn validate_multi_drop_indexes_value(value: &serde_json::Value) -> Result<(), St
|
||||
}
|
||||
}
|
||||
|
||||
fn is_default_id_index_specification(specification: &Document) -> bool {
|
||||
specification.len() == 1 && specification.get("_id").is_some_and(is_bson_numeric_one)
|
||||
}
|
||||
|
||||
fn is_bson_numeric_one(value: &Bson) -> bool {
|
||||
match value {
|
||||
Bson::Int32(value) => *value == 1,
|
||||
Bson::Int64(value) => *value == 1,
|
||||
Bson::Double(value) => *value == 1.0,
|
||||
Bson::Decimal128(value) => value.to_string().parse::<f64>().is_ok_and(|value| value == 1.0),
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn insert_document(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
@@ -2592,6 +2741,102 @@ mod tests {
|
||||
assert!(index.is_primary);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn mongo_namespace_validation_preserves_whitespace_and_rejects_only_empty_names() {
|
||||
assert_eq!(validate_mongo_namespace_name(" app ", "Database").unwrap(), " app ");
|
||||
assert_eq!(validate_mongo_namespace_name(" users ", "Collection").unwrap(), " users ");
|
||||
assert_eq!(validate_mongo_namespace_name(" ", "Collection").unwrap(), " ");
|
||||
|
||||
let error = validate_mongo_namespace_name("", "Collection").unwrap_err();
|
||||
assert_eq!(error, "Collection name is required");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_indexes_command_keeps_raw_options_and_generates_a_name() {
|
||||
let (command, name) = create_indexes_command(
|
||||
"users",
|
||||
r#"{"email":1,"createdAt":-1,"content":"text"}"#,
|
||||
Some(
|
||||
r#"{"unique":true,"partialFilterExpression":{"verified":true},"customServerOption":{"enabled":true}}"#,
|
||||
),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(name, "email_1_createdAt_-1_content_text");
|
||||
assert_eq!(command.get_str("createIndexes").unwrap(), "users");
|
||||
let index = command.get_array("indexes").unwrap()[0].as_document().unwrap();
|
||||
assert_eq!(
|
||||
index.get_document("key").unwrap().keys().collect::<Vec<_>>(),
|
||||
vec!["email", "createdAt", "content"],
|
||||
"compound-index key order is part of MongoDB index semantics"
|
||||
);
|
||||
assert_eq!(
|
||||
index.get_document("key").unwrap(),
|
||||
&doc! { "email": 1_i32, "createdAt": -1_i32, "content": "text" }
|
||||
);
|
||||
assert!(index.get_bool("unique").unwrap());
|
||||
assert_eq!(index.get_str("name").unwrap(), "email_1_createdAt_-1_content_text");
|
||||
assert_eq!(index.get_document("customServerOption").unwrap(), &doc! { "enabled": true });
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_indexes_command_canonicalizes_a_whole_double_in_the_default_name() {
|
||||
let (command, name) = create_indexes_command("users", r#"{"email":1.0}"#, None).unwrap();
|
||||
|
||||
assert_eq!(name, "email_1");
|
||||
let index = command.get_array("indexes").unwrap()[0].as_document().unwrap();
|
||||
assert_eq!(index.get_str("name").unwrap(), "email_1");
|
||||
assert!(
|
||||
matches!(index.get("key"), Some(Bson::Document(keys)) if matches!(keys.get("email"), Some(Bson::Double(value)) if *value == 1.0))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_indexes_command_preserves_extended_json_options() {
|
||||
let (command, name) = create_indexes_command(
|
||||
"events",
|
||||
r#"{"expiresAt":1}"#,
|
||||
Some(r#"{"name":"expires_ttl","partialFilterExpression":{"createdAt":{"$date":"2026-06-10T13:59:31.287Z"}}}"#),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(name, "expires_ttl");
|
||||
let index = command.get_array("indexes").unwrap()[0].as_document().unwrap();
|
||||
let filter = index.get_document("partialFilterExpression").unwrap();
|
||||
assert!(matches!(filter.get("createdAt"), Some(Bson::DateTime(_))));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_indexes_command_keeps_plain_json_strings_literal() {
|
||||
let (command, _) = create_indexes_command(
|
||||
"events",
|
||||
r#"{"expiresAt":1}"#,
|
||||
Some(r#"{"partialFilterExpression":{"source":"ISODate(\"2026-06-10T13:59:31.287Z\")"}}"#),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let index = command.get_array("indexes").unwrap()[0].as_document().unwrap();
|
||||
let filter = index.get_document("partialFilterExpression").unwrap();
|
||||
assert!(
|
||||
matches!(filter.get("source"), Some(Bson::String(value)) if value == "ISODate(\"2026-06-10T13:59:31.287Z\")")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_indexes_command_rejects_key_inside_options() {
|
||||
let error = create_indexes_command("users", r#"{"email":1}"#, Some(r#"{"key":{"other":1}}"#)).unwrap_err();
|
||||
|
||||
assert!(error.contains("cannot contain \"key\""), "{error}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_indexes_command_rejects_empty_or_non_string_names() {
|
||||
for options in [r#"{"name":""}"#, r#"{"name":" "}"#, r#"{"name":null}"#, r#"{"name":1}"#] {
|
||||
let error = create_indexes_command("users", r#"{"email":1}"#, Some(options)).unwrap_err();
|
||||
assert!(error.contains("non-empty string"), "{error}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_drop_indexes_value_validates_drop_index_arguments() {
|
||||
assert!(matches!(
|
||||
@@ -2642,6 +2887,50 @@ mod tests {
|
||||
assert!(invalid_array.contains("arrays of string index names"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn drop_indexes_validation_rejects_the_default_id_index() {
|
||||
for (indexes_json, single) in [
|
||||
(Some(r#""_id_""#), true),
|
||||
(Some(r#""_id_""#), false),
|
||||
(Some(r#"{"_id":1}"#), true),
|
||||
(Some(r#"{"_id":1}"#), false),
|
||||
(Some(r#"{"_id":1.0}"#), true),
|
||||
(Some(r#"{"_id":{"$numberInt":"1"}}"#), true),
|
||||
(Some(r#"{"_id":{"$numberLong":"1"}}"#), false),
|
||||
(Some(r#"{"_id":{"$numberDouble":"1.0"}}"#), false),
|
||||
(Some(r#"{"_id":{"$numberDecimal":"1.0"}}"#), true),
|
||||
(Some(r#"["email_1","_id_"]"#), false),
|
||||
] {
|
||||
let error = validate_drop_indexes_request(indexes_json, single).unwrap_err();
|
||||
assert!(error.contains("_id_"), "{error}");
|
||||
}
|
||||
|
||||
// MongoDB defines dropIndexes("*") to retain the default _id_ index.
|
||||
assert!(matches!(validate_drop_indexes_request(Some(r#""*""#), false), Ok(())));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serial_drop_index_names_is_portable_and_prevalidates_all_names() {
|
||||
assert_eq!(
|
||||
serial_drop_index_names(Some(r#"["email_1","email_1","createdAt_-1"]"#), false).unwrap(),
|
||||
Some(vec!["email_1".to_string(), "createdAt_-1".to_string()])
|
||||
);
|
||||
assert_eq!(serial_drop_index_names(Some(r#""*""#), false).unwrap(), None);
|
||||
|
||||
let error = serial_drop_index_names(Some(r#"["email_1","_id_"]"#), false).unwrap_err();
|
||||
assert!(error.contains("_id_"), "{error}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn serial_drop_indexes_fallback_is_limited_to_pre_42_servers() {
|
||||
for version in ["3.4.24", "4.0.28", "MongoDB 3.6.23"] {
|
||||
assert!(mongo_server_requires_serial_drop_indexes(version), "{version}");
|
||||
}
|
||||
for version in ["4.2.0", "4.4.29", "5.0.0-rc0", "8.0.1", "unknown", "4"] {
|
||||
assert!(!mongo_server_requires_serial_drop_indexes(version), "{version}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn diff_dropped_index_names_reports_removed_indexes() {
|
||||
let before = vec![
|
||||
|
||||
@@ -98,6 +98,37 @@ fn sort_mongo_collection_specs(
|
||||
specs
|
||||
}
|
||||
|
||||
/// Decode both generations of the Legacy Agent response. Existing installed
|
||||
/// Agents return names, while current Agents opt into `name` + `kind` specs.
|
||||
fn mongo_collection_specs_from_agent_response(
|
||||
value: serde_json::Value,
|
||||
) -> Result<Vec<mongo_driver::MongoCollectionSpec>, String> {
|
||||
let values =
|
||||
value.as_array().ok_or_else(|| "Invalid MongoDB legacy collection list: expected an array".to_string())?;
|
||||
|
||||
values
|
||||
.iter()
|
||||
.map(|value| match value {
|
||||
serde_json::Value::String(name) => Ok(mongo_driver::MongoCollectionSpec {
|
||||
name: name.clone(),
|
||||
kind: mongo_driver::MongoCollectionKind::Collection,
|
||||
}),
|
||||
serde_json::Value::Object(spec) => {
|
||||
let name = spec
|
||||
.get("name")
|
||||
.and_then(serde_json::Value::as_str)
|
||||
.ok_or_else(|| "Invalid MongoDB legacy collection spec: name is required".to_string())?;
|
||||
let kind = spec.get("kind").or_else(|| spec.get("type")).and_then(serde_json::Value::as_str);
|
||||
Ok(mongo_driver::MongoCollectionSpec {
|
||||
name: name.to_string(),
|
||||
kind: mongo_driver::MongoCollectionKind::from_metadata_kind(kind),
|
||||
})
|
||||
}
|
||||
_ => Err("Invalid MongoDB legacy collection list entry".to_string()),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub(crate) fn mongo_gridfs_bucket_names(names: &[String]) -> Vec<String> {
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
@@ -225,14 +256,12 @@ pub async fn list_collections_core(
|
||||
PoolKind::VectorDb(client) => vector_driver::list_collections_with_db(client, database).await,
|
||||
PoolKind::Agent(client) => {
|
||||
let mut client = client.lock().await;
|
||||
let names = sort_names(client.mongo_list_collections(database).await?);
|
||||
let specs = sort_mongo_collection_specs(mongo_collection_specs_from_agent_response(
|
||||
client.mongo_list_collection_specs(database).await?,
|
||||
)?);
|
||||
let names: Vec<String> = specs.iter().map(|spec| spec.name.clone()).collect();
|
||||
let mut infos = mongo_bucket_infos(&names);
|
||||
// Legacy agent returns names only; treat every entry as a plain collection.
|
||||
infos.extend(
|
||||
names
|
||||
.into_iter()
|
||||
.map(|name| mongo_collection_info(name, mongo_driver::MongoCollectionKind::Collection)),
|
||||
);
|
||||
infos.extend(specs.into_iter().map(|spec| mongo_collection_info(spec.name, spec.kind)));
|
||||
Ok(infos)
|
||||
}
|
||||
_ => Err("Not a MongoDB/Elasticsearch/vector connection".to_string()),
|
||||
@@ -581,9 +610,11 @@ pub async fn delete_document_core_with_type(
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
fallback_mongo_database, filter_and_sort_gridfs_bucket_infos, mongo_gridfs_bucket_names,
|
||||
mongo_list_databases_unauthorized, parse_gridfs_bucket_sort, sort_names, MongoGridFsBucketInfo,
|
||||
fallback_mongo_database, filter_and_sort_gridfs_bucket_infos, mongo_collection_specs_from_agent_response,
|
||||
mongo_gridfs_bucket_names, mongo_list_databases_unauthorized, parse_gridfs_bucket_sort, sort_names,
|
||||
MongoGridFsBucketInfo,
|
||||
};
|
||||
use crate::db::mongo_driver::MongoCollectionKind;
|
||||
|
||||
#[test]
|
||||
fn sorts_names_case_insensitively() {
|
||||
@@ -628,6 +659,27 @@ mod tests {
|
||||
assert_eq!(buckets, vec!["orders".to_string(), "reports".to_string()]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn decodes_legacy_collection_names_and_type_aware_specs() {
|
||||
let specs = mongo_collection_specs_from_agent_response(serde_json::json!([
|
||||
"orders",
|
||||
{ "name": "report_view", "kind": "view" },
|
||||
{ "name": "metrics", "kind": "timeseries" },
|
||||
{ "name": "future_kind", "kind": "unknown" }
|
||||
]))
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
specs.into_iter().map(|spec| (spec.name, spec.kind)).collect::<Vec<_>>(),
|
||||
vec![
|
||||
("orders".to_string(), MongoCollectionKind::Collection),
|
||||
("report_view".to_string(), MongoCollectionKind::View),
|
||||
("metrics".to_string(), MongoCollectionKind::Timeseries),
|
||||
("future_kind".to_string(), MongoCollectionKind::Collection),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn filters_gridfs_buckets_by_case_insensitive_name_match() {
|
||||
let buckets = filter_and_sort_gridfs_bucket_infos(
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
use crate::connection::{AppState, PoolKind};
|
||||
use crate::db::mongo_driver::{self, MongoCollectionStatsResult, MongoDocumentResult, MongoDropIndexesResult};
|
||||
use crate::db::agent_driver::AgentCapability;
|
||||
use crate::db::mongo_driver::{
|
||||
self, MongoCollectionStatsResult, MongoDocumentResult, MongoDropIndexFailure, MongoDropIndexesResult,
|
||||
};
|
||||
use crate::document_ops::CollectionInfo;
|
||||
|
||||
async fn ensure_document_pool(state: &AppState, connection_id: &str) -> Result<(), String> {
|
||||
@@ -29,11 +32,22 @@ pub async fn mongo_create_database_core(state: &AppState, connection_id: &str, d
|
||||
}
|
||||
|
||||
pub async fn mongo_drop_database_core(state: &AppState, connection_id: &str, database: &str) -> Result<(), String> {
|
||||
mongo_driver::validate_mongo_namespace_name(database, "Database")?;
|
||||
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::drop_database(client, database).await,
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support drop database".to_string()),
|
||||
PoolKind::Agent(client) => {
|
||||
let mut client = client.lock().await;
|
||||
if !client.supports_capability(AgentCapability::MongoDropDatabase) {
|
||||
return Err(
|
||||
"MongoDB Legacy Agent does not support drop database; upgrade or reinstall the MongoDB Legacy driver"
|
||||
.to_string(),
|
||||
);
|
||||
}
|
||||
let _: serde_json::Value = client.mongo_drop_database(serde_json::json!({ "database": database })).await?;
|
||||
Ok(())
|
||||
}
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
@@ -44,6 +58,8 @@ pub async fn mongo_drop_collection_core(
|
||||
database: &str,
|
||||
collection: &str,
|
||||
) -> Result<(), String> {
|
||||
mongo_driver::validate_mongo_namespace_name(database, "Database")?;
|
||||
mongo_driver::validate_mongo_namespace_name(collection, "Collection")?;
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
@@ -300,6 +316,11 @@ pub async fn mongo_create_index_core(
|
||||
keys_json: &str,
|
||||
options_json: Option<&str>,
|
||||
) -> Result<String, String> {
|
||||
// The Legacy Agent receives the original JSON, so validate the shared
|
||||
// createIndexes contract before selecting a driver implementation.
|
||||
mongo_driver::validate_mongo_namespace_name(database, "Database")?;
|
||||
mongo_driver::validate_mongo_namespace_name(collection, "Collection")?;
|
||||
mongo_driver::validate_create_index_request(keys_json, options_json)?;
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
@@ -316,7 +337,12 @@ pub async fn mongo_create_index_core(
|
||||
"options_json": options_json,
|
||||
}))
|
||||
.await?;
|
||||
Ok(result.get("name").and_then(|value| value.as_str()).unwrap_or("").to_string())
|
||||
result
|
||||
.get("name")
|
||||
.and_then(serde_json::Value::as_str)
|
||||
.filter(|name| !name.is_empty())
|
||||
.map(str::to_string)
|
||||
.ok_or_else(|| "MongoDB legacy agent returned no created index name".to_string())
|
||||
}
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
@@ -330,6 +356,52 @@ pub async fn mongo_drop_indexes_core(
|
||||
indexes_json: Option<&str>,
|
||||
single: bool,
|
||||
) -> Result<MongoDropIndexesResult, String> {
|
||||
mongo_driver::validate_mongo_namespace_name(database, "Database")?;
|
||||
mongo_driver::validate_mongo_namespace_name(collection, "Collection")?;
|
||||
let serial_names = mongo_driver::serial_drop_index_names(indexes_json, single)?;
|
||||
if let Some(names) = serial_names {
|
||||
let requires_serial_fallback = match mongo_server_version_core(state, connection_id, database).await {
|
||||
Ok(version) => mongo_driver::mongo_server_requires_serial_drop_indexes(&version),
|
||||
Err(error) => {
|
||||
log::warn!(
|
||||
"[mongo][drop-indexes] server version unavailable; preserving array command semantics: {error}"
|
||||
);
|
||||
false
|
||||
}
|
||||
};
|
||||
if requires_serial_fallback {
|
||||
let mut dropped_names = Vec::new();
|
||||
let mut failures = Vec::new();
|
||||
for name in names {
|
||||
let index_json = serde_json::to_string(&name).map_err(|error| error.to_string())?;
|
||||
match mongo_drop_indexes_once_core(state, connection_id, database, collection, Some(&index_json), true)
|
||||
.await
|
||||
{
|
||||
Ok(result) => {
|
||||
dropped_names.extend(result.dropped_names);
|
||||
failures.extend(result.failures);
|
||||
}
|
||||
Err(message) => failures.push(MongoDropIndexFailure { name, message }),
|
||||
}
|
||||
}
|
||||
return Ok(MongoDropIndexesResult { affected_rows: dropped_names.len() as u64, dropped_names, failures });
|
||||
}
|
||||
}
|
||||
|
||||
mongo_drop_indexes_once_core(state, connection_id, database, collection, indexes_json, single).await
|
||||
}
|
||||
|
||||
async fn mongo_drop_indexes_once_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
collection: &str,
|
||||
indexes_json: Option<&str>,
|
||||
single: bool,
|
||||
) -> Result<MongoDropIndexesResult, String> {
|
||||
// Apply the same argument and default-index protection before dispatching
|
||||
// to Native MongoDB or the Legacy Agent.
|
||||
mongo_driver::validate_drop_indexes_request(indexes_json, single)?;
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
@@ -550,3 +622,392 @@ pub async fn mongo_find_one_and_delete_core(
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[cfg(unix)]
|
||||
use crate::db::agent_driver::{AgentDriverClient, AgentLaunchSpec};
|
||||
#[cfg(unix)]
|
||||
use crate::models::connection::ConnectionConfig;
|
||||
#[cfg(unix)]
|
||||
use crate::storage::Storage;
|
||||
|
||||
#[cfg(unix)]
|
||||
async fn legacy_mongo_state(
|
||||
expected_method: &str,
|
||||
expected_params: serde_json::Value,
|
||||
expected_result: serde_json::Value,
|
||||
) -> (AppState, tempfile::TempDir) {
|
||||
legacy_mongo_state_with_options(
|
||||
expected_method,
|
||||
expected_params,
|
||||
expected_result,
|
||||
&[AgentCapability::MongoDropDatabase.as_str()],
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
async fn legacy_mongo_state_with_capabilities(
|
||||
expected_method: &str,
|
||||
expected_params: serde_json::Value,
|
||||
expected_result: serde_json::Value,
|
||||
capabilities: &[&str],
|
||||
) -> (AppState, tempfile::TempDir) {
|
||||
legacy_mongo_state_with_options(expected_method, expected_params, expected_result, capabilities, None, None)
|
||||
.await
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
async fn legacy_mongo_state_with_server_version(
|
||||
expected_method: &str,
|
||||
expected_params: serde_json::Value,
|
||||
expected_result: serde_json::Value,
|
||||
server_version: &str,
|
||||
) -> (AppState, tempfile::TempDir) {
|
||||
legacy_mongo_state_with_options(
|
||||
expected_method,
|
||||
expected_params,
|
||||
expected_result,
|
||||
&[AgentCapability::MongoDropDatabase.as_str()],
|
||||
Some(server_version),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
async fn legacy_mongo_state_with_server_error(
|
||||
expected_method: &str,
|
||||
expected_params: serde_json::Value,
|
||||
expected_error: &str,
|
||||
server_version: &str,
|
||||
) -> (AppState, tempfile::TempDir) {
|
||||
legacy_mongo_state_with_options(
|
||||
expected_method,
|
||||
expected_params,
|
||||
serde_json::Value::Null,
|
||||
&[AgentCapability::MongoDropDatabase.as_str()],
|
||||
Some(server_version),
|
||||
Some(expected_error),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
async fn legacy_mongo_state_with_options(
|
||||
expected_method: &str,
|
||||
expected_params: serde_json::Value,
|
||||
expected_result: serde_json::Value,
|
||||
capabilities: &[&str],
|
||||
server_version: Option<&str>,
|
||||
expected_error: Option<&str>,
|
||||
) -> (AppState, tempfile::TempDir) {
|
||||
use std::io::Write;
|
||||
|
||||
let directory = tempfile::tempdir().unwrap();
|
||||
let mut script = tempfile::NamedTempFile::new_in(directory.path()).unwrap();
|
||||
let expected_method = serde_json::to_string(expected_method).unwrap();
|
||||
let expected_params = serde_json::to_string(&serde_json::to_string(&expected_params).unwrap()).unwrap();
|
||||
let expected_result = serde_json::to_string(&serde_json::to_string(&expected_result).unwrap()).unwrap();
|
||||
let capabilities = serde_json::to_string(capabilities).unwrap();
|
||||
let python_optional_string = |value: Option<&str>| {
|
||||
value.map(|value| serde_json::to_string(value).unwrap()).unwrap_or_else(|| "None".to_string())
|
||||
};
|
||||
let server_version = python_optional_string(server_version);
|
||||
let expected_error = python_optional_string(expected_error);
|
||||
write!(
|
||||
script,
|
||||
r#"import json
|
||||
import sys
|
||||
|
||||
EXPECTED_METHOD = {expected_method}
|
||||
EXPECTED_PARAMS = json.loads({expected_params})
|
||||
EXPECTED_RESULT = json.loads({expected_result})
|
||||
CAPABILITIES = {capabilities}
|
||||
SERVER_VERSION = {server_version}
|
||||
EXPECTED_ERROR = {expected_error}
|
||||
|
||||
print(json.dumps({{"ready": True}}), flush=True)
|
||||
for line in sys.stdin:
|
||||
request = json.loads(line)
|
||||
if request.get("method") == "handshake":
|
||||
result = {{"protocolVersion": 1, "agentProtocolVersion": 1, "capabilities": CAPABILITIES}}
|
||||
print(json.dumps({{"jsonrpc": "2.0", "id": request["id"], "result": result}}), flush=True)
|
||||
continue
|
||||
if request.get("method") == "validate_connection":
|
||||
print(json.dumps({{"jsonrpc": "2.0", "id": request["id"], "result": {{"ok": True}}}}), flush=True)
|
||||
continue
|
||||
if request.get("method") == "server_version" and SERVER_VERSION is not None:
|
||||
print(json.dumps({{"jsonrpc": "2.0", "id": request["id"], "result": SERVER_VERSION}}), flush=True)
|
||||
continue
|
||||
if request.get("method") != EXPECTED_METHOD or request.get("params") != EXPECTED_PARAMS:
|
||||
print(json.dumps({{"jsonrpc": "2.0", "id": request["id"], "error": {{"code": -1, "message": "unexpected MongoDB RPC"}}}}), flush=True)
|
||||
continue
|
||||
if EXPECTED_ERROR is not None:
|
||||
print(json.dumps({{"jsonrpc": "2.0", "id": request["id"], "error": {{"code": -1, "message": EXPECTED_ERROR}}}}), flush=True)
|
||||
else:
|
||||
print(json.dumps({{"jsonrpc": "2.0", "id": request["id"], "result": EXPECTED_RESULT}}), flush=True)
|
||||
"#
|
||||
)
|
||||
.unwrap();
|
||||
script.flush().unwrap();
|
||||
// Keep the script available until the spawned interpreter has opened it.
|
||||
let (_, script_path) = script.keep().unwrap();
|
||||
|
||||
let mut client = AgentDriverClient::spawn(
|
||||
AgentLaunchSpec::new("python3").with_args([script_path.to_string_lossy().to_string()]),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
client.try_optional_handshake("test").await.unwrap();
|
||||
let storage = Storage::open(&directory.path().join("storage.db")).await.unwrap();
|
||||
let state = AppState::new(storage);
|
||||
let config: ConnectionConfig = serde_json::from_value(serde_json::json!({
|
||||
"id": "legacy",
|
||||
"name": "Legacy MongoDB",
|
||||
"db_type": "mongodb",
|
||||
"driver_profile": "mongodb-legacy",
|
||||
"host": "localhost",
|
||||
"port": 27017,
|
||||
"username": "",
|
||||
"password": "",
|
||||
"database": null,
|
||||
}))
|
||||
.unwrap();
|
||||
state.configs.write().await.insert("legacy".to_string(), config);
|
||||
state.connections.write().await.insert("legacy".to_string(), PoolKind::agent(client));
|
||||
(state, directory)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_drop_database_routes_legacy_connections_to_the_agent() {
|
||||
let (state, _directory) = legacy_mongo_state(
|
||||
"drop_database",
|
||||
serde_json::json!({ "database": "app" }),
|
||||
serde_json::json!({ "ok": true }),
|
||||
)
|
||||
.await;
|
||||
|
||||
mongo_drop_database_core(&state, "legacy", "app").await.unwrap();
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_drop_database_requires_an_explicit_legacy_agent_capability() {
|
||||
let (state, _directory) = legacy_mongo_state_with_capabilities(
|
||||
"drop_database",
|
||||
serde_json::json!({ "database": "app" }),
|
||||
serde_json::json!({ "ok": true }),
|
||||
&[],
|
||||
)
|
||||
.await;
|
||||
|
||||
let error = mongo_drop_database_core(&state, "legacy", "app").await.unwrap_err();
|
||||
|
||||
assert!(error.contains("upgrade or reinstall"), "{error}");
|
||||
assert!(!error.contains("Unknown method"), "{error}");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_drop_collection_routes_legacy_connections_to_the_agent() {
|
||||
let (state, _directory) = legacy_mongo_state(
|
||||
"drop_collection",
|
||||
serde_json::json!({
|
||||
"database": "app",
|
||||
"collection": "users",
|
||||
}),
|
||||
serde_json::json!({ "ok": true }),
|
||||
)
|
||||
.await;
|
||||
|
||||
mongo_drop_collection_core(&state, "legacy", "app", "users").await.unwrap();
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_list_collections_requests_legacy_collection_types() {
|
||||
let (state, _directory) = legacy_mongo_state(
|
||||
"list_collections",
|
||||
serde_json::json!({ "database": "app", "include_types": true }),
|
||||
serde_json::json!([
|
||||
{ "name": "report_view", "kind": "view" },
|
||||
{ "name": "metrics", "kind": "timeseries" },
|
||||
"orders"
|
||||
]),
|
||||
)
|
||||
.await;
|
||||
|
||||
let collections = mongo_list_collections_core(&state, "legacy", "app").await.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
collections.into_iter().map(|collection| (collection.name, collection.kind)).collect::<Vec<_>>(),
|
||||
vec![
|
||||
("metrics".to_string(), Some("timeseries".to_string())),
|
||||
("orders".to_string(), Some("collection".to_string())),
|
||||
("report_view".to_string(), Some("view".to_string())),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_create_index_routes_legacy_connections_to_the_agent() {
|
||||
let keys_json = r#"{"email":1}"#;
|
||||
let options_json = r#"{"name":"email_1","unique":true}"#;
|
||||
let (state, _directory) = legacy_mongo_state(
|
||||
"create_index",
|
||||
serde_json::json!({
|
||||
"database": "app",
|
||||
"collection": "users",
|
||||
"keys_json": keys_json,
|
||||
"options_json": options_json,
|
||||
}),
|
||||
serde_json::json!({ "name": "email_1" }),
|
||||
)
|
||||
.await;
|
||||
|
||||
let name =
|
||||
mongo_create_index_core(&state, "legacy", "app", "users", keys_json, Some(options_json)).await.unwrap();
|
||||
|
||||
assert_eq!(name, "email_1");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_create_index_requires_a_name_from_the_legacy_agent() {
|
||||
let (state, _directory) = legacy_mongo_state(
|
||||
"create_index",
|
||||
serde_json::json!({
|
||||
"database": "app",
|
||||
"collection": "users",
|
||||
"keys_json": "{\"email\":1}",
|
||||
"options_json": null,
|
||||
}),
|
||||
serde_json::json!({}),
|
||||
)
|
||||
.await;
|
||||
|
||||
let error =
|
||||
mongo_create_index_core(&state, "legacy", "app", "users", r#"{"email":1}"#, None).await.unwrap_err();
|
||||
|
||||
assert!(error.contains("no created index name"), "{error}");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_create_index_rejects_key_options_before_agent_dispatch() {
|
||||
let (state, _directory) = legacy_mongo_state("unexpected", serde_json::json!({}), serde_json::json!({})).await;
|
||||
|
||||
let error =
|
||||
mongo_create_index_core(&state, "legacy", "app", "users", r#"{"email":1}"#, Some(r#"{"key":{"other":1}}"#))
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(error.contains("cannot contain \"key\""), "{error}");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_create_index_rejects_an_empty_name_before_agent_dispatch() {
|
||||
let (state, _directory) = legacy_mongo_state("unexpected", serde_json::json!({}), serde_json::json!({})).await;
|
||||
|
||||
let error = mongo_create_index_core(&state, "legacy", "app", "users", r#"{"email":1}"#, Some(r#"{"name":""}"#))
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(error.contains("non-empty string"), "{error}");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_drop_index_routes_legacy_connections_to_the_agent() {
|
||||
let (state, _directory) = legacy_mongo_state(
|
||||
"drop_indexes",
|
||||
serde_json::json!({
|
||||
"database": "app",
|
||||
"collection": "users",
|
||||
"indexes_json": "\"email_1\"",
|
||||
"single": true,
|
||||
}),
|
||||
serde_json::json!({ "affected_rows": 1, "dropped_names": ["email_1"] }),
|
||||
)
|
||||
.await;
|
||||
|
||||
let result =
|
||||
mongo_drop_indexes_core(&state, "legacy", "app", "users", Some(r#""email_1""#), true).await.unwrap();
|
||||
|
||||
assert_eq!(result.dropped_names, ["email_1"]);
|
||||
assert_eq!(result.affected_rows, 1);
|
||||
assert!(result.failures.is_empty());
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_drop_indexes_returns_partial_results_for_mongodb_34_agents() {
|
||||
let (state, _directory) = legacy_mongo_state_with_server_version(
|
||||
"drop_indexes",
|
||||
serde_json::json!({
|
||||
"database": "app",
|
||||
"collection": "users",
|
||||
"indexes_json": "\"email_1\"",
|
||||
"single": true,
|
||||
}),
|
||||
serde_json::json!({ "affected_rows": 1, "dropped_names": ["email_1"] }),
|
||||
"3.4.24",
|
||||
)
|
||||
.await;
|
||||
|
||||
let result =
|
||||
mongo_drop_indexes_core(&state, "legacy", "app", "users", Some(r#"["email_1","missing_1"]"#), false)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result.dropped_names, ["email_1"]);
|
||||
assert_eq!(result.affected_rows, 1);
|
||||
assert_eq!(result.failures.len(), 1);
|
||||
assert_eq!(result.failures[0].name, "missing_1");
|
||||
assert!(result.failures[0].message.contains("unexpected MongoDB RPC"));
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_drop_indexes_preserves_modern_agent_batch_failure_semantics() {
|
||||
let (state, _directory) = legacy_mongo_state_with_server_error(
|
||||
"drop_indexes",
|
||||
serde_json::json!({
|
||||
"database": "app",
|
||||
"collection": "users",
|
||||
"indexes_json": "[\"email_1\",\"missing_1\"]",
|
||||
"single": false,
|
||||
}),
|
||||
"index not found; no indexes dropped",
|
||||
"4.2.0",
|
||||
)
|
||||
.await;
|
||||
|
||||
let error =
|
||||
mongo_drop_indexes_core(&state, "legacy", "app", "users", Some(r#"["email_1","missing_1"]"#), false)
|
||||
.await
|
||||
.unwrap_err();
|
||||
|
||||
assert!(error.contains("no indexes dropped"), "{error}");
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn mongo_drop_indexes_rejects_the_default_id_index_before_agent_dispatch() {
|
||||
let (state, _directory) = legacy_mongo_state("unexpected", serde_json::json!({}), serde_json::json!({})).await;
|
||||
|
||||
let error =
|
||||
mongo_drop_indexes_core(&state, "legacy", "app", "users", Some(r#""_id_""#), true).await.unwrap_err();
|
||||
|
||||
assert!(error.contains("_id_"), "{error}");
|
||||
}
|
||||
}
|
||||
@@ -2200,10 +2200,8 @@ fn ensure_mcp_connection_change_allowed_in_tx(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn persist_connection_in_tx(tx: &rusqlite::Transaction<'_>, config: &ConnectionConfig) -> Result<(), String> {
|
||||
let config = config.clone().canonicalized();
|
||||
let config_id = config.id.clone();
|
||||
let mut sanitized = config.clone();
|
||||
fn sanitized_connection_config(config: &ConnectionConfig) -> ConnectionConfig {
|
||||
let mut sanitized = config.clone().canonicalized();
|
||||
sanitized.password = String::new();
|
||||
scrub_transport_layer_secrets(&mut sanitized);
|
||||
sanitized.redis_sentinel_password = String::new();
|
||||
@@ -2212,6 +2210,13 @@ fn persist_connection_in_tx(tx: &rusqlite::Transaction<'_>, config: &ConnectionC
|
||||
scrub_mq_auth_secrets(&mut sanitized);
|
||||
scrub_mq_token_signing_secret(&mut sanitized);
|
||||
scrub_nacos_auth_secrets(&mut sanitized);
|
||||
sanitized
|
||||
}
|
||||
|
||||
fn persist_connection_in_tx(tx: &rusqlite::Transaction<'_>, config: &ConnectionConfig) -> Result<(), String> {
|
||||
let config = config.clone().canonicalized();
|
||||
let config_id = config.id.clone();
|
||||
let sanitized = sanitized_connection_config(&config);
|
||||
let json = serde_json::to_string(&sanitized).map_err(|e| e.to_string())?;
|
||||
|
||||
tx.execute("INSERT INTO connections (id, config_json) VALUES (?1, ?2)", params![config_id, json])
|
||||
@@ -2433,6 +2438,65 @@ impl Storage {
|
||||
.await
|
||||
}
|
||||
|
||||
/// Update only the persisted driver identity for an existing connection.
|
||||
/// This is used after runtime driver fallback and deliberately leaves all
|
||||
/// other connection metadata and separately stored secrets untouched.
|
||||
pub async fn save_connection_driver_profile(
|
||||
&self,
|
||||
expected_config: &ConnectionConfig,
|
||||
driver_profile: Option<String>,
|
||||
driver_label: Option<String>,
|
||||
) -> Result<bool, String> {
|
||||
let expected_config = sanitized_connection_config(expected_config);
|
||||
let connection_id = expected_config.id.clone();
|
||||
self.with_conn(move |conn| {
|
||||
let Some(json) = conn
|
||||
.query_row("SELECT config_json FROM connections WHERE id = ?1", [&connection_id], |row| {
|
||||
row.get::<_, String>(0)
|
||||
})
|
||||
.optional()
|
||||
.map_err(|error| error.to_string())?
|
||||
else {
|
||||
return Ok(false);
|
||||
};
|
||||
let current: ConnectionConfig = serde_json::from_str(&json).map_err(|error| error.to_string())?;
|
||||
let mut current_identity = sanitized_connection_config(¤t);
|
||||
let mut expected_identity = expected_config.clone();
|
||||
current_identity.note.clear();
|
||||
current_identity.database_info = None;
|
||||
expected_identity.note.clear();
|
||||
expected_identity.database_info = None;
|
||||
|
||||
let identity_matches = current_identity == expected_identity
|
||||
|| (current_identity.driver_profile == driver_profile
|
||||
&& current_identity.driver_label == driver_label
|
||||
&& {
|
||||
current_identity.driver_profile = expected_identity.driver_profile.clone();
|
||||
current_identity.driver_label = expected_identity.driver_label.clone();
|
||||
current_identity == expected_identity
|
||||
});
|
||||
if !identity_matches {
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
if current.driver_profile == driver_profile && current.driver_label == driver_label {
|
||||
return Ok(true);
|
||||
}
|
||||
|
||||
let mut updated = sanitized_connection_config(¤t);
|
||||
updated.driver_profile = driver_profile;
|
||||
updated.driver_label = driver_label;
|
||||
let updated_json = serde_json::to_string(&updated).map_err(|error| error.to_string())?;
|
||||
conn.execute(
|
||||
"UPDATE connections SET config_json = ?1 WHERE id = ?2 AND config_json = ?3",
|
||||
params![updated_json, connection_id, json],
|
||||
)
|
||||
.map(|updated| updated > 0)
|
||||
.map_err(|error| error.to_string())
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn load_connections(&self) -> Result<Vec<ConnectionConfig>, String> {
|
||||
let rows: Vec<(String, String)> = self
|
||||
.with_conn(|conn| {
|
||||
@@ -4357,6 +4421,72 @@ mod tests {
|
||||
let _ = std::fs::remove_file(path);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn save_connection_driver_profile_updates_only_the_target_metadata() {
|
||||
let path = temp_db_path("connection-driver-profile");
|
||||
let storage = Storage::open(&path).await.unwrap();
|
||||
let target = mq_connection("target", "target-secret");
|
||||
let untouched = mq_connection("untouched", "untouched-secret");
|
||||
storage.save_connections(&[target.clone(), untouched.clone()]).await.unwrap();
|
||||
|
||||
assert!(storage
|
||||
.save_connection_driver_profile(
|
||||
&target,
|
||||
Some("mongodb-legacy".to_string()),
|
||||
Some("MongoDB (Legacy)".to_string()),
|
||||
)
|
||||
.await
|
||||
.unwrap());
|
||||
let mut wrong_type = untouched.clone();
|
||||
wrong_type.db_type = DatabaseType::MongoDb;
|
||||
assert!(!storage
|
||||
.save_connection_driver_profile(&wrong_type, Some("mongodb-legacy".to_string()), None,)
|
||||
.await
|
||||
.unwrap());
|
||||
let mut missing = wrong_type;
|
||||
missing.id = "missing".to_string();
|
||||
assert!(!storage
|
||||
.save_connection_driver_profile(&missing, Some("mongodb-legacy".to_string()), None)
|
||||
.await
|
||||
.unwrap());
|
||||
|
||||
let loaded = storage.load_connections().await.unwrap();
|
||||
let target = loaded.iter().find(|config| config.id == "target").unwrap();
|
||||
assert_eq!(target.driver_profile.as_deref(), Some("mongodb-legacy"));
|
||||
assert_eq!(target.driver_label.as_deref(), Some("MongoDB (Legacy)"));
|
||||
assert_eq!(mq_token(target), Some("target-secret"));
|
||||
assert_eq!(loaded.iter().find(|config| config.id == "untouched"), Some(&untouched));
|
||||
|
||||
let _ = std::fs::remove_file(path);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn save_connection_driver_profile_rejects_a_stale_connection_config() {
|
||||
let path = temp_db_path("connection-driver-profile-stale");
|
||||
let storage = Storage::open(&path).await.unwrap();
|
||||
let original = mq_connection("target", "target-secret");
|
||||
storage.save_connections(std::slice::from_ref(&original)).await.unwrap();
|
||||
|
||||
let mut replacement = original.clone();
|
||||
replacement.host = "replacement.example.com".to_string();
|
||||
replacement.name = "Replacement".to_string();
|
||||
storage.save_connections(std::slice::from_ref(&replacement)).await.unwrap();
|
||||
|
||||
assert!(!storage
|
||||
.save_connection_driver_profile(
|
||||
&original,
|
||||
Some("mongodb-legacy".to_string()),
|
||||
Some("MongoDB (Legacy)".to_string()),
|
||||
)
|
||||
.await
|
||||
.unwrap());
|
||||
|
||||
let loaded = storage.load_connections().await.unwrap();
|
||||
assert_eq!(loaded, vec![replacement]);
|
||||
|
||||
let _ = std::fs::remove_file(path);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn save_connections_moves_mq_auth_token_to_secret_table_and_restores_it() {
|
||||
let path = temp_db_path("mq-token-secrets");
|
||||
|
||||
Reference in new issue
Block a user