feat(mongo): add GridFS manager for MongoDB
* feat(mongo): browse GridFS buckets and files Refs #2342 * feat(mongo): add GridFS manager for MongoDB * chore(i18n): autofill new translations --------- Co-authored-by: dbx-i18n-bot <github-actions[bot]@users.noreply.github.com> Co-authored-by: t8y2 <1156263951@qq.com>
This commit is contained in:
35 files changed
+1959
-39
No files matched your search
@@ -1,13 +1,14 @@
|
||||
use mongodb::{
|
||||
bson::{doc, oid::ObjectId, Bson, DateTime, Document},
|
||||
options::{ClientOptions, IndexOptions},
|
||||
Client, IndexModel,
|
||||
options::{ClientOptions, GridFsBucketOptions, IndexOptions},
|
||||
Client, Database, IndexModel,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use super::with_connection_timeout;
|
||||
use crate::document_ops::{MongoGridFsBucketInfo, MongoGridFsFileInfo};
|
||||
use crate::types::IndexInfo;
|
||||
use futures::TryStreamExt;
|
||||
use futures::{io::AsyncReadExt, io::AsyncWriteExt, TryStreamExt};
|
||||
use percent_encoding::percent_decode_str;
|
||||
use std::{collections::HashSet, time::Duration};
|
||||
|
||||
@@ -143,6 +144,253 @@ pub async fn list_collections(client: &Client, database: &str) -> Result<Vec<Str
|
||||
client.database(database).list_collection_names().await.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
pub async fn list_gridfs_files(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
bucket: &str,
|
||||
) -> Result<Vec<MongoGridFsFileInfo>, String> {
|
||||
let bucket = normalized_gridfs_bucket_name(bucket)?;
|
||||
let collection_name = format!("{bucket}.files");
|
||||
let collection = client.database(database).collection::<Document>(&collection_name);
|
||||
let mut cursor =
|
||||
collection.find(doc! {}).sort(doc! { "uploadDate": -1_i32, "_id": -1_i32 }).await.map_err(|e| e.to_string())?;
|
||||
let mut files = Vec::new();
|
||||
while cursor.advance().await.map_err(|e| e.to_string())? {
|
||||
let doc = cursor.deserialize_current().map_err(|e| e.to_string())?;
|
||||
files.push(gridfs_file_info_from_document(&doc));
|
||||
}
|
||||
Ok(files)
|
||||
}
|
||||
|
||||
pub async fn gridfs_bucket_summary(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
bucket: &str,
|
||||
) -> Result<MongoGridFsBucketInfo, String> {
|
||||
let bucket = normalized_gridfs_bucket_name(bucket)?;
|
||||
let collection = client.database(database).collection::<Document>(&format!("{bucket}.files"));
|
||||
let mut cursor = collection
|
||||
.aggregate(vec![doc! {
|
||||
"$group": {
|
||||
"_id": Bson::Null,
|
||||
"fileCount": { "$sum": 1_i32 },
|
||||
"totalBytes": { "$sum": "$length" },
|
||||
}
|
||||
}])
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let mut file_count = 0_u64;
|
||||
let mut total_bytes = 0_i64;
|
||||
if cursor.advance().await.map_err(|e| e.to_string())? {
|
||||
let doc = cursor.deserialize_current().map_err(|e| e.to_string())?;
|
||||
file_count =
|
||||
doc.get_i64("fileCount").or_else(|_| doc.get_i32("fileCount").map(i64::from)).unwrap_or(0).max(0) as u64;
|
||||
total_bytes = doc.get_i64("totalBytes").or_else(|_| doc.get_i32("totalBytes").map(i64::from)).unwrap_or(0);
|
||||
}
|
||||
|
||||
Ok(MongoGridFsBucketInfo { name: bucket, file_count, total_bytes })
|
||||
}
|
||||
|
||||
pub async fn create_gridfs_bucket(client: &Client, database: &str, bucket: &str) -> Result<(), String> {
|
||||
let bucket = normalized_gridfs_bucket_name(bucket)?;
|
||||
let database = client.database(database);
|
||||
let files_name = format!("{bucket}.files");
|
||||
let chunks_name = format!("{bucket}.chunks");
|
||||
let existing: HashSet<String> =
|
||||
database.list_collection_names().await.map_err(|e| e.to_string())?.into_iter().collect();
|
||||
|
||||
if !existing.contains(&files_name) {
|
||||
database.create_collection(&files_name).await.map_err(|e| e.to_string())?;
|
||||
}
|
||||
if !existing.contains(&chunks_name) {
|
||||
database.create_collection(&chunks_name).await.map_err(|e| e.to_string())?;
|
||||
}
|
||||
|
||||
database
|
||||
.collection::<Document>(&files_name)
|
||||
.create_index(
|
||||
IndexModel::builder()
|
||||
.keys(doc! { "filename": 1_i32, "uploadDate": 1_i32 })
|
||||
.options(IndexOptions::builder().name(Some("filename_1_uploadDate_1".to_string())).build())
|
||||
.build(),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
database
|
||||
.collection::<Document>(&chunks_name)
|
||||
.create_index(
|
||||
IndexModel::builder()
|
||||
.keys(doc! { "files_id": 1_i32, "n": 1_i32 })
|
||||
.options(IndexOptions::builder().name(Some("files_id_1_n_1".to_string())).unique(Some(true)).build())
|
||||
.build(),
|
||||
)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn delete_gridfs_bucket(client: &Client, database: &str, bucket: &str) -> Result<(), String> {
|
||||
let bucket = normalized_gridfs_bucket_name(bucket)?;
|
||||
let database = client.database(database);
|
||||
drop_collection_if_exists(&database, &format!("{bucket}.files")).await?;
|
||||
drop_collection_if_exists(&database, &format!("{bucket}.chunks")).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn download_gridfs_file(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
bucket: &str,
|
||||
file_id: &str,
|
||||
) -> Result<Vec<u8>, String> {
|
||||
let bucket = normalized_gridfs_bucket_name(bucket)?;
|
||||
let trimmed = file_id.trim();
|
||||
if trimmed.is_empty() {
|
||||
return Err("GridFS file id is required".to_string());
|
||||
}
|
||||
|
||||
let bson_id = parse_gridfs_file_id(trimmed)?;
|
||||
|
||||
let files_collection = client.database(database).collection::<Document>(&format!("{bucket}.files"));
|
||||
let file_doc = files_collection
|
||||
.find_one(doc! { "_id": bson_id.clone() })
|
||||
.await
|
||||
.map_err(|e| e.to_string())?
|
||||
.ok_or_else(|| "GridFS file not found".to_string())?;
|
||||
let bucket =
|
||||
client.database(database).gridfs_bucket(GridFsBucketOptions::builder().bucket_name(bucket.to_string()).build());
|
||||
let mut stream = bucket
|
||||
.open_download_stream(file_doc.get("_id").cloned().unwrap_or(bson_id))
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
let mut bytes = Vec::new();
|
||||
stream.read_to_end(&mut bytes).await.map_err(|e| e.to_string())?;
|
||||
Ok(bytes)
|
||||
}
|
||||
|
||||
pub async fn upload_gridfs_file(
|
||||
client: &Client,
|
||||
database: &str,
|
||||
bucket: &str,
|
||||
file_name: &str,
|
||||
data: &[u8],
|
||||
content_type: Option<&str>,
|
||||
) -> Result<String, String> {
|
||||
let bucket = normalized_gridfs_bucket_name(bucket)?;
|
||||
let file_name = file_name.trim();
|
||||
if file_name.is_empty() {
|
||||
return Err("GridFS file name is required".to_string());
|
||||
}
|
||||
|
||||
create_gridfs_bucket(client, database, &bucket).await?;
|
||||
|
||||
let database_ref = client.database(database);
|
||||
let gridfs_bucket = database_ref.gridfs_bucket(GridFsBucketOptions::builder().bucket_name(bucket.clone()).build());
|
||||
let mut upload_action = gridfs_bucket.open_upload_stream(file_name);
|
||||
if let Some(content_type) = content_type.map(str::trim).filter(|value| !value.is_empty()) {
|
||||
upload_action = upload_action.metadata(doc! { "contentType": content_type });
|
||||
}
|
||||
let mut stream = upload_action.await.map_err(|e| e.to_string())?;
|
||||
let file_id = stream.id().clone();
|
||||
stream.write_all(data).await.map_err(|e| e.to_string())?;
|
||||
stream.close().await.map_err(|e| e.to_string())?;
|
||||
|
||||
if let Some(content_type) = content_type.map(str::trim).filter(|value| !value.is_empty()) {
|
||||
database_ref
|
||||
.collection::<Document>(&format!("{bucket}.files"))
|
||||
.update_one(doc! { "_id": file_id.clone() }, doc! { "$set": { "contentType": content_type } })
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
}
|
||||
|
||||
Ok(gridfs_file_id_to_string(&file_id))
|
||||
}
|
||||
|
||||
pub async fn delete_gridfs_file(client: &Client, database: &str, bucket: &str, file_id: &str) -> Result<(), String> {
|
||||
let bucket = normalized_gridfs_bucket_name(bucket)?;
|
||||
let bson_id = parse_gridfs_file_id(file_id.trim())?;
|
||||
client
|
||||
.database(database)
|
||||
.gridfs_bucket(GridFsBucketOptions::builder().bucket_name(bucket).build())
|
||||
.delete(bson_id)
|
||||
.await
|
||||
.map_err(|e| e.to_string())
|
||||
}
|
||||
|
||||
fn gridfs_file_id_to_string(id: &Bson) -> String {
|
||||
match id {
|
||||
Bson::ObjectId(value) => value.to_hex(),
|
||||
Bson::String(value) => value.clone(),
|
||||
_ => id.clone().into_relaxed_extjson().to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
fn gridfs_upload_date_to_string(value: &DateTime) -> String {
|
||||
value.try_to_rfc3339_string().unwrap_or_else(|_| value.timestamp_millis().to_string())
|
||||
}
|
||||
|
||||
fn gridfs_file_info_from_document(doc: &Document) -> MongoGridFsFileInfo {
|
||||
let id = gridfs_file_id_to_string(doc.get("_id").unwrap_or(&Bson::Null));
|
||||
let filename = doc.get_str("filename").ok().map(str::to_string);
|
||||
let length = doc.get_i64("length").or_else(|_| doc.get_i32("length").map(i64::from)).unwrap_or(0);
|
||||
let chunk_size =
|
||||
doc.get_i32("chunkSize").or_else(|_| doc.get_i64("chunkSize").map(|value| value as i32)).unwrap_or(0);
|
||||
let upload_date = doc.get_datetime("uploadDate").ok().map(gridfs_upload_date_to_string);
|
||||
let metadata = doc.get_document("metadata").ok().map(|value| Bson::Document(value.clone()).into_relaxed_extjson());
|
||||
let md5 = doc.get_str("md5").ok().map(str::to_string);
|
||||
let content_type = doc.get_str("contentType").ok().map(str::to_string).or_else(|| {
|
||||
doc.get_document("metadata").ok().and_then(|meta| meta.get_str("contentType").ok().map(str::to_string))
|
||||
});
|
||||
let aliases = doc.get_array("aliases").ok().and_then(|values| {
|
||||
let aliases: Vec<String> = values.iter().filter_map(|value| value.as_str().map(str::to_string)).collect();
|
||||
if aliases.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(aliases)
|
||||
}
|
||||
});
|
||||
|
||||
MongoGridFsFileInfo { id, filename, length, chunk_size, upload_date, metadata, md5, content_type, aliases }
|
||||
}
|
||||
|
||||
fn parse_gridfs_file_id(file_id: &str) -> Result<Bson, String> {
|
||||
if let Ok(value) = serde_json::from_str::<serde_json::Value>(file_id) {
|
||||
return Bson::try_from(value).map_err(|e| format!("Invalid GridFS file id: {e}"));
|
||||
}
|
||||
|
||||
if let Ok(object_id) = ObjectId::parse_str(file_id) {
|
||||
return Ok(Bson::ObjectId(object_id));
|
||||
}
|
||||
|
||||
Ok(Bson::String(file_id.to_string()))
|
||||
}
|
||||
|
||||
fn normalized_gridfs_bucket_name(bucket: &str) -> Result<String, String> {
|
||||
let bucket = bucket.trim();
|
||||
if bucket.is_empty() {
|
||||
return Err("GridFS bucket name is required".to_string());
|
||||
}
|
||||
if bucket.ends_with(".files") || bucket.ends_with(".chunks") {
|
||||
return Err("Use the GridFS bucket name without the .files or .chunks suffix".to_string());
|
||||
}
|
||||
Ok(bucket.to_string())
|
||||
}
|
||||
|
||||
async fn drop_collection_if_exists(database: &Database, collection_name: &str) -> Result<(), String> {
|
||||
match database.collection::<Document>(collection_name).drop().await {
|
||||
Ok(()) => Ok(()),
|
||||
Err(error) if mongo_namespace_missing(&error.to_string()) => Ok(()),
|
||||
Err(error) => Err(error.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
fn mongo_namespace_missing(error: &str) -> bool {
|
||||
let lower = error.to_lowercase();
|
||||
lower.contains("ns not found") || lower.contains("namespace not found")
|
||||
}
|
||||
|
||||
pub async fn create_database(client: &Client, database: &str) -> Result<(), String> {
|
||||
let database = database.trim();
|
||||
if database.is_empty() {
|
||||
@@ -1236,6 +1484,46 @@ mod tests {
|
||||
assert!(matches!(doc.get("count"), Some(Bson::Int64(42))));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_gridfs_file_id_accepts_extended_json_object_id() {
|
||||
let id = parse_gridfs_file_id(r#"{"$oid":"507f1f77bcf86cd799439011"}"#).unwrap();
|
||||
|
||||
assert!(matches!(id, Bson::ObjectId(oid) if oid.to_hex() == "507f1f77bcf86cd799439011"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_gridfs_file_id_accepts_json_string_ids() {
|
||||
let id = parse_gridfs_file_id(r#""report-42""#).unwrap();
|
||||
|
||||
assert!(matches!(id, Bson::String(value) if value == "report-42"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gridfs_file_id_to_string_keeps_plain_strings_unquoted() {
|
||||
let id = gridfs_file_id_to_string(&Bson::String("report-42".to_string()));
|
||||
|
||||
assert_eq!(id, "report-42");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn gridfs_file_info_includes_navicat_style_metadata_fields() {
|
||||
let info = gridfs_file_info_from_document(&doc! {
|
||||
"_id": "report-42",
|
||||
"filename": "report.zip",
|
||||
"length": 128_i64,
|
||||
"chunkSize": 255_i32,
|
||||
"md5": "abc123",
|
||||
"contentType": "application/zip",
|
||||
"aliases": ["archive", "nightly"],
|
||||
});
|
||||
|
||||
assert_eq!(info.id, "report-42");
|
||||
assert_eq!(info.filename.as_deref(), Some("report.zip"));
|
||||
assert_eq!(info.md5.as_deref(), Some("abc123"));
|
||||
assert_eq!(info.content_type.as_deref(), Some("application/zip"));
|
||||
assert_eq!(info.aliases, Some(vec!["archive".to_string(), "nightly".to_string()]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn json_object_to_document_parses_find_projection() {
|
||||
let value = serde_json::json!({
|
||||
|
||||
@@ -28,10 +28,13 @@ const PATH_SEGMENT_ENCODE_SET: &AsciiSet = &CONTROLS
|
||||
const QUERY_VALUE_ENCODE_SET: &AsciiSet = &PATH_SEGMENT_ENCODE_SET.add(b'&').add(b'=').add(b'+');
|
||||
|
||||
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct CollectionInfo {
|
||||
pub name: String,
|
||||
pub id: String,
|
||||
pub dimension: Option<u32>,
|
||||
pub kind: Option<String>,
|
||||
pub bucket_name: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
@@ -177,7 +180,13 @@ async fn list_qdrant_collections(client: &VectorClient) -> Result<Vec<Collection
|
||||
.flatten()
|
||||
.filter_map(|item| {
|
||||
let name = item.get("name").and_then(Value::as_str)?;
|
||||
Some(CollectionInfo { name: name.to_string(), id: name.to_string(), dimension: None })
|
||||
Some(CollectionInfo {
|
||||
name: name.to_string(),
|
||||
id: name.to_string(),
|
||||
dimension: None,
|
||||
kind: None,
|
||||
bucket_name: None,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
infos.sort_by(|a, b| a.name.cmp(&b.name));
|
||||
@@ -196,7 +205,7 @@ async fn list_milvus_collections(client: &VectorClient, database: &str) -> Resul
|
||||
.iter()
|
||||
.filter_map(|item| {
|
||||
let name = collection_name_from_milvus_item(item)?;
|
||||
Some(CollectionInfo { name: name.clone(), id: name, dimension: None })
|
||||
Some(CollectionInfo { name: name.clone(), id: name, dimension: None, kind: None, bucket_name: None })
|
||||
})
|
||||
.collect(),
|
||||
_ => Vec::new(),
|
||||
@@ -216,7 +225,7 @@ async fn list_weaviate_collections(client: &VectorClient) -> Result<Vec<Collecti
|
||||
let body = send_json(client.get("/v1/schema"), "Weaviate").await?;
|
||||
let mut infos: Vec<CollectionInfo> = weaviate_collection_names_from_schema(&body)
|
||||
.into_iter()
|
||||
.map(|name| CollectionInfo { name: name.clone(), id: name, dimension: None })
|
||||
.map(|name| CollectionInfo { name: name.clone(), id: name, dimension: None, kind: None, bucket_name: None })
|
||||
.collect();
|
||||
infos.sort_by(|a, b| a.name.cmp(&b.name));
|
||||
Ok(infos)
|
||||
@@ -234,7 +243,13 @@ async fn list_chroma_collections(client: &VectorClient) -> Result<Vec<Collection
|
||||
let name = item.get("name").and_then(Value::as_str)?;
|
||||
let id = item.get("id").and_then(Value::as_str)?;
|
||||
let dimension = item.get("dimension").and_then(|v| v.as_u64()).map(|d| d as u32);
|
||||
Some(CollectionInfo { name: name.to_string(), id: id.to_string(), dimension })
|
||||
Some(CollectionInfo {
|
||||
name: name.to_string(),
|
||||
id: id.to_string(),
|
||||
dimension,
|
||||
kind: None,
|
||||
bucket_name: None,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
infos.sort_by(|a, b| a.name.cmp(&b.name));
|
||||
@@ -251,7 +266,13 @@ pub async fn get_collection_detail(
|
||||
VectorDbKind::Milvus => get_milvus_collection_detail(client, database, collection).await,
|
||||
VectorDbKind::Weaviate => {
|
||||
// Weaviate REST API does not expose vector dimension
|
||||
Ok(CollectionInfo { name: collection.to_string(), id: collection.to_string(), dimension: None })
|
||||
Ok(CollectionInfo {
|
||||
name: collection.to_string(),
|
||||
id: collection.to_string(),
|
||||
dimension: None,
|
||||
kind: None,
|
||||
bucket_name: None,
|
||||
})
|
||||
}
|
||||
VectorDbKind::ChromaDb => get_chroma_collection_detail(client, collection).await,
|
||||
}
|
||||
@@ -268,7 +289,13 @@ async fn get_qdrant_collection_detail(client: &VectorClient, collection: &str) -
|
||||
.and_then(|obj| obj.values().find_map(|v| v.get("size").and_then(|s| s.as_u64())))
|
||||
})
|
||||
.map(|d| d as u32);
|
||||
Ok(CollectionInfo { name: collection.to_string(), id: collection.to_string(), dimension: dim })
|
||||
Ok(CollectionInfo {
|
||||
name: collection.to_string(),
|
||||
id: collection.to_string(),
|
||||
dimension: dim,
|
||||
kind: None,
|
||||
bucket_name: None,
|
||||
})
|
||||
}
|
||||
|
||||
fn milvus_vector_dim_from_field(field: &Value) -> Option<u32> {
|
||||
@@ -319,7 +346,13 @@ async fn get_milvus_collection_detail(
|
||||
})
|
||||
})
|
||||
.and_then(milvus_vector_dim_from_field);
|
||||
Ok(CollectionInfo { name: collection.to_string(), id: collection.to_string(), dimension: dim })
|
||||
Ok(CollectionInfo {
|
||||
name: collection.to_string(),
|
||||
id: collection.to_string(),
|
||||
dimension: dim,
|
||||
kind: None,
|
||||
bucket_name: None,
|
||||
})
|
||||
}
|
||||
|
||||
async fn get_chroma_collection_detail(client: &VectorClient, collection: &str) -> Result<CollectionInfo, String> {
|
||||
@@ -334,7 +367,7 @@ async fn get_chroma_collection_detail(client: &VectorClient, collection: &str) -
|
||||
let name = body.get("name").and_then(Value::as_str).unwrap_or(collection);
|
||||
let id = body.get("id").and_then(Value::as_str).unwrap_or(collection);
|
||||
let dimension = body.get("dimension").and_then(|v| v.as_u64()).map(|d| d as u32);
|
||||
Ok(CollectionInfo { name: name.to_string(), id: id.to_string(), dimension })
|
||||
Ok(CollectionInfo { name: name.to_string(), id: id.to_string(), dimension, kind: None, bucket_name: None })
|
||||
}
|
||||
|
||||
fn chroma_get_response_to_rows(body: &Value) -> Vec<Value> {
|
||||
@@ -773,7 +806,13 @@ mod tests {
|
||||
.filter_map(|item| {
|
||||
let name = item.get("name").and_then(|v| v.as_str())?;
|
||||
let id = item.get("id").and_then(|v| v.as_str())?;
|
||||
Some(CollectionInfo { name: name.to_string(), id: id.to_string(), dimension: None })
|
||||
Some(CollectionInfo {
|
||||
name: name.to_string(),
|
||||
id: id.to_string(),
|
||||
dimension: None,
|
||||
kind: None,
|
||||
bucket_name: None,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
assert_eq!(infos.len(), 2);
|
||||
|
||||
@@ -5,6 +5,28 @@ use crate::db::{elasticsearch_driver, mongo_driver, vector_driver};
|
||||
|
||||
pub use crate::db::vector_driver::CollectionInfo;
|
||||
|
||||
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MongoGridFsFileInfo {
|
||||
pub id: String,
|
||||
pub filename: Option<String>,
|
||||
pub length: i64,
|
||||
pub chunk_size: i32,
|
||||
pub upload_date: Option<String>,
|
||||
pub metadata: Option<serde_json::Value>,
|
||||
pub md5: Option<String>,
|
||||
pub content_type: Option<String>,
|
||||
pub aliases: Option<Vec<String>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MongoGridFsBucketInfo {
|
||||
pub name: String,
|
||||
pub file_count: u64,
|
||||
pub total_bytes: i64,
|
||||
}
|
||||
|
||||
fn sort_names(mut names: Vec<String>) -> Vec<String> {
|
||||
names.sort_by(|left, right| {
|
||||
let left_lower = left.to_lowercase();
|
||||
@@ -61,6 +83,45 @@ fn mongo_list_databases_unauthorized(error: &str) -> bool {
|
||||
lower.contains("not authorized") && lower.contains("listdatabases")
|
||||
}
|
||||
|
||||
fn mongo_collection_info(name: String) -> CollectionInfo {
|
||||
CollectionInfo {
|
||||
name: name.clone(),
|
||||
id: name,
|
||||
dimension: None,
|
||||
kind: Some("collection".to_string()),
|
||||
bucket_name: None,
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn mongo_gridfs_bucket_names(names: &[String]) -> Vec<String> {
|
||||
use std::collections::BTreeSet;
|
||||
|
||||
let name_set: BTreeSet<&str> = names.iter().map(String::as_str).collect();
|
||||
let bucket_names: Vec<String> = names
|
||||
.iter()
|
||||
.filter_map(|name| name.strip_suffix(".files"))
|
||||
.filter(|prefix| name_set.contains(format!("{prefix}.chunks").as_str()))
|
||||
.map(str::to_string)
|
||||
.collect::<BTreeSet<_>>()
|
||||
.into_iter()
|
||||
.collect();
|
||||
|
||||
sort_names(bucket_names)
|
||||
}
|
||||
|
||||
fn mongo_bucket_infos(names: &[String]) -> Vec<CollectionInfo> {
|
||||
mongo_gridfs_bucket_names(names)
|
||||
.into_iter()
|
||||
.map(|bucket_name| CollectionInfo {
|
||||
name: bucket_name.clone(),
|
||||
id: format!("bucket:{bucket_name}"),
|
||||
dimension: None,
|
||||
kind: Some("bucket".to_string()),
|
||||
bucket_name: Some(bucket_name),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub async fn list_collections_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
@@ -71,22 +132,148 @@ pub async fn list_collections_core(
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::MongoDb(client) => {
|
||||
let names = sort_names(mongo_driver::list_collections(client, database).await?);
|
||||
Ok(names.into_iter().map(|n| CollectionInfo { name: n.clone(), id: n, dimension: None }).collect())
|
||||
let mut infos = mongo_bucket_infos(&names);
|
||||
infos.extend(names.into_iter().map(mongo_collection_info));
|
||||
Ok(infos)
|
||||
}
|
||||
PoolKind::Elasticsearch(client) => {
|
||||
let names = sort_names(elasticsearch_driver::list_indices(client).await?);
|
||||
Ok(names.into_iter().map(|n| CollectionInfo { name: n.clone(), id: n, dimension: None }).collect())
|
||||
Ok(names
|
||||
.into_iter()
|
||||
.map(|n| CollectionInfo { name: n.clone(), id: n, dimension: None, kind: None, bucket_name: None })
|
||||
.collect())
|
||||
}
|
||||
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?);
|
||||
Ok(names.into_iter().map(|n| CollectionInfo { name: n.clone(), id: n, dimension: None }).collect())
|
||||
let mut infos = mongo_bucket_infos(&names);
|
||||
infos.extend(names.into_iter().map(mongo_collection_info));
|
||||
Ok(infos)
|
||||
}
|
||||
_ => Err("Not a MongoDB/Elasticsearch/vector connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn list_gridfs_files_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
bucket: &str,
|
||||
) -> Result<Vec<MongoGridFsFileInfo>, String> {
|
||||
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::list_gridfs_files(client, database, bucket).await,
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support GridFS file browsing".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn list_gridfs_buckets_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
) -> Result<Vec<MongoGridFsBucketInfo>, String> {
|
||||
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) => {
|
||||
let names = sort_names(mongo_driver::list_collections(client, database).await?);
|
||||
let bucket_names = mongo_gridfs_bucket_names(&names);
|
||||
let mut buckets = Vec::with_capacity(bucket_names.len());
|
||||
for bucket_name in bucket_names {
|
||||
buckets.push(mongo_driver::gridfs_bucket_summary(client, database, &bucket_name).await?);
|
||||
}
|
||||
Ok(buckets)
|
||||
}
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support GridFS bucket browsing".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn create_gridfs_bucket_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
bucket: &str,
|
||||
) -> Result<(), String> {
|
||||
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_gridfs_bucket(client, database, bucket).await,
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support GridFS bucket creation".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn delete_gridfs_bucket_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
bucket: &str,
|
||||
) -> Result<(), String> {
|
||||
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::delete_gridfs_bucket(client, database, bucket).await,
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support GridFS bucket deletion".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn download_gridfs_file_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
bucket: &str,
|
||||
file_id: &str,
|
||||
) -> Result<Vec<u8>, String> {
|
||||
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::download_gridfs_file(client, database, bucket, file_id).await,
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support GridFS download".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn upload_gridfs_file_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
bucket: &str,
|
||||
file_name: &str,
|
||||
data: &[u8],
|
||||
content_type: Option<&str>,
|
||||
) -> Result<String, String> {
|
||||
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::upload_gridfs_file(client, database, bucket, file_name, data, content_type).await
|
||||
}
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support GridFS uploads".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn delete_gridfs_file_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
bucket: &str,
|
||||
file_id: &str,
|
||||
) -> Result<(), String> {
|
||||
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::delete_gridfs_file(client, database, bucket, file_id).await,
|
||||
PoolKind::Agent(_) => Err("MongoDB legacy agent does not support GridFS file deletion".to_string()),
|
||||
_ => Err("Not a MongoDB connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn find_documents_core(
|
||||
state: &AppState,
|
||||
@@ -233,7 +420,7 @@ pub async fn delete_document_core(
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{fallback_mongo_database, mongo_list_databases_unauthorized, sort_names};
|
||||
use super::{fallback_mongo_database, mongo_gridfs_bucket_names, mongo_list_databases_unauthorized, sort_names};
|
||||
|
||||
#[test]
|
||||
fn sorts_names_case_insensitively() {
|
||||
@@ -263,4 +450,18 @@ mod tests {
|
||||
);
|
||||
assert_eq!(fallback_mongo_database("not authorized", None).unwrap_err(), "not authorized");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn extracts_gridfs_bucket_names_from_matching_files_and_chunks_collections() {
|
||||
let buckets = mongo_gridfs_bucket_names(&[
|
||||
"orders.files".to_string(),
|
||||
"orders.chunks".to_string(),
|
||||
"reports.files".to_string(),
|
||||
"reports.chunks".to_string(),
|
||||
"reports.files".to_string(),
|
||||
"loose.files".to_string(),
|
||||
]);
|
||||
|
||||
assert_eq!(buckets, vec!["orders".to_string(), "reports".to_string()]);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user