feat(dev): also break up the large webdav_service.rs file into smaller ones
This commit is contained in:
parent
ed942d02c7
commit
b9e0e5b905
|
|
@ -1,5 +1,5 @@
|
|||
use anyhow::Result;
|
||||
use sqlx::{QueryBuilder, Postgres};
|
||||
use sqlx::{QueryBuilder, Postgres, Row};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::models::{Document, UserRole, FacetItem};
|
||||
|
|
@ -53,11 +53,16 @@ impl Database {
|
|||
let doc_id: Uuid = row.get("document_id");
|
||||
let label = Label {
|
||||
id: row.get("label_id"),
|
||||
user_id: row.get("user_id"),
|
||||
user_id: Some(row.get("user_id")),
|
||||
name: row.get("name"),
|
||||
description: None,
|
||||
color: row.get("color"),
|
||||
background_color: None,
|
||||
icon: None,
|
||||
is_system: false,
|
||||
created_at: row.get("created_at"),
|
||||
updated_at: row.get("updated_at"),
|
||||
document_count: 0,
|
||||
};
|
||||
|
||||
if Some(doc_id) != current_doc_id {
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
use anyhow::Result;
|
||||
use sqlx::{QueryBuilder, Postgres};
|
||||
use sqlx::{QueryBuilder, Postgres, Row};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::models::{Document, UserRole, SearchRequest, SearchMode, SearchSnippet, HighlightRange, EnhancedDocumentResponse};
|
||||
|
|
|
|||
|
|
@ -443,7 +443,7 @@ async fn trigger_deep_scan(
|
|||
})?;
|
||||
|
||||
// Create WebDAV service
|
||||
let webdav_config = crate::services::webdav_service::WebDAVConfig {
|
||||
let webdav_config = crate::services::webdav::WebDAVConfig {
|
||||
server_url: config.server_url.clone(),
|
||||
username: config.username.clone(),
|
||||
password: config.password.clone(),
|
||||
|
|
@ -453,7 +453,7 @@ async fn trigger_deep_scan(
|
|||
server_type: config.server_type.clone(),
|
||||
};
|
||||
|
||||
let webdav_service = crate::services::webdav_service::WebDAVService::new(webdav_config.clone())
|
||||
let webdav_service = crate::services::webdav::WebDAVService::new(webdav_config.clone())
|
||||
.map_err(|e| {
|
||||
error!("Failed to create WebDAV service for deep scan: {}", e);
|
||||
StatusCode::INTERNAL_SERVER_ERROR
|
||||
|
|
@ -795,7 +795,7 @@ async fn test_connection(
|
|||
let config: crate::models::WebDAVSourceConfig = serde_json::from_value(source.config)
|
||||
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
|
||||
|
||||
match crate::services::webdav_service::test_webdav_connection(
|
||||
match crate::services::webdav::test_webdav_connection(
|
||||
&config.server_url,
|
||||
&config.username,
|
||||
&config.password,
|
||||
|
|
@ -963,7 +963,7 @@ async fn estimate_webdav_crawl_internal(
|
|||
config: &crate::models::WebDAVSourceConfig,
|
||||
) -> Result<Json<serde_json::Value>, StatusCode> {
|
||||
// Create WebDAV service config
|
||||
let webdav_config = crate::services::webdav_service::WebDAVConfig {
|
||||
let webdav_config = crate::services::webdav::WebDAVConfig {
|
||||
server_url: config.server_url.clone(),
|
||||
username: config.username.clone(),
|
||||
password: config.password.clone(),
|
||||
|
|
@ -974,7 +974,7 @@ async fn estimate_webdav_crawl_internal(
|
|||
};
|
||||
|
||||
// Create WebDAV service and estimate crawl
|
||||
match crate::services::webdav_service::WebDAVService::new(webdav_config) {
|
||||
match crate::services::webdav::WebDAVService::new(webdav_config) {
|
||||
Ok(webdav_service) => {
|
||||
match webdav_service.estimate_crawl(&config.watch_folders).await {
|
||||
Ok(estimate) => Ok(Json(serde_json::to_value(estimate).unwrap())),
|
||||
|
|
@ -1031,7 +1031,7 @@ async fn test_connection_with_config(
|
|||
let config: crate::models::WebDAVSourceConfig = serde_json::from_value(request.config)
|
||||
.map_err(|_| StatusCode::BAD_REQUEST)?;
|
||||
|
||||
match crate::services::webdav_service::test_webdav_connection(
|
||||
match crate::services::webdav::test_webdav_connection(
|
||||
&config.server_url,
|
||||
&config.username,
|
||||
&config.password,
|
||||
|
|
|
|||
|
|
@ -17,8 +17,8 @@ use crate::{
|
|||
},
|
||||
AppState,
|
||||
};
|
||||
use crate::services::webdav_service::WebDAVConfig;
|
||||
use crate::services::webdav_service::WebDAVService;
|
||||
use crate::services::webdav::WebDAVConfig;
|
||||
use crate::services::webdav::WebDAVService;
|
||||
|
||||
pub mod webdav_sync;
|
||||
use webdav_sync::perform_webdav_sync_with_tracking;
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ use crate::{
|
|||
models::{CreateWebDAVFile, UpdateWebDAVSyncState},
|
||||
services::file_service::FileService,
|
||||
ingestion::document_ingestion::{DocumentIngestionService, IngestionResult},
|
||||
services::webdav_service::{WebDAVConfig, WebDAVService},
|
||||
services::webdav::{WebDAVConfig, WebDAVService},
|
||||
};
|
||||
|
||||
pub async fn perform_webdav_sync_with_tracking(
|
||||
|
|
|
|||
|
|
@ -662,8 +662,8 @@ impl SourceScheduler {
|
|||
// Trigger the deep scan via the API endpoint
|
||||
// We'll reuse the existing deep scan logic from the sources route
|
||||
let webdav_config: WebDAVSourceConfig = serde_json::from_value(source.config.clone())?;
|
||||
let webdav_service = crate::services::webdav_service::WebDAVService::new(
|
||||
crate::services::webdav_service::WebDAVConfig {
|
||||
let webdav_service = crate::services::webdav::WebDAVService::new(
|
||||
crate::services::webdav::WebDAVConfig {
|
||||
server_url: webdav_config.server_url.clone(),
|
||||
username: webdav_config.username.clone(),
|
||||
password: webdav_config.password.clone(),
|
||||
|
|
@ -950,7 +950,7 @@ impl SourceScheduler {
|
|||
let config: WebDAVSourceConfig = serde_json::from_value(source.config.clone())
|
||||
.map_err(|e| format!("Config parse error: {}", e))?;
|
||||
|
||||
let webdav_config = crate::services::webdav_service::WebDAVConfig {
|
||||
let webdav_config = crate::services::webdav::WebDAVConfig {
|
||||
server_url: config.server_url.clone(),
|
||||
username: config.username.clone(),
|
||||
password: config.password.clone(),
|
||||
|
|
@ -960,7 +960,7 @@ impl SourceScheduler {
|
|||
server_type: config.server_type.clone(),
|
||||
};
|
||||
|
||||
let webdav_service = crate::services::webdav_service::WebDAVService::new(webdav_config)
|
||||
let webdav_service = crate::services::webdav::WebDAVService::new(webdav_config)
|
||||
.map_err(|e| format!("Service creation failed: {}", e))?;
|
||||
|
||||
let test_config = crate::models::WebDAVTestConnection {
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ use crate::{
|
|||
ingestion::document_ingestion::{DocumentIngestionService, IngestionResult},
|
||||
services::local_folder_service::LocalFolderService,
|
||||
services::s3_service::S3Service,
|
||||
services::webdav_service::{WebDAVService, WebDAVConfig},
|
||||
services::webdav::{WebDAVService, WebDAVConfig},
|
||||
};
|
||||
|
||||
#[derive(Clone)]
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@ use crate::{
|
|||
services::file_service::FileService,
|
||||
AppState,
|
||||
};
|
||||
use crate::services::webdav_service::{WebDAVConfig, WebDAVService};
|
||||
use crate::services::webdav::{WebDAVConfig, WebDAVService};
|
||||
use crate::routes::webdav::webdav_sync::perform_webdav_sync_with_tracking;
|
||||
|
||||
pub struct WebDAVScheduler {
|
||||
|
|
|
|||
|
|
@ -3,4 +3,4 @@ pub mod local_folder_service;
|
|||
pub mod ocr_retry_service;
|
||||
pub mod s3_service;
|
||||
pub mod s3_service_stub;
|
||||
pub mod webdav_service;
|
||||
pub mod webdav;
|
||||
|
|
@ -0,0 +1,141 @@
|
|||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// WebDAV server configuration
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct WebDAVConfig {
|
||||
pub server_url: String,
|
||||
pub username: String,
|
||||
pub password: String,
|
||||
pub watch_folders: Vec<String>,
|
||||
pub file_extensions: Vec<String>,
|
||||
pub timeout_seconds: u64,
|
||||
pub server_type: Option<String>, // "nextcloud", "owncloud", "generic"
|
||||
}
|
||||
|
||||
/// Retry configuration for WebDAV operations
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct RetryConfig {
|
||||
pub max_retries: u32,
|
||||
pub initial_delay_ms: u64,
|
||||
pub max_delay_ms: u64,
|
||||
pub backoff_multiplier: f64,
|
||||
pub timeout_seconds: u64,
|
||||
pub rate_limit_backoff_ms: u64, // Additional backoff for 429 responses
|
||||
}
|
||||
|
||||
/// Concurrency configuration for WebDAV operations
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ConcurrencyConfig {
|
||||
pub max_concurrent_scans: usize,
|
||||
pub max_concurrent_downloads: usize,
|
||||
pub adaptive_rate_limiting: bool,
|
||||
}
|
||||
|
||||
impl Default for RetryConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_retries: 3,
|
||||
initial_delay_ms: 1000, // 1 second
|
||||
max_delay_ms: 30000, // 30 seconds
|
||||
backoff_multiplier: 2.0,
|
||||
timeout_seconds: 30,
|
||||
rate_limit_backoff_ms: 5000, // 5 seconds
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for ConcurrencyConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_concurrent_scans: 4,
|
||||
max_concurrent_downloads: 8,
|
||||
adaptive_rate_limiting: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl WebDAVConfig {
|
||||
/// Creates a new WebDAV configuration
|
||||
pub fn new(
|
||||
server_url: String,
|
||||
username: String,
|
||||
password: String,
|
||||
watch_folders: Vec<String>,
|
||||
file_extensions: Vec<String>,
|
||||
) -> Self {
|
||||
Self {
|
||||
server_url,
|
||||
username,
|
||||
password,
|
||||
watch_folders,
|
||||
file_extensions,
|
||||
timeout_seconds: 30,
|
||||
server_type: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Validates the configuration
|
||||
pub fn validate(&self) -> anyhow::Result<()> {
|
||||
if self.server_url.is_empty() {
|
||||
return Err(anyhow::anyhow!("Server URL cannot be empty"));
|
||||
}
|
||||
|
||||
if self.username.is_empty() {
|
||||
return Err(anyhow::anyhow!("Username cannot be empty"));
|
||||
}
|
||||
|
||||
if self.password.is_empty() {
|
||||
return Err(anyhow::anyhow!("Password cannot be empty"));
|
||||
}
|
||||
|
||||
if self.watch_folders.is_empty() {
|
||||
return Err(anyhow::anyhow!("At least one watch folder must be specified"));
|
||||
}
|
||||
|
||||
// Validate URL format
|
||||
if !self.server_url.starts_with("http://") && !self.server_url.starts_with("https://") {
|
||||
return Err(anyhow::anyhow!("Server URL must start with http:// or https://"));
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Returns the base URL for WebDAV operations
|
||||
pub fn webdav_url(&self) -> String {
|
||||
let mut url = self.server_url.trim_end_matches('/').to_string();
|
||||
|
||||
// Add WebDAV path based on server type
|
||||
match self.server_type.as_deref() {
|
||||
Some("nextcloud") => {
|
||||
if !url.contains("/remote.php/dav/files/") {
|
||||
url.push_str(&format!("/remote.php/dav/files/{}", self.username));
|
||||
}
|
||||
}
|
||||
Some("owncloud") => {
|
||||
if !url.contains("/remote.php/webdav") {
|
||||
url.push_str("/remote.php/webdav");
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
// Generic WebDAV - use the URL as provided
|
||||
}
|
||||
}
|
||||
|
||||
url
|
||||
}
|
||||
|
||||
/// Checks if a file extension is supported
|
||||
pub fn is_supported_extension(&self, filename: &str) -> bool {
|
||||
if self.file_extensions.is_empty() {
|
||||
return true; // If no extensions specified, support all
|
||||
}
|
||||
|
||||
let extension = filename.split('.').last().unwrap_or("");
|
||||
self.file_extensions.iter().any(|ext| ext.eq_ignore_ascii_case(extension))
|
||||
}
|
||||
|
||||
/// Gets the timeout duration
|
||||
pub fn timeout(&self) -> std::time::Duration {
|
||||
std::time::Duration::from_secs(self.timeout_seconds)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,307 @@
|
|||
use anyhow::{anyhow, Result};
|
||||
use reqwest::{Client, Method};
|
||||
use std::time::Duration;
|
||||
use tokio::time::sleep;
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
use crate::models::{WebDAVConnectionResult, WebDAVTestConnection};
|
||||
use super::config::{WebDAVConfig, RetryConfig};
|
||||
|
||||
pub struct WebDAVConnection {
|
||||
client: Client,
|
||||
config: WebDAVConfig,
|
||||
retry_config: RetryConfig,
|
||||
}
|
||||
|
||||
impl WebDAVConnection {
|
||||
pub fn new(config: WebDAVConfig, retry_config: RetryConfig) -> Result<Self> {
|
||||
// Validate configuration first
|
||||
config.validate()?;
|
||||
let client = Client::builder()
|
||||
.timeout(config.timeout())
|
||||
.build()?;
|
||||
|
||||
Ok(Self {
|
||||
client,
|
||||
config,
|
||||
retry_config,
|
||||
})
|
||||
}
|
||||
|
||||
/// Tests WebDAV connection with the provided configuration
|
||||
pub async fn test_connection(&self) -> Result<WebDAVConnectionResult> {
|
||||
info!("🔍 Testing WebDAV connection to: {}", self.config.server_url);
|
||||
|
||||
// Validate configuration first
|
||||
if let Err(e) = self.config.validate() {
|
||||
return Ok(WebDAVConnectionResult {
|
||||
success: false,
|
||||
message: format!("Configuration error: {}", e),
|
||||
server_version: None,
|
||||
server_type: None,
|
||||
});
|
||||
}
|
||||
|
||||
// Test basic connectivity with OPTIONS request
|
||||
match self.test_options_request().await {
|
||||
Ok((server_version, server_type)) => {
|
||||
info!("✅ WebDAV connection successful");
|
||||
Ok(WebDAVConnectionResult {
|
||||
success: true,
|
||||
message: "Connection successful".to_string(),
|
||||
server_version,
|
||||
server_type,
|
||||
})
|
||||
}
|
||||
Err(e) => {
|
||||
error!("❌ WebDAV connection failed: {}", e);
|
||||
Ok(WebDAVConnectionResult {
|
||||
success: false,
|
||||
message: format!("Connection failed: {}", e),
|
||||
server_version: None,
|
||||
server_type: None,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Tests connection with provided credentials (for configuration testing)
|
||||
pub async fn test_connection_with_config(test_config: &WebDAVTestConnection) -> Result<WebDAVConnectionResult> {
|
||||
let config = WebDAVConfig {
|
||||
server_url: test_config.server_url.clone(),
|
||||
username: test_config.username.clone(),
|
||||
password: test_config.password.clone(),
|
||||
watch_folders: vec!["/".to_string()],
|
||||
file_extensions: vec![],
|
||||
timeout_seconds: 30,
|
||||
server_type: test_config.server_type.clone(),
|
||||
};
|
||||
|
||||
let connection = Self::new(config, RetryConfig::default())?;
|
||||
connection.test_connection().await
|
||||
}
|
||||
|
||||
/// Performs OPTIONS request to test basic connectivity
|
||||
async fn test_options_request(&self) -> Result<(Option<String>, Option<String>)> {
|
||||
let webdav_url = self.config.webdav_url();
|
||||
|
||||
let response = self.client
|
||||
.request(Method::OPTIONS, &webdav_url)
|
||||
.basic_auth(&self.config.username, Some(&self.config.password))
|
||||
.send()
|
||||
.await?;
|
||||
|
||||
if !response.status().is_success() {
|
||||
return Err(anyhow!(
|
||||
"OPTIONS request failed with status: {} - {}",
|
||||
response.status(),
|
||||
response.text().await.unwrap_or_default()
|
||||
));
|
||||
}
|
||||
|
||||
// Extract server information from headers
|
||||
let server_version = response
|
||||
.headers()
|
||||
.get("server")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map(|s| s.to_string());
|
||||
|
||||
let server_type = self.detect_server_type(&response, &server_version).await;
|
||||
|
||||
Ok((server_version, server_type))
|
||||
}
|
||||
|
||||
/// Detects the WebDAV server type based on response headers and capabilities
|
||||
async fn detect_server_type(
|
||||
&self,
|
||||
response: &reqwest::Response,
|
||||
server_version: &Option<String>,
|
||||
) -> Option<String> {
|
||||
// Check server header first
|
||||
if let Some(ref server) = server_version {
|
||||
let server_lower = server.to_lowercase();
|
||||
if server_lower.contains("nextcloud") {
|
||||
return Some("nextcloud".to_string());
|
||||
}
|
||||
if server_lower.contains("owncloud") {
|
||||
return Some("owncloud".to_string());
|
||||
}
|
||||
if server_lower.contains("apache") || server_lower.contains("nginx") {
|
||||
// Could be generic WebDAV
|
||||
}
|
||||
}
|
||||
|
||||
// Check DAV capabilities
|
||||
if let Some(dav_header) = response.headers().get("dav") {
|
||||
if let Ok(dav_str) = dav_header.to_str() {
|
||||
debug!("DAV capabilities: {}", dav_str);
|
||||
// Different servers expose different DAV levels
|
||||
if dav_str.contains("3") {
|
||||
return Some("webdav_level_3".to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Test for Nextcloud/ownCloud specific endpoints
|
||||
if self.test_nextcloud_capabilities().await.is_ok() {
|
||||
return Some("nextcloud".to_string());
|
||||
}
|
||||
|
||||
Some("generic".to_string())
|
||||
}
|
||||
|
||||
/// Tests for Nextcloud-specific capabilities
|
||||
async fn test_nextcloud_capabilities(&self) -> Result<()> {
|
||||
let capabilities_url = format!("{}/ocs/v1.php/cloud/capabilities",
|
||||
self.config.server_url.trim_end_matches('/'));
|
||||
|
||||
let response = self.client
|
||||
.get(&capabilities_url)
|
||||
.basic_auth(&self.config.username, Some(&self.config.password))
|
||||
.header("OCS-APIRequest", "true")
|
||||
.send()
|
||||
.await?;
|
||||
|
||||
if response.status().is_success() {
|
||||
debug!("Nextcloud capabilities endpoint accessible");
|
||||
Ok(())
|
||||
} else {
|
||||
Err(anyhow!("Nextcloud capabilities not accessible"))
|
||||
}
|
||||
}
|
||||
|
||||
/// Tests PROPFIND request on root directory
|
||||
pub async fn test_propfind(&self, path: &str) -> Result<()> {
|
||||
let url = format!("{}{}", self.config.webdav_url(), path);
|
||||
|
||||
let propfind_body = r#"<?xml version="1.0" encoding="utf-8"?>
|
||||
<D:propfind xmlns:D="DAV:">
|
||||
<D:prop>
|
||||
<D:displayname/>
|
||||
<D:getcontentlength/>
|
||||
<D:getlastmodified/>
|
||||
<D:getetag/>
|
||||
<D:resourcetype/>
|
||||
</D:prop>
|
||||
</D:propfind>"#;
|
||||
|
||||
let response = self.client
|
||||
.request(Method::from_bytes(b"PROPFIND")?)
|
||||
.url(&url)
|
||||
.basic_auth(&self.config.username, Some(&self.config.password))
|
||||
.header("Depth", "1")
|
||||
.header("Content-Type", "application/xml")
|
||||
.body(propfind_body)
|
||||
.send()
|
||||
.await?;
|
||||
|
||||
if response.status().as_u16() == 207 {
|
||||
debug!("PROPFIND successful for path: {}", path);
|
||||
Ok(())
|
||||
} else {
|
||||
Err(anyhow!(
|
||||
"PROPFIND failed for path '{}' with status: {} - {}",
|
||||
path,
|
||||
response.status(),
|
||||
response.text().await.unwrap_or_default()
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
/// Performs authenticated request with retry logic
|
||||
pub async fn authenticated_request(
|
||||
&self,
|
||||
method: Method,
|
||||
url: &str,
|
||||
body: Option<String>,
|
||||
headers: Option<Vec<(&str, &str)>>,
|
||||
) -> Result<reqwest::Response> {
|
||||
let mut attempt = 0;
|
||||
let mut delay = self.retry_config.initial_delay_ms;
|
||||
|
||||
loop {
|
||||
let mut request = self.client
|
||||
.request(method.clone(), url)
|
||||
.basic_auth(&self.config.username, Some(&self.config.password));
|
||||
|
||||
if let Some(ref body_content) = body {
|
||||
request = request.body(body_content.clone());
|
||||
}
|
||||
|
||||
if let Some(ref headers_list) = headers {
|
||||
for (key, value) in headers_list {
|
||||
request = request.header(*key, *value);
|
||||
}
|
||||
}
|
||||
|
||||
match request.send().await {
|
||||
Ok(response) => {
|
||||
let status = response.status();
|
||||
|
||||
if status.is_success() || status.as_u16() == 207 {
|
||||
return Ok(response);
|
||||
}
|
||||
|
||||
// Handle rate limiting
|
||||
if status.as_u16() == 429 {
|
||||
warn!("Rate limited, backing off for {}ms", self.retry_config.rate_limit_backoff_ms);
|
||||
sleep(Duration::from_millis(self.retry_config.rate_limit_backoff_ms)).await;
|
||||
continue;
|
||||
}
|
||||
|
||||
// Handle client errors (don't retry)
|
||||
if status.is_client_error() && status.as_u16() != 429 {
|
||||
return Err(anyhow!("Client error: {} - {}", status,
|
||||
response.text().await.unwrap_or_default()));
|
||||
}
|
||||
|
||||
// Handle server errors (retry)
|
||||
if status.is_server_error() && attempt < self.retry_config.max_retries {
|
||||
warn!("Server error {}, retrying in {}ms (attempt {}/{})",
|
||||
status, delay, attempt + 1, self.retry_config.max_retries);
|
||||
|
||||
sleep(Duration::from_millis(delay)).await;
|
||||
delay = std::cmp::min(
|
||||
(delay as f64 * self.retry_config.backoff_multiplier) as u64,
|
||||
self.retry_config.max_delay_ms
|
||||
);
|
||||
attempt += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
return Err(anyhow!("Request failed: {} - {}", status,
|
||||
response.text().await.unwrap_or_default()));
|
||||
}
|
||||
Err(e) => {
|
||||
if attempt < self.retry_config.max_retries {
|
||||
warn!("Request error: {}, retrying in {}ms (attempt {}/{})",
|
||||
e, delay, attempt + 1, self.retry_config.max_retries);
|
||||
|
||||
sleep(Duration::from_millis(delay)).await;
|
||||
delay = std::cmp::min(
|
||||
(delay as f64 * self.retry_config.backoff_multiplier) as u64,
|
||||
self.retry_config.max_delay_ms
|
||||
);
|
||||
attempt += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
return Err(anyhow!("Request failed after {} attempts: {}",
|
||||
self.retry_config.max_retries, e));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Gets the WebDAV URL for a specific path
|
||||
pub fn get_url_for_path(&self, path: &str) -> String {
|
||||
let base_url = self.config.webdav_url();
|
||||
let clean_path = path.trim_start_matches('/');
|
||||
|
||||
if clean_path.is_empty() {
|
||||
base_url
|
||||
} else {
|
||||
format!("{}/{}", base_url.trim_end_matches('/'), clean_path)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,349 @@
|
|||
use anyhow::{anyhow, Result};
|
||||
use reqwest::Method;
|
||||
use std::collections::HashSet;
|
||||
use tokio::sync::Semaphore;
|
||||
use futures_util::stream::{self, StreamExt};
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
use crate::models::{FileInfo, WebDAVCrawlEstimate, WebDAVFolderInfo};
|
||||
use crate::webdav_xml_parser::{parse_propfind_response, parse_propfind_response_with_directories};
|
||||
use super::config::{WebDAVConfig, ConcurrencyConfig};
|
||||
use super::connection::WebDAVConnection;
|
||||
|
||||
pub struct WebDAVDiscovery {
|
||||
connection: WebDAVConnection,
|
||||
config: WebDAVConfig,
|
||||
concurrency_config: ConcurrencyConfig,
|
||||
}
|
||||
|
||||
impl WebDAVDiscovery {
|
||||
pub fn new(
|
||||
connection: WebDAVConnection,
|
||||
config: WebDAVConfig,
|
||||
concurrency_config: ConcurrencyConfig
|
||||
) -> Self {
|
||||
Self {
|
||||
connection,
|
||||
config,
|
||||
concurrency_config
|
||||
}
|
||||
}
|
||||
|
||||
/// Discovers files in a directory with support for pagination and filtering
|
||||
pub async fn discover_files(&self, directory_path: &str, recursive: bool) -> Result<Vec<FileInfo>> {
|
||||
info!("🔍 Discovering files in directory: {}", directory_path);
|
||||
|
||||
if recursive {
|
||||
self.discover_files_recursive(directory_path).await
|
||||
} else {
|
||||
self.discover_files_single_directory(directory_path).await
|
||||
}
|
||||
}
|
||||
|
||||
/// Discovers files in a single directory (non-recursive)
|
||||
async fn discover_files_single_directory(&self, directory_path: &str) -> Result<Vec<FileInfo>> {
|
||||
let url = self.connection.get_url_for_path(directory_path);
|
||||
|
||||
let propfind_body = r#"<?xml version="1.0" encoding="utf-8"?>
|
||||
<D:propfind xmlns:D="DAV:">
|
||||
<D:prop>
|
||||
<D:displayname/>
|
||||
<D:getcontentlength/>
|
||||
<D:getlastmodified/>
|
||||
<D:getetag/>
|
||||
<D:resourcetype/>
|
||||
<D:creationdate/>
|
||||
</D:prop>
|
||||
</D:propfind>"#;
|
||||
|
||||
let response = self.connection
|
||||
.authenticated_request(
|
||||
Method::from_bytes(b"PROPFIND")?,
|
||||
&url,
|
||||
Some(propfind_body.to_string()),
|
||||
Some(vec![
|
||||
("Depth", "1"),
|
||||
("Content-Type", "application/xml"),
|
||||
]),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let body = response.text().await?;
|
||||
let files = parse_propfind_response(&body)?;
|
||||
|
||||
// Filter files based on supported extensions
|
||||
let filtered_files: Vec<FileInfo> = files
|
||||
.into_iter()
|
||||
.filter(|file| {
|
||||
!file.is_directory && self.config.is_supported_extension(&file.name)
|
||||
})
|
||||
.collect();
|
||||
|
||||
debug!("Found {} supported files in directory: {}", filtered_files.len(), directory_path);
|
||||
Ok(filtered_files)
|
||||
}
|
||||
|
||||
/// Discovers files recursively in directory tree
|
||||
async fn discover_files_recursive(&self, root_directory: &str) -> Result<Vec<FileInfo>> {
|
||||
let mut all_files = Vec::new();
|
||||
let mut directories_to_scan = vec![root_directory.to_string()];
|
||||
let semaphore = Semaphore::new(self.concurrency_config.max_concurrent_scans);
|
||||
|
||||
while !directories_to_scan.is_empty() {
|
||||
let current_batch: Vec<String> = directories_to_scan
|
||||
.drain(..)
|
||||
.take(self.concurrency_config.max_concurrent_scans)
|
||||
.collect();
|
||||
|
||||
let tasks = current_batch.into_iter().map(|dir| {
|
||||
let semaphore = &semaphore;
|
||||
async move {
|
||||
let _permit = semaphore.acquire().await.unwrap();
|
||||
self.scan_directory_with_subdirs(&dir).await
|
||||
}
|
||||
});
|
||||
|
||||
let results = stream::iter(tasks)
|
||||
.buffer_unordered(self.concurrency_config.max_concurrent_scans)
|
||||
.collect::<Vec<_>>()
|
||||
.await;
|
||||
|
||||
for result in results {
|
||||
match result {
|
||||
Ok((files, subdirs)) => {
|
||||
all_files.extend(files);
|
||||
directories_to_scan.extend(subdirs);
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Failed to scan directory: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
info!("Recursive discovery found {} total files", all_files.len());
|
||||
Ok(all_files)
|
||||
}
|
||||
|
||||
/// Scans a directory and returns both files and subdirectories
|
||||
async fn scan_directory_with_subdirs(&self, directory_path: &str) -> Result<(Vec<FileInfo>, Vec<String>)> {
|
||||
let url = self.connection.get_url_for_path(directory_path);
|
||||
|
||||
let propfind_body = r#"<?xml version="1.0" encoding="utf-8"?>
|
||||
<D:propfind xmlns:D="DAV:">
|
||||
<D:prop>
|
||||
<D:displayname/>
|
||||
<D:getcontentlength/>
|
||||
<D:getlastmodified/>
|
||||
<D:getetag/>
|
||||
<D:resourcetype/>
|
||||
<D:creationdate/>
|
||||
</D:prop>
|
||||
</D:propfind>"#;
|
||||
|
||||
let response = self.connection
|
||||
.authenticated_request(
|
||||
Method::from_bytes(b"PROPFIND")?,
|
||||
&url,
|
||||
Some(propfind_body.to_string()),
|
||||
Some(vec![
|
||||
("Depth", "1"),
|
||||
("Content-Type", "application/xml"),
|
||||
]),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let body = response.text().await?;
|
||||
let (files, directories) = parse_propfind_response_with_directories(&body)?;
|
||||
|
||||
// Filter files by supported extensions
|
||||
let filtered_files: Vec<FileInfo> = files
|
||||
.into_iter()
|
||||
.filter(|file| self.config.is_supported_extension(&file.name))
|
||||
.collect();
|
||||
|
||||
// Convert directory paths to full paths
|
||||
let full_dir_paths: Vec<String> = directories
|
||||
.into_iter()
|
||||
.map(|dir| {
|
||||
if directory_path == "/" {
|
||||
format!("/{}", dir.trim_start_matches('/'))
|
||||
} else {
|
||||
format!("{}/{}", directory_path.trim_end_matches('/'), dir.trim_start_matches('/'))
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
debug!("Directory '{}': {} files, {} subdirectories",
|
||||
directory_path, filtered_files.len(), full_dir_paths.len());
|
||||
|
||||
Ok((filtered_files, full_dir_paths))
|
||||
}
|
||||
|
||||
/// Estimates crawl time and file counts for watch folders
|
||||
pub async fn estimate_crawl(&self) -> Result<WebDAVCrawlEstimate> {
|
||||
info!("📊 Estimating crawl for WebDAV watch folders");
|
||||
|
||||
let mut folders = Vec::new();
|
||||
let mut total_files = 0;
|
||||
let mut total_supported_files = 0;
|
||||
let mut total_size_mb = 0.0;
|
||||
|
||||
for watch_folder in &self.config.watch_folders {
|
||||
match self.estimate_folder(watch_folder).await {
|
||||
Ok(folder_info) => {
|
||||
total_files += folder_info.total_files;
|
||||
total_supported_files += folder_info.supported_files;
|
||||
total_size_mb += folder_info.total_size_mb;
|
||||
folders.push(folder_info);
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Failed to estimate folder '{}': {}", watch_folder, e);
|
||||
// Add empty folder info for failed estimates
|
||||
folders.push(WebDAVFolderInfo {
|
||||
path: watch_folder.clone(),
|
||||
total_files: 0,
|
||||
supported_files: 0,
|
||||
estimated_time_hours: 0.0,
|
||||
total_size_mb: 0.0,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Estimate total time based on file count and average processing time
|
||||
let avg_time_per_file_seconds = 2.0; // Conservative estimate
|
||||
let total_estimated_time_hours = (total_supported_files as f32 * avg_time_per_file_seconds) / 3600.0;
|
||||
|
||||
Ok(WebDAVCrawlEstimate {
|
||||
folders,
|
||||
total_files,
|
||||
total_supported_files,
|
||||
total_estimated_time_hours,
|
||||
total_size_mb,
|
||||
})
|
||||
}
|
||||
|
||||
/// Estimates file count and size for a specific folder
|
||||
async fn estimate_folder(&self, folder_path: &str) -> Result<WebDAVFolderInfo> {
|
||||
debug!("Estimating folder: {}", folder_path);
|
||||
|
||||
// Sample a few subdirectories to estimate the total
|
||||
let sample_files = self.discover_files_single_directory(folder_path).await?;
|
||||
|
||||
// Get subdirectories for deeper estimation
|
||||
let subdirs = self.get_subdirectories(folder_path).await?;
|
||||
|
||||
let mut total_files = sample_files.len() as i64;
|
||||
let mut total_size: i64 = sample_files.iter().map(|f| f.size).sum();
|
||||
|
||||
// Sample a few subdirectories to extrapolate
|
||||
let sample_size = std::cmp::min(5, subdirs.len());
|
||||
if sample_size > 0 {
|
||||
let mut sample_total = 0i64;
|
||||
|
||||
for subdir in subdirs.iter().take(sample_size) {
|
||||
if let Ok(subdir_files) = self.discover_files_single_directory(subdir).await {
|
||||
sample_total += subdir_files.len() as i64;
|
||||
}
|
||||
}
|
||||
|
||||
// Extrapolate based on sample
|
||||
if sample_total > 0 {
|
||||
let avg_files_per_subdir = sample_total as f64 / sample_size as f64;
|
||||
total_files += (avg_files_per_subdir * subdirs.len() as f64) as i64;
|
||||
}
|
||||
}
|
||||
|
||||
// Filter for supported files
|
||||
let supported_files = (total_files as f64 * self.calculate_support_ratio(&sample_files)) as i64;
|
||||
|
||||
let total_size_mb = total_size as f64 / (1024.0 * 1024.0);
|
||||
let estimated_time_hours = (supported_files as f32 * 2.0) / 3600.0; // 2 seconds per file
|
||||
|
||||
Ok(WebDAVFolderInfo {
|
||||
path: folder_path.to_string(),
|
||||
total_files,
|
||||
supported_files,
|
||||
estimated_time_hours,
|
||||
total_size_mb,
|
||||
})
|
||||
}
|
||||
|
||||
/// Gets subdirectories for a given path
|
||||
async fn get_subdirectories(&self, directory_path: &str) -> Result<Vec<String>> {
|
||||
let url = self.connection.get_url_for_path(directory_path);
|
||||
|
||||
let propfind_body = r#"<?xml version="1.0" encoding="utf-8"?>
|
||||
<D:propfind xmlns:D="DAV:">
|
||||
<D:prop>
|
||||
<D:resourcetype/>
|
||||
</D:prop>
|
||||
</D:propfind>"#;
|
||||
|
||||
let response = self.connection
|
||||
.authenticated_request(
|
||||
Method::from_bytes(b"PROPFIND")?,
|
||||
&url,
|
||||
Some(propfind_body.to_string()),
|
||||
Some(vec![
|
||||
("Depth", "1"),
|
||||
("Content-Type", "application/xml"),
|
||||
]),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let body = response.text().await?;
|
||||
let (_, directories) = parse_propfind_response_with_directories(&body)?;
|
||||
|
||||
Ok(directories)
|
||||
}
|
||||
|
||||
/// Calculates the ratio of supported files in a sample
|
||||
fn calculate_support_ratio(&self, sample_files: &[FileInfo]) -> f64 {
|
||||
if sample_files.is_empty() {
|
||||
return 1.0; // Assume all files are supported if no sample
|
||||
}
|
||||
|
||||
let supported_count = sample_files
|
||||
.iter()
|
||||
.filter(|file| self.config.is_supported_extension(&file.name))
|
||||
.count();
|
||||
|
||||
supported_count as f64 / sample_files.len() as f64
|
||||
}
|
||||
|
||||
/// Filters files by last modified date (for incremental syncs)
|
||||
pub fn filter_files_by_date(&self, files: Vec<FileInfo>, since: chrono::DateTime<chrono::Utc>) -> Vec<FileInfo> {
|
||||
files
|
||||
.into_iter()
|
||||
.filter(|file| {
|
||||
file.last_modified
|
||||
.map(|modified| modified > since)
|
||||
.unwrap_or(true) // Include files without modification date
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Deduplicates files by ETag or path
|
||||
pub fn deduplicate_files(&self, files: Vec<FileInfo>) -> Vec<FileInfo> {
|
||||
let mut seen_etags = HashSet::new();
|
||||
let mut seen_paths = HashSet::new();
|
||||
let mut deduplicated = Vec::new();
|
||||
|
||||
for file in files {
|
||||
let is_duplicate = if !file.etag.is_empty() {
|
||||
!seen_etags.insert(file.etag.clone())
|
||||
} else {
|
||||
!seen_paths.insert(file.path.clone())
|
||||
};
|
||||
|
||||
if !is_duplicate {
|
||||
deduplicated.push(file);
|
||||
}
|
||||
}
|
||||
|
||||
debug!("Deduplicated {} files", deduplicated.len());
|
||||
deduplicated
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
// WebDAV service modules organized by functionality
|
||||
|
||||
pub mod config;
|
||||
pub mod connection;
|
||||
pub mod discovery;
|
||||
pub mod validation;
|
||||
pub mod service;
|
||||
|
||||
// Re-export main types for convenience
|
||||
pub use config::{WebDAVConfig, RetryConfig, ConcurrencyConfig};
|
||||
pub use connection::WebDAVConnection;
|
||||
pub use discovery::WebDAVDiscovery;
|
||||
pub use validation::{
|
||||
WebDAVValidator, ValidationReport, ValidationIssue, ValidationIssueType,
|
||||
ValidationSeverity, ValidationRecommendation, ValidationAction, ValidationSummary
|
||||
};
|
||||
pub use service::{WebDAVService, ServerCapabilities, HealthStatus, test_webdav_connection};
|
||||
|
|
@ -0,0 +1,391 @@
|
|||
use anyhow::{anyhow, Result};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::Semaphore;
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
use crate::models::{
|
||||
FileInfo, WebDAVConnectionResult, WebDAVCrawlEstimate, WebDAVTestConnection,
|
||||
};
|
||||
|
||||
use super::config::{WebDAVConfig, RetryConfig, ConcurrencyConfig};
|
||||
use super::connection::WebDAVConnection;
|
||||
use super::discovery::WebDAVDiscovery;
|
||||
use super::validation::{WebDAVValidator, ValidationReport};
|
||||
|
||||
/// Main WebDAV service that coordinates all WebDAV operations
|
||||
pub struct WebDAVService {
|
||||
connection: Arc<WebDAVConnection>,
|
||||
discovery: Arc<WebDAVDiscovery>,
|
||||
validator: Arc<WebDAVValidator>,
|
||||
config: WebDAVConfig,
|
||||
retry_config: RetryConfig,
|
||||
concurrency_config: ConcurrencyConfig,
|
||||
scan_semaphore: Arc<Semaphore>,
|
||||
download_semaphore: Arc<Semaphore>,
|
||||
}
|
||||
|
||||
impl WebDAVService {
|
||||
/// Creates a new WebDAV service with default configurations
|
||||
pub fn new(config: WebDAVConfig) -> Result<Self> {
|
||||
Self::new_with_configs(config, RetryConfig::default(), ConcurrencyConfig::default())
|
||||
}
|
||||
|
||||
/// Creates a new WebDAV service with custom retry configuration
|
||||
pub fn new_with_retry(config: WebDAVConfig, retry_config: RetryConfig) -> Result<Self> {
|
||||
Self::new_with_configs(config, retry_config, ConcurrencyConfig::default())
|
||||
}
|
||||
|
||||
/// Creates a new WebDAV service with all custom configurations
|
||||
pub fn new_with_configs(
|
||||
config: WebDAVConfig,
|
||||
retry_config: RetryConfig,
|
||||
concurrency_config: ConcurrencyConfig
|
||||
) -> Result<Self> {
|
||||
// Validate configuration
|
||||
config.validate()?;
|
||||
|
||||
// Create connection handler
|
||||
let connection = Arc::new(WebDAVConnection::new(config.clone(), retry_config.clone())?);
|
||||
|
||||
// Create discovery handler
|
||||
let discovery = Arc::new(WebDAVDiscovery::new(
|
||||
connection.as_ref().clone(),
|
||||
config.clone(),
|
||||
concurrency_config.clone(),
|
||||
));
|
||||
|
||||
// Create validator
|
||||
let validator = Arc::new(WebDAVValidator::new(
|
||||
connection.as_ref().clone(),
|
||||
config.clone(),
|
||||
));
|
||||
|
||||
// Create semaphores for concurrency control
|
||||
let scan_semaphore = Arc::new(Semaphore::new(concurrency_config.max_concurrent_scans));
|
||||
let download_semaphore = Arc::new(Semaphore::new(concurrency_config.max_concurrent_downloads));
|
||||
|
||||
Ok(Self {
|
||||
connection,
|
||||
discovery,
|
||||
validator,
|
||||
config,
|
||||
retry_config,
|
||||
concurrency_config,
|
||||
scan_semaphore,
|
||||
download_semaphore,
|
||||
})
|
||||
}
|
||||
|
||||
/// Tests the WebDAV connection
|
||||
pub async fn test_connection(&self) -> Result<WebDAVConnectionResult> {
|
||||
info!("🔍 Testing WebDAV connection for service");
|
||||
self.connection.test_connection().await
|
||||
}
|
||||
|
||||
/// Tests WebDAV connection with provided configuration (static method)
|
||||
pub async fn test_connection_with_config(test_config: &WebDAVTestConnection) -> Result<WebDAVConnectionResult> {
|
||||
WebDAVConnection::test_connection_with_config(test_config).await
|
||||
}
|
||||
}
|
||||
|
||||
/// Tests WebDAV connection with provided configuration (standalone function for backward compatibility)
|
||||
pub async fn test_webdav_connection(test_config: &WebDAVTestConnection) -> Result<WebDAVConnectionResult> {
|
||||
WebDAVConnection::test_connection_with_config(test_config).await
|
||||
}
|
||||
|
||||
impl WebDAVService {
|
||||
/// Performs a comprehensive system validation
|
||||
pub async fn validate_system(&self) -> Result<ValidationReport> {
|
||||
info!("🔍 Performing comprehensive WebDAV system validation");
|
||||
self.validator.validate_system().await
|
||||
}
|
||||
|
||||
/// Estimates crawl time and resource requirements
|
||||
pub async fn estimate_crawl(&self) -> Result<WebDAVCrawlEstimate> {
|
||||
info!("📊 Estimating WebDAV crawl requirements");
|
||||
self.discovery.estimate_crawl().await
|
||||
}
|
||||
|
||||
/// Discovers all files in watch folders
|
||||
pub async fn discover_all_files(&self) -> Result<Vec<FileInfo>> {
|
||||
info!("🔍 Discovering all files in watch folders");
|
||||
let mut all_files = Vec::new();
|
||||
|
||||
for watch_folder in &self.config.watch_folders {
|
||||
info!("📁 Scanning watch folder: {}", watch_folder);
|
||||
|
||||
match self.discovery.discover_files(watch_folder, true).await {
|
||||
Ok(files) => {
|
||||
info!("✅ Found {} files in {}", files.len(), watch_folder);
|
||||
all_files.extend(files);
|
||||
}
|
||||
Err(e) => {
|
||||
error!("❌ Failed to scan watch folder '{}': {}", watch_folder, e);
|
||||
return Err(anyhow!("Failed to scan watch folder '{}': {}", watch_folder, e));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Deduplicate files across folders
|
||||
let deduplicated_files = self.discovery.deduplicate_files(all_files);
|
||||
|
||||
info!("🎯 Total unique files discovered: {}", deduplicated_files.len());
|
||||
Ok(deduplicated_files)
|
||||
}
|
||||
|
||||
/// Discovers files changed since a specific date (for incremental syncs)
|
||||
pub async fn discover_changed_files(&self, since: chrono::DateTime<chrono::Utc>) -> Result<Vec<FileInfo>> {
|
||||
info!("🔍 Discovering files changed since: {}", since);
|
||||
|
||||
let all_files = self.discover_all_files().await?;
|
||||
let changed_files = self.discovery.filter_files_by_date(all_files, since);
|
||||
|
||||
info!("📈 Found {} files changed since {}", changed_files.len(), since);
|
||||
Ok(changed_files)
|
||||
}
|
||||
|
||||
/// Discovers files in a specific directory
|
||||
pub async fn discover_files_in_directory(&self, directory_path: &str, recursive: bool) -> Result<Vec<FileInfo>> {
|
||||
info!("🔍 Discovering files in directory: {} (recursive: {})", directory_path, recursive);
|
||||
self.discovery.discover_files(directory_path, recursive).await
|
||||
}
|
||||
|
||||
/// Downloads a file from WebDAV server
|
||||
pub async fn download_file(&self, file_info: &FileInfo) -> Result<Vec<u8>> {
|
||||
let _permit = self.download_semaphore.acquire().await?;
|
||||
|
||||
debug!("⬇️ Downloading file: {}", file_info.path);
|
||||
|
||||
let url = self.connection.get_url_for_path(&file_info.path);
|
||||
|
||||
let response = self.connection
|
||||
.authenticated_request(
|
||||
reqwest::Method::GET,
|
||||
&url,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
|
||||
if !response.status().is_success() {
|
||||
return Err(anyhow!(
|
||||
"Failed to download file '{}': HTTP {}",
|
||||
file_info.path,
|
||||
response.status()
|
||||
));
|
||||
}
|
||||
|
||||
let content = response.bytes().await?;
|
||||
debug!("✅ Downloaded {} bytes for file: {}", content.len(), file_info.path);
|
||||
|
||||
Ok(content.to_vec())
|
||||
}
|
||||
|
||||
/// Downloads multiple files concurrently
|
||||
pub async fn download_files(&self, files: &[FileInfo]) -> Result<Vec<(FileInfo, Result<Vec<u8>>)>> {
|
||||
info!("⬇️ Downloading {} files concurrently", files.len());
|
||||
|
||||
let tasks = files.iter().map(|file| {
|
||||
let file_clone = file.clone();
|
||||
let service_clone = self.clone();
|
||||
|
||||
async move {
|
||||
let result = service_clone.download_file(&file_clone).await;
|
||||
(file_clone, result)
|
||||
}
|
||||
});
|
||||
|
||||
let results = futures_util::future::join_all(tasks).await;
|
||||
|
||||
let success_count = results.iter().filter(|(_, result)| result.is_ok()).count();
|
||||
let failure_count = results.len() - success_count;
|
||||
|
||||
info!("📊 Download completed: {} successful, {} failed", success_count, failure_count);
|
||||
|
||||
Ok(results)
|
||||
}
|
||||
|
||||
/// Gets file metadata without downloading content
|
||||
pub async fn get_file_metadata(&self, file_path: &str) -> Result<FileInfo> {
|
||||
debug!("📋 Getting metadata for file: {}", file_path);
|
||||
|
||||
let url = self.connection.get_url_for_path(file_path);
|
||||
|
||||
let propfind_body = r#"<?xml version="1.0" encoding="utf-8"?>
|
||||
<D:propfind xmlns:D="DAV:">
|
||||
<D:prop>
|
||||
<D:displayname/>
|
||||
<D:getcontentlength/>
|
||||
<D:getlastmodified/>
|
||||
<D:getetag/>
|
||||
<D:resourcetype/>
|
||||
<D:creationdate/>
|
||||
</D:prop>
|
||||
</D:propfind>"#;
|
||||
|
||||
let response = self.connection
|
||||
.authenticated_request(
|
||||
reqwest::Method::from_bytes(b"PROPFIND")?,
|
||||
&url,
|
||||
Some(propfind_body.to_string()),
|
||||
Some(vec![
|
||||
("Depth", "0"),
|
||||
("Content-Type", "application/xml"),
|
||||
]),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let body = response.text().await?;
|
||||
let files = crate::webdav_xml_parser::parse_propfind_response(&body)?;
|
||||
|
||||
files.into_iter()
|
||||
.find(|f| f.path == file_path)
|
||||
.ok_or_else(|| anyhow!("File metadata not found: {}", file_path))
|
||||
}
|
||||
|
||||
/// Checks if a file exists on the WebDAV server
|
||||
pub async fn file_exists(&self, file_path: &str) -> Result<bool> {
|
||||
match self.get_file_metadata(file_path).await {
|
||||
Ok(_) => Ok(true),
|
||||
Err(_) => Ok(false),
|
||||
}
|
||||
}
|
||||
|
||||
/// Gets the server capabilities and features
|
||||
pub async fn get_server_capabilities(&self) -> Result<ServerCapabilities> {
|
||||
debug!("🔍 Checking server capabilities");
|
||||
|
||||
let options_response = self.connection
|
||||
.authenticated_request(
|
||||
reqwest::Method::OPTIONS,
|
||||
&self.config.webdav_url(),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let dav_header = options_response
|
||||
.headers()
|
||||
.get("dav")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
|
||||
let allow_header = options_response
|
||||
.headers()
|
||||
.get("allow")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
|
||||
let server_header = options_response
|
||||
.headers()
|
||||
.get("server")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map(|s| s.to_string());
|
||||
|
||||
Ok(ServerCapabilities {
|
||||
dav_compliance: dav_header,
|
||||
allowed_methods: allow_header,
|
||||
server_software: server_header,
|
||||
supports_etag: dav_header.contains("1") || dav_header.contains("2"),
|
||||
supports_depth_infinity: dav_header.contains("1"),
|
||||
})
|
||||
}
|
||||
|
||||
/// Performs a health check on the WebDAV service
|
||||
pub async fn health_check(&self) -> Result<HealthStatus> {
|
||||
info!("🏥 Performing WebDAV service health check");
|
||||
|
||||
let start_time = std::time::Instant::now();
|
||||
|
||||
// Test basic connectivity
|
||||
let connection_result = self.test_connection().await?;
|
||||
if !connection_result.success {
|
||||
return Ok(HealthStatus {
|
||||
healthy: false,
|
||||
message: format!("Connection failed: {}", connection_result.message),
|
||||
response_time_ms: start_time.elapsed().as_millis() as u64,
|
||||
details: None,
|
||||
});
|
||||
}
|
||||
|
||||
// Test each watch folder
|
||||
for folder in &self.config.watch_folders {
|
||||
if let Err(e) = self.connection.test_propfind(folder).await {
|
||||
return Ok(HealthStatus {
|
||||
healthy: false,
|
||||
message: format!("Watch folder '{}' is inaccessible: {}", folder, e),
|
||||
response_time_ms: start_time.elapsed().as_millis() as u64,
|
||||
details: Some(serde_json::json!({
|
||||
"failed_folder": folder,
|
||||
"error": e.to_string()
|
||||
})),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
let response_time = start_time.elapsed().as_millis() as u64;
|
||||
|
||||
Ok(HealthStatus {
|
||||
healthy: true,
|
||||
message: "All systems operational".to_string(),
|
||||
response_time_ms: response_time,
|
||||
details: Some(serde_json::json!({
|
||||
"tested_folders": self.config.watch_folders,
|
||||
"server_type": connection_result.server_type,
|
||||
"server_version": connection_result.server_version
|
||||
})),
|
||||
})
|
||||
}
|
||||
|
||||
/// Gets configuration information
|
||||
pub fn get_config(&self) -> &WebDAVConfig {
|
||||
&self.config
|
||||
}
|
||||
|
||||
/// Gets retry configuration
|
||||
pub fn get_retry_config(&self) -> &RetryConfig {
|
||||
&self.retry_config
|
||||
}
|
||||
|
||||
/// Gets concurrency configuration
|
||||
pub fn get_concurrency_config(&self) -> &ConcurrencyConfig {
|
||||
&self.concurrency_config
|
||||
}
|
||||
}
|
||||
|
||||
// Implement Clone to allow sharing the service
|
||||
impl Clone for WebDAVService {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
connection: Arc::clone(&self.connection),
|
||||
discovery: Arc::clone(&self.discovery),
|
||||
validator: Arc::clone(&self.validator),
|
||||
config: self.config.clone(),
|
||||
retry_config: self.retry_config.clone(),
|
||||
concurrency_config: self.concurrency_config.clone(),
|
||||
scan_semaphore: Arc::clone(&self.scan_semaphore),
|
||||
download_semaphore: Arc::clone(&self.download_semaphore),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Server capabilities information
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ServerCapabilities {
|
||||
pub dav_compliance: String,
|
||||
pub allowed_methods: String,
|
||||
pub server_software: Option<String>,
|
||||
pub supports_etag: bool,
|
||||
pub supports_depth_infinity: bool,
|
||||
}
|
||||
|
||||
/// Health status information
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct HealthStatus {
|
||||
pub healthy: bool,
|
||||
pub message: String,
|
||||
pub response_time_ms: u64,
|
||||
pub details: Option<serde_json::Value>,
|
||||
}
|
||||
|
|
@ -0,0 +1,352 @@
|
|||
use anyhow::{anyhow, Result};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
use super::config::WebDAVConfig;
|
||||
use super::connection::WebDAVConnection;
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ValidationReport {
|
||||
pub overall_health_score: i32, // 0-100
|
||||
pub issues: Vec<ValidationIssue>,
|
||||
pub recommendations: Vec<ValidationRecommendation>,
|
||||
pub summary: ValidationSummary,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ValidationIssue {
|
||||
pub issue_type: ValidationIssueType,
|
||||
pub severity: ValidationSeverity,
|
||||
pub directory_path: String,
|
||||
pub description: String,
|
||||
pub details: Option<serde_json::Value>,
|
||||
pub detected_at: chrono::DateTime<chrono::Utc>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum ValidationIssueType {
|
||||
/// Directory exists on server but not in our tracking
|
||||
Untracked,
|
||||
/// Directory in our tracking but missing on server
|
||||
Missing,
|
||||
/// ETag mismatch between server and our cache
|
||||
ETagMismatch,
|
||||
/// Directory hasn't been scanned in a very long time
|
||||
Stale,
|
||||
/// Server errors when accessing directory
|
||||
Inaccessible,
|
||||
/// ETag support seems unreliable for this directory
|
||||
ETagUnreliable,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum ValidationSeverity {
|
||||
Info, // No action needed, just FYI
|
||||
Warning, // Should investigate but not urgent
|
||||
Error, // Needs immediate attention
|
||||
Critical, // System integrity at risk
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ValidationRecommendation {
|
||||
pub action: ValidationAction,
|
||||
pub reason: String,
|
||||
pub affected_directories: Vec<String>,
|
||||
pub priority: ValidationSeverity,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub enum ValidationAction {
|
||||
/// Run a deep scan of specific directories
|
||||
DeepScanRequired,
|
||||
/// Clear and rebuild directory tracking
|
||||
RebuildTracking,
|
||||
/// ETag support is unreliable, switch to periodic scans
|
||||
DisableETagOptimization,
|
||||
/// Clean up orphaned database entries
|
||||
CleanupDatabase,
|
||||
/// Server configuration issue needs attention
|
||||
CheckServerConfiguration,
|
||||
/// No action needed, system is healthy
|
||||
NoActionRequired,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct ValidationSummary {
|
||||
pub total_directories_checked: usize,
|
||||
pub healthy_directories: usize,
|
||||
pub directories_with_issues: usize,
|
||||
pub critical_issues: usize,
|
||||
pub warning_issues: usize,
|
||||
pub info_issues: usize,
|
||||
pub validation_duration_ms: u64,
|
||||
}
|
||||
|
||||
pub struct WebDAVValidator {
|
||||
connection: WebDAVConnection,
|
||||
config: WebDAVConfig,
|
||||
}
|
||||
|
||||
impl WebDAVValidator {
|
||||
pub fn new(connection: WebDAVConnection, config: WebDAVConfig) -> Self {
|
||||
Self { connection, config }
|
||||
}
|
||||
|
||||
/// Performs comprehensive validation of WebDAV setup and directory tracking
|
||||
pub async fn validate_system(&self) -> Result<ValidationReport> {
|
||||
let start_time = std::time::Instant::now();
|
||||
info!("🔍 Starting WebDAV system validation");
|
||||
|
||||
let mut issues = Vec::new();
|
||||
let mut total_checked = 0;
|
||||
|
||||
// Test basic connectivity
|
||||
match self.connection.test_connection().await {
|
||||
Ok(result) if !result.success => {
|
||||
issues.push(ValidationIssue {
|
||||
issue_type: ValidationIssueType::Inaccessible,
|
||||
severity: ValidationSeverity::Critical,
|
||||
directory_path: "/".to_string(),
|
||||
description: format!("WebDAV server connection failed: {}", result.message),
|
||||
details: None,
|
||||
detected_at: chrono::Utc::now(),
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
issues.push(ValidationIssue {
|
||||
issue_type: ValidationIssueType::Inaccessible,
|
||||
severity: ValidationSeverity::Critical,
|
||||
directory_path: "/".to_string(),
|
||||
description: format!("WebDAV server connectivity test failed: {}", e),
|
||||
details: None,
|
||||
detected_at: chrono::Utc::now(),
|
||||
});
|
||||
}
|
||||
_ => {
|
||||
debug!("✅ Basic connectivity test passed");
|
||||
}
|
||||
}
|
||||
|
||||
// Validate each watch folder
|
||||
for folder in &self.config.watch_folders {
|
||||
total_checked += 1;
|
||||
if let Err(e) = self.validate_watch_folder(folder, &mut issues).await {
|
||||
warn!("Failed to validate watch folder '{}': {}", folder, e);
|
||||
}
|
||||
}
|
||||
|
||||
// Test ETag reliability
|
||||
self.validate_etag_support(&mut issues).await?;
|
||||
|
||||
// Generate recommendations based on issues
|
||||
let recommendations = self.generate_recommendations(&issues);
|
||||
|
||||
let validation_duration = start_time.elapsed().as_millis() as u64;
|
||||
let health_score = self.calculate_health_score(&issues);
|
||||
|
||||
let summary = ValidationSummary {
|
||||
total_directories_checked: total_checked,
|
||||
healthy_directories: total_checked - issues.len(),
|
||||
directories_with_issues: issues.len(),
|
||||
critical_issues: issues.iter().filter(|i| matches!(i.severity, ValidationSeverity::Critical)).count(),
|
||||
warning_issues: issues.iter().filter(|i| matches!(i.severity, ValidationSeverity::Warning)).count(),
|
||||
info_issues: issues.iter().filter(|i| matches!(i.severity, ValidationSeverity::Info)).count(),
|
||||
validation_duration_ms: validation_duration,
|
||||
};
|
||||
|
||||
info!("✅ WebDAV validation completed in {}ms. Health score: {}/100",
|
||||
validation_duration, health_score);
|
||||
|
||||
Ok(ValidationReport {
|
||||
overall_health_score: health_score,
|
||||
issues,
|
||||
recommendations,
|
||||
summary,
|
||||
})
|
||||
}
|
||||
|
||||
/// Validates a specific watch folder
|
||||
async fn validate_watch_folder(&self, folder: &str, issues: &mut Vec<ValidationIssue>) -> Result<()> {
|
||||
debug!("Validating watch folder: {}", folder);
|
||||
|
||||
// Test PROPFIND access
|
||||
match self.connection.test_propfind(folder).await {
|
||||
Ok(_) => {
|
||||
debug!("✅ Watch folder '{}' is accessible", folder);
|
||||
}
|
||||
Err(e) => {
|
||||
issues.push(ValidationIssue {
|
||||
issue_type: ValidationIssueType::Inaccessible,
|
||||
severity: ValidationSeverity::Error,
|
||||
directory_path: folder.to_string(),
|
||||
description: format!("Cannot access watch folder: {}", e),
|
||||
details: Some(serde_json::json!({
|
||||
"error": e.to_string(),
|
||||
"folder": folder
|
||||
})),
|
||||
detected_at: chrono::Utc::now(),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Tests ETag support reliability
|
||||
async fn validate_etag_support(&self, issues: &mut Vec<ValidationIssue>) -> Result<()> {
|
||||
debug!("Testing ETag support reliability");
|
||||
|
||||
// Test ETag consistency across multiple requests
|
||||
for folder in &self.config.watch_folders {
|
||||
if let Err(e) = self.test_etag_consistency(folder, issues).await {
|
||||
warn!("ETag consistency test failed for '{}': {}", folder, e);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Tests ETag consistency for a specific folder
|
||||
async fn test_etag_consistency(&self, folder: &str, issues: &mut Vec<ValidationIssue>) -> Result<()> {
|
||||
// Make two consecutive PROPFIND requests and compare ETags
|
||||
let etag1 = self.get_folder_etag(folder).await?;
|
||||
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
|
||||
let etag2 = self.get_folder_etag(folder).await?;
|
||||
|
||||
if etag1 != etag2 && etag1.is_some() && etag2.is_some() {
|
||||
issues.push(ValidationIssue {
|
||||
issue_type: ValidationIssueType::ETagUnreliable,
|
||||
severity: ValidationSeverity::Warning,
|
||||
directory_path: folder.to_string(),
|
||||
description: "ETag values are inconsistent across requests".to_string(),
|
||||
details: Some(serde_json::json!({
|
||||
"etag1": etag1,
|
||||
"etag2": etag2,
|
||||
"folder": folder
|
||||
})),
|
||||
detected_at: chrono::Utc::now(),
|
||||
});
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Gets the ETag for a folder
|
||||
async fn get_folder_etag(&self, folder: &str) -> Result<Option<String>> {
|
||||
let url = self.connection.get_url_for_path(folder);
|
||||
|
||||
let propfind_body = r#"<?xml version="1.0" encoding="utf-8"?>
|
||||
<D:propfind xmlns:D="DAV:">
|
||||
<D:prop>
|
||||
<D:getetag/>
|
||||
</D:prop>
|
||||
</D:propfind>"#;
|
||||
|
||||
let response = self.connection
|
||||
.authenticated_request(
|
||||
reqwest::Method::from_bytes(b"PROPFIND")?,
|
||||
&url,
|
||||
Some(propfind_body.to_string()),
|
||||
Some(vec![
|
||||
("Depth", "0"),
|
||||
("Content-Type", "application/xml"),
|
||||
]),
|
||||
)
|
||||
.await?;
|
||||
|
||||
let body = response.text().await?;
|
||||
|
||||
// Parse ETag from XML response (simplified)
|
||||
if let Some(start) = body.find("<D:getetag>") {
|
||||
if let Some(end) = body[start..].find("</D:getetag>") {
|
||||
let etag = &body[start + 11..start + end];
|
||||
return Ok(Some(etag.trim_matches('"').to_string()));
|
||||
}
|
||||
}
|
||||
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
/// Generates recommendations based on detected issues
|
||||
fn generate_recommendations(&self, issues: &Vec<ValidationIssue>) -> Vec<ValidationRecommendation> {
|
||||
let mut recommendations = Vec::new();
|
||||
let mut directories_by_issue: HashMap<ValidationIssueType, Vec<String>> = HashMap::new();
|
||||
|
||||
// Group directories by issue type
|
||||
for issue in issues {
|
||||
directories_by_issue
|
||||
.entry(issue.issue_type.clone())
|
||||
.or_insert_with(Vec::new)
|
||||
.push(issue.directory_path.clone());
|
||||
}
|
||||
|
||||
// Generate recommendations for each issue type
|
||||
for (issue_type, directories) in directories_by_issue {
|
||||
let recommendation = match issue_type {
|
||||
ValidationIssueType::Inaccessible => ValidationRecommendation {
|
||||
action: ValidationAction::CheckServerConfiguration,
|
||||
reason: "Some directories are inaccessible. Check server configuration and permissions.".to_string(),
|
||||
affected_directories: directories,
|
||||
priority: ValidationSeverity::Critical,
|
||||
},
|
||||
ValidationIssueType::ETagUnreliable => ValidationRecommendation {
|
||||
action: ValidationAction::DisableETagOptimization,
|
||||
reason: "ETag support appears unreliable. Consider disabling ETag optimization.".to_string(),
|
||||
affected_directories: directories,
|
||||
priority: ValidationSeverity::Warning,
|
||||
},
|
||||
ValidationIssueType::Missing => ValidationRecommendation {
|
||||
action: ValidationAction::CleanupDatabase,
|
||||
reason: "Some tracked directories no longer exist on the server.".to_string(),
|
||||
affected_directories: directories,
|
||||
priority: ValidationSeverity::Warning,
|
||||
},
|
||||
ValidationIssueType::Stale => ValidationRecommendation {
|
||||
action: ValidationAction::DeepScanRequired,
|
||||
reason: "Some directories haven't been scanned recently.".to_string(),
|
||||
affected_directories: directories,
|
||||
priority: ValidationSeverity::Info,
|
||||
},
|
||||
_ => ValidationRecommendation {
|
||||
action: ValidationAction::DeepScanRequired,
|
||||
reason: "General validation issues detected.".to_string(),
|
||||
affected_directories: directories,
|
||||
priority: ValidationSeverity::Warning,
|
||||
},
|
||||
};
|
||||
recommendations.push(recommendation);
|
||||
}
|
||||
|
||||
if recommendations.is_empty() {
|
||||
recommendations.push(ValidationRecommendation {
|
||||
action: ValidationAction::NoActionRequired,
|
||||
reason: "System validation passed successfully.".to_string(),
|
||||
affected_directories: Vec::new(),
|
||||
priority: ValidationSeverity::Info,
|
||||
});
|
||||
}
|
||||
|
||||
recommendations
|
||||
}
|
||||
|
||||
/// Calculates overall health score based on issues
|
||||
fn calculate_health_score(&self, issues: &Vec<ValidationIssue>) -> i32 {
|
||||
if issues.is_empty() {
|
||||
return 100;
|
||||
}
|
||||
|
||||
let mut penalty = 0;
|
||||
for issue in issues {
|
||||
let issue_penalty = match issue.severity {
|
||||
ValidationSeverity::Critical => 30,
|
||||
ValidationSeverity::Error => 20,
|
||||
ValidationSeverity::Warning => 10,
|
||||
ValidationSeverity::Info => 5,
|
||||
};
|
||||
penalty += issue_penalty;
|
||||
}
|
||||
|
||||
std::cmp::max(0, 100 - penalty)
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -1,4 +1,4 @@
|
|||
use crate::services::webdav_service::{WebDAVConfig, WebDAVService};
|
||||
use crate::services::webdav::{WebDAVConfig, WebDAVService};
|
||||
|
||||
fn create_test_config() -> WebDAVConfig {
|
||||
WebDAVConfig {
|
||||
|
|
|
|||
Loading…
Reference in New Issue