nextcloud_uploads/
client.rs

1//! Asynchronous client for Nextcloud's WebDAV Chunked Upload v2 API.
2
3use std::fmt;
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::time::{SystemTime, UNIX_EPOCH};
6
7pub use nextcloud_dav::DESTINATION_HEADER;
8use nextcloud_sdk_core::{
9    ApiPath, AuthProvider, Error, Headers, HttpMethod, Request, Response, StatusError, Transport,
10};
11
12use crate::path::{
13    abort_path, assembly_path, chunk_path, destination_header_value, upload_collection_path,
14};
15use crate::session::UploadSession;
16
17#[cfg(test)]
18mod tests;
19
20/// Header conveying total file size during chunked upload and assembly.
21pub const TOTAL_LENGTH_HEADER: &str = "OC-Total-Length";
22
23/// Header conveying target modification time (in unix seconds) during chunk assembly.
24pub const MTIME_HEADER: &str = "X-OC-Mtime";
25
26static UPLOAD_COUNTER: AtomicU64 = AtomicU64::new(1);
27
28/// Generates a unique upload session token from system timestamp and an atomic counter.
29fn generate_upload_id() -> String {
30    let now = SystemTime::now()
31        .duration_since(UNIX_EPOCH)
32        .unwrap_or_default()
33        .as_millis();
34    let count = UPLOAD_COUNTER.fetch_add(1, Ordering::Relaxed);
35    format!("chunked-upload-{now}-{count}")
36}
37
38/// Client for Nextcloud's WebDAV Chunked Upload v2 protocol.
39///
40/// Manages creating upload directories, streaming individual chunks, assembling the
41/// uploaded chunks into the destination file, or aborting the upload.
42///
43/// Generic over the [`Transport`] and [`AuthProvider`], held by reference so that
44/// connection pools and credentials can be shared across tasks.
45#[derive(Clone, Copy, Debug)]
46pub struct UploadClient<'a, T: Transport, A: AuthProvider> {
47    transport: &'a T,
48    auth: &'a A,
49}
50
51impl<'a, T: Transport, A: AuthProvider> UploadClient<'a, T, A> {
52    /// Binds an upload client to a transport and credential provider.
53    pub const fn new(transport: &'a T, auth: &'a A) -> Self {
54        Self { transport, auth }
55    }
56
57    /// The transport this client sends requests through.
58    pub const fn transport(&self) -> &'a T {
59        self.transport
60    }
61
62    /// The credential provider applied to every request.
63    pub const fn auth(&self) -> &'a A {
64        self.auth
65    }
66
67    /// Initializes a new chunked upload session by sending an `MKCOL` request with an auto-generated upload ID.
68    ///
69    /// Sends `MKCOL /remote.php/dav/uploads/{user_id}/{upload_id}` with the `Destination` header set.
70    pub async fn start(
71        &self,
72        user_id: &str,
73        destination_path: &str,
74        total_size: u64,
75    ) -> Result<UploadSession, Error> {
76        let upload_id = generate_upload_id();
77        self.start_with_id(user_id, destination_path, total_size, &upload_id)
78            .await
79    }
80
81    /// Initializes a new chunked upload session using an explicit `upload_id`.
82    ///
83    /// Sends `MKCOL /remote.php/dav/uploads/{user_id}/{upload_id}` with the `Destination` header set.
84    pub async fn start_with_id(
85        &self,
86        user_id: &str,
87        destination_path: &str,
88        total_size: u64,
89        upload_id: &str,
90    ) -> Result<UploadSession, Error> {
91        let destination = destination_header_value(user_id, destination_path)?;
92        let path = upload_collection_path(user_id, upload_id)?;
93
94        let mut headers = self.authenticated_headers();
95        headers.set(DESTINATION_HEADER, &destination);
96
97        let request = Request::new(HttpMethod::Mkcol, path.clone()).with_headers(headers);
98        let response = self.transport.execute(request).await?;
99
100        ensure_success(HttpMethod::Mkcol, &path, &response)?;
101
102        Ok(UploadSession::new(
103            user_id,
104            upload_id,
105            destination,
106            total_size,
107        ))
108    }
109
110    /// Uploads an individual chunk into the upload collection.
111    ///
112    /// Sends `PUT /remote.php/dav/uploads/{user_id}/{upload_id}/{chunk_number}` with:
113    /// - `Destination: {destination_path}`
114    /// - `OC-Total-Length: {total_size}`
115    /// - payload containing `chunk_data`
116    ///
117    /// Automatically marks the chunk index as uploaded in the session upon success.
118    pub async fn upload_chunk(
119        &self,
120        session: &UploadSession,
121        chunk_index: u32,
122        chunk_data: Vec<u8>,
123    ) -> Result<(), Error> {
124        let path = chunk_path(session.user_id(), session.upload_id(), chunk_index)?;
125
126        let mut headers = self.authenticated_headers();
127        headers.set(DESTINATION_HEADER, session.destination_path());
128        headers.set(TOTAL_LENGTH_HEADER, session.total_size().to_string());
129
130        let request = Request::new(HttpMethod::Put, path.clone())
131            .with_headers(headers)
132            .body_bytes(chunk_data);
133
134        let response = self.transport.execute(request).await?;
135        ensure_success(HttpMethod::Put, &path, &response)?;
136
137        session.mark_chunk_uploaded(chunk_index);
138        Ok(())
139    }
140
141    /// Finalizes the upload by moving `.file` to the destination path.
142    ///
143    /// Sends `MOVE /remote.php/dav/uploads/{user_id}/{upload_id}/.file` with:
144    /// - `Destination: {destination_path}`
145    /// - `OC-Total-Length: {total_size}`
146    /// - optionally `X-OC-Mtime: {mtime}`
147    pub async fn finish(&self, session: &UploadSession, mtime: Option<u64>) -> Result<(), Error> {
148        let path = assembly_path(session.user_id(), session.upload_id())?;
149
150        let mut headers = self.authenticated_headers();
151        headers.set(DESTINATION_HEADER, session.destination_path());
152        headers.set(TOTAL_LENGTH_HEADER, session.total_size().to_string());
153        if let Some(mtime) = mtime {
154            headers.set(MTIME_HEADER, mtime.to_string());
155        }
156
157        let request = Request::new(HttpMethod::Move, path.clone()).with_headers(headers);
158        let response = self.transport.execute(request).await?;
159
160        ensure_success(HttpMethod::Move, &path, &response)
161    }
162
163    /// Cancels and removes the temporary upload collection.
164    ///
165    /// Sends `DELETE /remote.php/dav/uploads/{user_id}/{upload_id}/`.
166    pub async fn abort(&self, session: &UploadSession) -> Result<(), Error> {
167        let path = abort_path(session.user_id(), session.upload_id())?;
168
169        let headers = self.authenticated_headers();
170        let request = Request::new(HttpMethod::Delete, path.clone()).with_headers(headers);
171        let response = self.transport.execute(request).await?;
172
173        ensure_success(HttpMethod::Delete, &path, &response)
174    }
175
176    fn authenticated_headers(&self) -> Headers {
177        let mut headers = Headers::new();
178        self.auth.authenticate(&mut headers);
179        headers
180    }
181}
182
183fn ensure_success(method: HttpMethod, path: &ApiPath, response: &Response) -> Result<(), Error> {
184    if !response.status().is_success() {
185        return Err(Error::Status(StatusError::new(
186            method.as_str(),
187            path.clone(),
188            response.status(),
189            response.body(),
190        )));
191    }
192    Ok(())
193}
194
195impl<T: Transport, A: AuthProvider> fmt::Display for UploadClient<'_, T, A> {
196    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
197        f.write_str("UploadClient")
198    }
199}
nextcloud_uploads/client.rs

nextcloud_uploads/
client.rs
