Implements webdav uploads.

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 {

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};

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<u64> },
+
+    /// Transfer progress update with running count of transferred bytes.
+    Progress {
+        bytes_transferred: u64,
+        total_bytes: Option<u64>,
+    },
+
+    /// Transfer has finished successfully.
+    Completed { total_bytes: u64 },
+}
+
+/// A thread-safe callback invoked when transfer progress occurs.
+pub type ProgressCallback = Arc<dyn Fn(ProgressEvent) + Send + Sync + 'static>;
+
+pin_project! {
+    /// A Stream wrapper that tracks transferred bytes and fires progress callbacks.
+    pub struct ProgressStream<S> {
+        #[pin]
+        inner: S,
+        total_bytes: Option<u64>,
+        bytes_transferred: u64,
+        callback: Option<ProgressCallback>,
+        started: bool,
+        completed: bool,
+    }
+}
+
+impl<S> ProgressStream<S> {
+    pub fn new(
+        inner: S,
+        total_bytes: Option<u64>,
+        callback: Option<ProgressCallback>,
+    ) -> Self {
+        Self {
+            inner,
+            total_bytes,
+            bytes_transferred: 0,
+            callback,
+            started: false,
+            completed: false,
+        }
+    }
+}
+
+impl<S, E> Stream for ProgressStream<S>
+where
+    S: Stream<Item = std::result::Result<Bytes, E>>,
+{
+    type Item = std::result::Result<Bytes, E>;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        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::<Bytes, std::io::Error>(chunk1),
+            Ok::<Bytes, std::io::Error>(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 });
+    }
+}

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<S, E>(
+        &self,
+        remote_path: &str,
+        stream: S,
+        content_length: Option<u64>,
+        callback: Option<ProgressCallback>,
+    ) -> Result<u64>
+    where
+        S: Stream<Item = std::result::Result<Bytes, E>> + Send + Sync + 'static,
+        E: Into<Box<dyn std::error::Error + Send + Sync>> + '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<P: AsRef<Path>>(
+        &self,
+        local_path: P,
+        remote_path: &str,
+        callback: Option<ProgressCallback>,
+    ) -> Result<u64> {
+        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<R>(
+        &self,
+        reader: R,
+        remote_path: &str,
+        total_size: Option<u64>,
+        callback: Option<ProgressCallback>,
+    ) -> Result<u64>
+    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,
+            })
+        }
+    }
+}