From 165a671eb79f227665dd5a38308ffad5e834bfd3 Mon Sep 17 00:00:00 2001 From: Tom Hicks Date: Sun, 23 Aug 2026 06:13:29 -0700 Subject: [PATCH] Implements webdav uploads. --- Cargo.lock | 4 + Tasks.md | 14 +-- nextcloud_client/Cargo.toml | 4 + nextcloud_client/src/client.rs | 4 +- nextcloud_client/src/lib.rs | 3 + nextcloud_client/src/progress.rs | 152 +++++++++++++++++++++++++++++++ nextcloud_client/src/webdav.rs | 152 +++++++++++++++++++++++++++++++ 7 files changed, 324 insertions(+), 9 deletions(-) create mode 100644 nextcloud_client/src/progress.rs create mode 100644 nextcloud_client/src/webdav.rs diff --git a/Cargo.lock b/Cargo.lock index c530fc8..28705c8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2174,11 +2174,15 @@ checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086" name = "nextcloud_client" version = "0.1.0" dependencies = [ + "bytes", + "futures-util", + "pin-project-lite", "reqwest 0.12.28", "serde", "serde_json", "thiserror 2.0.20", "tokio", + "tokio-util", "url", ] diff --git a/Tasks.md b/Tasks.md index a1de89f..c6ddbf4 100644 --- a/Tasks.md +++ b/Tasks.md @@ -71,7 +71,7 @@ This document defines the complete project roadmap and task tracking system for |---|---|---|---| | [NUT-001](#nut-001) | Establish Repository Structure | Fixed | Foundation | | [NUT-002](#nut-002) | Implement Shared Rust Backend Library | Fixed | Foundation | -| [NUT-003](#nut-003) | Implement WebDAV Upload Logic | Triage | Feature | +| [NUT-003](#nut-003) | Implement WebDAV Upload Logic | Fixed | Feature | | [NUT-004](#nut-004) | Implement OCS Share Link Generation | Triage | Feature | | [NUT-005](#nut-005) | Implement Direct Download URL Builder | Triage | Feature | | [NUT-006](#nut-006) | Implement Credential Storage System | Triage | Feature | @@ -136,20 +136,20 @@ Implement the shared Rust library that provides all core functionality: WebDAV u - NUT-001 - + ### Implement WebDAV Upload Logic **ID:** NUT-003 -**Status:** Triage +**Status:** Fixed **Type:** Feature **Description:** Implement file upload using Nextcloud’s WebDAV API. Support streaming uploads, file size detection, and progress callbacks. **Requirements:** -- [ ] Implement PUT request to WebDAV endpoint -- [ ] Support streaming from file or stdin -- [ ] Provide progress callback API -- [ ] Handle authentication +- [x] Implement PUT request to WebDAV endpoint +- [x] Support streaming from file or stdin +- [x] Provide progress callback API +- [x] Handle authentication **Dependencies:** - NUT-002 diff --git a/nextcloud_client/Cargo.toml b/nextcloud_client/Cargo.toml index ccf391d..212952a 100644 --- a/nextcloud_client/Cargo.toml +++ b/nextcloud_client/Cargo.toml @@ -6,6 +6,10 @@ edition = "2024" [dependencies] reqwest = { version = "0.12", default-features = false, features = ["json", "stream", "default-tls"] } tokio = { version = "1", features = ["sync", "fs", "io-util", "macros", "time"] } +tokio-util = { version = "0.7", features = ["io"] } +futures-util = { version = "0.3", default-features = false } +pin-project-lite = "0.2" +bytes = "1.0" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" thiserror = "2.0" diff --git a/nextcloud_client/src/client.rs b/nextcloud_client/src/client.rs index e8c6923..4cf645a 100644 --- a/nextcloud_client/src/client.rs +++ b/nextcloud_client/src/client.rs @@ -24,8 +24,8 @@ pub struct ServerStatus { /// The core Nextcloud client instance. #[derive(Debug, Clone)] pub struct NextcloudClient { - config: ClientConfig, - http: reqwest::Client, + pub(crate) config: ClientConfig, + pub(crate) http: reqwest::Client, } impl NextcloudClient { diff --git a/nextcloud_client/src/lib.rs b/nextcloud_client/src/lib.rs index 44315c1..017baf1 100644 --- a/nextcloud_client/src/lib.rs +++ b/nextcloud_client/src/lib.rs @@ -7,9 +7,12 @@ pub mod client; pub mod config; pub mod error; pub mod models; +pub mod progress; +pub mod webdav; // Convenient top-level re-exports pub use client::{NextcloudClient, ServerStatus}; pub use config::{AccountCredentials, ClientConfig}; pub use error::{NextcloudError, Result}; pub use models::{UploadOptions, UploadResult}; +pub use progress::{ProgressCallback, ProgressEvent}; diff --git a/nextcloud_client/src/progress.rs b/nextcloud_client/src/progress.rs new file mode 100644 index 0000000..b2182f8 --- /dev/null +++ b/nextcloud_client/src/progress.rs @@ -0,0 +1,152 @@ +use bytes::Bytes; +use futures_util::Stream; +use pin_project_lite::pin_project; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; + +/// Event fired during streaming file uploads to report transfer progress. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ProgressEvent { + /// Transfer has started. `total_bytes` is `None` if streaming from an unsized pipe / stdin. + Started { total_bytes: Option }, + + /// Transfer progress update with running count of transferred bytes. + Progress { + bytes_transferred: u64, + total_bytes: Option, + }, + + /// Transfer has finished successfully. + Completed { total_bytes: u64 }, +} + +/// A thread-safe callback invoked when transfer progress occurs. +pub type ProgressCallback = Arc; + +pin_project! { + /// A Stream wrapper that tracks transferred bytes and fires progress callbacks. + pub struct ProgressStream { + #[pin] + inner: S, + total_bytes: Option, + bytes_transferred: u64, + callback: Option, + started: bool, + completed: bool, + } +} + +impl ProgressStream { + pub fn new( + inner: S, + total_bytes: Option, + callback: Option, + ) -> Self { + Self { + inner, + total_bytes, + bytes_transferred: 0, + callback, + started: false, + completed: false, + } + } +} + +impl Stream for ProgressStream +where + S: Stream>, +{ + type Item = std::result::Result; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = self.project(); + + if !*this.started { + *this.started = true; + if let Some(cb) = this.callback.as_ref() { + cb(ProgressEvent::Started { + total_bytes: *this.total_bytes, + }); + } + } + + match this.inner.poll_next(cx) { + Poll::Ready(Some(Ok(chunk))) => { + *this.bytes_transferred += chunk.len() as u64; + if let Some(cb) = this.callback.as_ref() { + cb(ProgressEvent::Progress { + bytes_transferred: *this.bytes_transferred, + total_bytes: *this.total_bytes, + }); + } + Poll::Ready(Some(Ok(chunk))) + } + Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(err))), + Poll::Ready(None) => { + if !*this.completed { + *this.completed = true; + if let Some(cb) = this.callback.as_ref() { + cb(ProgressEvent::Completed { + total_bytes: *this.bytes_transferred, + }); + } + } + Poll::Ready(None) + } + Poll::Pending => Poll::Pending, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use futures_util::stream; + use futures_util::StreamExt; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::Mutex; + + #[tokio::test] + async fn test_progress_stream_events() { + let chunk1 = Bytes::from_static(b"hello "); + let chunk2 = Bytes::from_static(b"world!"); + let total_size = 12u64; + + let events = Arc::new(Mutex::new(Vec::new())); + let last_bytes = Arc::new(AtomicU64::new(0)); + + let events_clone = Arc::clone(&events); + let last_bytes_clone = Arc::clone(&last_bytes); + + let callback = Arc::new(move |event: ProgressEvent| { + if let ProgressEvent::Progress { bytes_transferred, .. } = event { + last_bytes_clone.store(bytes_transferred, Ordering::SeqCst); + } + events_clone.lock().unwrap().push(event); + }); + + let raw_stream = stream::iter(vec![ + Ok::(chunk1), + Ok::(chunk2), + ]); + + let mut progress_stream = ProgressStream::new(raw_stream, Some(total_size), Some(callback)); + + let mut collected = Vec::new(); + while let Some(item) = progress_stream.next().await { + collected.extend_from_slice(&item.unwrap()); + } + + assert_eq!(&collected, b"hello world!"); + assert_eq!(last_bytes.load(Ordering::SeqCst), 12); + + let captured = events.lock().unwrap().clone(); + assert_eq!(captured.len(), 4); // Started, Progress(6), Progress(12), Completed(12) + assert_eq!(captured[0], ProgressEvent::Started { total_bytes: Some(12) }); + assert_eq!(captured[1], ProgressEvent::Progress { bytes_transferred: 6, total_bytes: Some(12) }); + assert_eq!(captured[2], ProgressEvent::Progress { bytes_transferred: 12, total_bytes: Some(12) }); + assert_eq!(captured[3], ProgressEvent::Completed { total_bytes: 12 }); + } +} diff --git a/nextcloud_client/src/webdav.rs b/nextcloud_client/src/webdav.rs new file mode 100644 index 0000000..9c59f1a --- /dev/null +++ b/nextcloud_client/src/webdav.rs @@ -0,0 +1,152 @@ +use bytes::Bytes; +use futures_util::Stream; +use reqwest::header::CONTENT_LENGTH; +use reqwest::Method; +use std::path::Path; +use tokio::io::AsyncRead; +use tokio_util::io::ReaderStream; + +use crate::client::NextcloudClient; +use crate::error::{NextcloudError, Result}; +use crate::progress::{ProgressCallback, ProgressStream}; + +/// Default buffer size for streaming file uploads (64 KiB). +pub const DEFAULT_CHUNK_SIZE: usize = 64 * 1024; + +impl NextcloudClient { + /// Upload a stream of bytes to a Nextcloud WebDAV destination with optional progress reporting. + /// + /// # Arguments + /// * `remote_path` - The destination path on Nextcloud (e.g. `"Uploads/photo.jpg"`). + /// * `stream` - An async stream yielding chunks of bytes. + /// * `content_length` - The total size of the stream in bytes, if known. + /// * `callback` - Optional callback invoked as byte chunks are transferred. + pub async fn upload_stream( + &self, + remote_path: &str, + stream: S, + content_length: Option, + callback: Option, + ) -> Result + where + S: Stream> + Send + Sync + 'static, + E: Into> + 'static, + { + let url = self.webdav_url(remote_path)?; + let progress_stream = ProgressStream::new(stream, content_length, callback); + let body = reqwest::Body::wrap_stream(progress_stream); + + let mut req = self.http.put(url).body(body); + if let Some(len) = content_length { + req = req.header(CONTENT_LENGTH, len); + } + + let response = req.send().await?; + let status = response.status(); + + if status.is_success() { + // 200 OK, 201 Created, or 204 No Content + Ok(content_length.unwrap_or(0)) + } else if status == reqwest::StatusCode::UNAUTHORIZED { + Err(NextcloudError::AuthenticationFailed { + username: self.username().unwrap_or("unknown").to_string(), + message: "Invalid credentials or unauthorized WebDAV access".to_string(), + }) + } else if status == reqwest::StatusCode::NOT_FOUND { + Err(NextcloudError::NotFound { + path: remote_path.to_string(), + }) + } else if status == reqwest::StatusCode::CONFLICT { + Err(NextcloudError::Other(format!( + "WebDAV Conflict (409) at '{remote_path}'. Ensure the parent directory exists on Nextcloud." + ))) + } else { + let error_text = response.text().await.unwrap_or_default(); + Err(NextcloudError::ServerError { + status: status.as_u16(), + message: error_text, + }) + } + } + + /// Upload a local file from disk to Nextcloud WebDAV. + /// + /// # Arguments + /// * `local_path` - Path to the local file to read and upload. + /// * `remote_path` - The destination path on Nextcloud. + /// * `callback` - Optional callback for real-time upload progress. + pub async fn upload_file>( + &self, + local_path: P, + remote_path: &str, + callback: Option, + ) -> Result { + let local_path = local_path.as_ref(); + let metadata = tokio::fs::metadata(local_path).await?; + + if !metadata.is_file() { + return Err(NextcloudError::InvalidPath { + path: local_path.display().to_string(), + reason: "Specified path is not a file".to_string(), + }); + } + + let total_size = metadata.len(); + let file = tokio::fs::File::open(local_path).await?; + let stream = ReaderStream::with_capacity(file, DEFAULT_CHUNK_SIZE); + + self.upload_stream(remote_path, stream, Some(total_size), callback) + .await + } + + /// Upload data from any async reader (e.g. standard input `tokio::io::stdin()`). + /// + /// # Arguments + /// * `reader` - An async reader implementing [`AsyncRead`]. + /// * `remote_path` - The destination path on Nextcloud. + /// * `total_size` - Known byte size, or `None` for chunked transfer. + /// * `callback` - Optional callback for upload progress. + pub async fn upload_reader( + &self, + reader: R, + remote_path: &str, + total_size: Option, + callback: Option, + ) -> Result + where + R: AsyncRead + Send + Sync + 'static, + { + let stream = ReaderStream::with_capacity(reader, DEFAULT_CHUNK_SIZE); + self.upload_stream(remote_path, stream, total_size, callback) + .await + } + + /// Create a remote directory using the WebDAV `MKCOL` method. + /// + /// Returns `Ok(())` if the directory was created (201) or already exists (405). + pub async fn create_folder(&self, remote_dir: &str) -> Result<()> { + let url = self.webdav_url(remote_dir)?; + let mkcol_method = Method::from_bytes(b"MKCOL") + .map_err(|e| NextcloudError::Other(e.to_string()))?; + + let response = self.http.request(mkcol_method, url).send().await?; + let status = response.status(); + + // 201 Created = success + // 405 Method Not Allowed = directory already exists in WebDAV + if status.is_success() || status == reqwest::StatusCode::METHOD_NOT_ALLOWED { + Ok(()) + } else if status == reqwest::StatusCode::UNAUTHORIZED { + Err(NextcloudError::AuthenticationFailed { + username: self.username().unwrap_or("unknown").to_string(), + message: "Unauthorized to create folder".to_string(), + }) + } else { + let error_text = response.text().await.unwrap_or_default(); + Err(NextcloudError::ServerError { + status: status.as_u16(), + message: error_text, + }) + } + } +}