bobashare_web/api/v1/
upload.rs

1//! API to create an upload
2
3use anyhow::Context;
4use axum::{
5    body::Body,
6    extract::{Path, State},
7    response::{IntoResponse, Response},
8    Json,
9};
10use axum_extra::{extract::WithRejection, typed_header::TypedHeaderRejection, TypedHeader};
11use bobashare::{generate_randomized_id, storage::file::CreateUploadError};
12use chrono::{DateTime, TimeDelta, Utc};
13use displaydoc::Display;
14use futures_util::TryStreamExt;
15use headers::{ContentLength, ContentType};
16use hyper::{header, HeaderMap, StatusCode};
17use serde::Serialize;
18use thiserror::Error;
19use tokio::io::{AsyncWriteExt, BufWriter};
20use tracing::{event, instrument, Level};
21
22use super::ApiErrorExt;
23use crate::{clamp_expiry, str_to_duration, AppState};
24
25/// The JSON API response after uploading a file
26#[derive(Debug, Clone, Serialize)]
27pub struct UploadResponse {
28    /// ID of the upload (used in URL)
29    pub id: String,
30    /// url to the upload
31    pub url: String,
32    /// direct url to download the raw uploaded file
33    pub direct_url: String,
34    /// the name of the file
35    pub filename: String,
36    /// the MIME type of the uploaded file
37    pub mimetype: String,
38    /// expiration date in RFC 3339 format, null if the upload never expires
39    pub expiry_date: Option<DateTime<Utc>>,
40    /// key to delete the upload later before it's expired
41    pub delete_key: String,
42}
43
44/// Errors that could occur during upload
45#[derive(Debug, Error, Display)]
46pub enum UploadError {
47    /// an upload already exists with the same id
48    AlreadyExists,
49    /// error parsing `{name}` header
50    ParseHeader { name: String, source: anyhow::Error },
51    /// file is too large ({size} > {max})
52    TooLarge { size: u64, max: u64 },
53
54    /// upload was cancelled
55    Cancelled(#[source] anyhow::Error),
56
57    /// internal server error
58    InternalServer(#[from] anyhow::Error),
59}
60impl From<TypedHeaderRejection> for UploadError {
61    fn from(rej: TypedHeaderRejection) -> Self {
62        Self::ParseHeader {
63            name: rej.name().to_string(),
64            source: rej.into(),
65        }
66    }
67}
68impl IntoResponse for UploadError {
69    fn into_response(self) -> Response {
70        let code = match self {
71            Self::AlreadyExists => StatusCode::CONFLICT,
72            Self::ParseHeader { name: _, source: _ } => StatusCode::BAD_REQUEST,
73            Self::TooLarge { size: _, max: _ } => StatusCode::PAYLOAD_TOO_LARGE,
74            Self::Cancelled(_) => StatusCode::INTERNAL_SERVER_ERROR, // unused
75            Self::InternalServer(_) => StatusCode::INTERNAL_SERVER_ERROR,
76        };
77
78        if let Self::Cancelled(_) = self {
79            let error = anyhow::Error::new(self);
80            event!(
81                Level::INFO,
82                error = format!("{error:#}"),
83                "returning empty response to cancelled upload"
84            );
85            ().into_response()
86        } else {
87            self.into_response_with_code(code)
88        }
89    }
90}
91
92/// Create an upload
93///
94/// # Request
95///
96/// `PUT /api/v1/upload/:filename`
97///
98/// NOTE: The first URL will use the randomized upload ID as the filename.
99///
100/// ## Headers
101///
102/// - `Content-Type` (required) -- mimetype -- the mime type (file format) of
103///   the file.
104/// - `Bobashare-Expiry` (optional) -- number -- duration until the upload
105///   should expire
106///   - specify `0` for no expiry
107///   - examples (see [`str_to_duration`] for more information):
108///     - `1d` -- 1 day
109///     - `1h` -- 1 hour
110///     - `1m` -- 1 minute
111///     - `1s` -- 1 second
112///
113/// - `Bobashare-Delete-Key` (optional) -- string -- custom key to use for
114///   deleting the file later; if not provided, one will be randomly generated
115///
116/// ## Body
117///
118/// Should contain the contents of the file to upload
119///
120/// # Response
121///
122/// ## Success
123///
124/// - 201 Created
125/// - `Location` header containing the URL of the upload
126/// - JSON body created from [`UploadResponse`]
127#[instrument(skip(state, filename, headers, body), fields(id))]
128pub async fn put(
129    state: State<&'static AppState>,
130    filename: Path<String>,
131    WithRejection(TypedHeader(mimetype), _): WithRejection<TypedHeader<ContentType>, UploadError>,
132    WithRejection(TypedHeader(content_length), _): WithRejection<
133        TypedHeader<ContentLength>,
134        UploadError,
135    >,
136    headers: HeaderMap,
137    body: Body,
138) -> Result<impl IntoResponse, UploadError> {
139    // hyper will automatically make sure the body is <= the content-length, so
140    // we can rely on it here
141    //
142    // also note that hyper seems to intercept the Content-Length header and
143    // return its own empty response instead of using WithRejection here
144    if content_length.0 > state.max_file_size {
145        event!(
146            Level::INFO,
147            size = content_length.0,
148            max = state.max_file_size,
149            "file is too large"
150        );
151        return Err(UploadError::TooLarge {
152            size: content_length.0,
153            max: state.max_file_size,
154        });
155    }
156    let id = generate_randomized_id(state.id_length);
157    tracing::Span::current().record("id", &id);
158    event!(Level::DEBUG, "generated random ID for upload");
159
160    let mimetype = mimetype.into();
161
162    let expiry = match headers.get("Bobashare-Expiry") {
163        // if header not found, use default expiry
164        None => {
165            event!(
166                Level::DEBUG,
167                "`Bobashare-Expiry` header not provided, using default"
168            );
169            Some(state.default_expiry)
170        }
171        // otherwise, clamp the requested expiry to the max
172        Some(e) => {
173            let expiry = e.to_str().map_err(|e| UploadError::ParseHeader {
174                name: String::from("Bobashare-Expiry"),
175                source: anyhow::Error::new(e).context("error converting to string"),
176            })?;
177
178            event!(Level::DEBUG, "`Bobashare-Expiry` header says {}", expiry);
179
180            let expiry = if expiry == "never" {
181                None
182            } else {
183                Some(
184                    TimeDelta::from_std(str_to_duration(expiry).map_err(|e| {
185                        UploadError::ParseHeader {
186                            name: String::from("Bobashare-Expiry"),
187                            source: anyhow::Error::new(e).context("error parsing duration string"),
188                        }
189                    })?)
190                    .map_err(|e| UploadError::ParseHeader {
191                        name: String::from("Bobashare-Expiry"),
192                        source: anyhow::Error::new(e).context("error converting duration"),
193                    })?,
194                )
195            };
196
197            // TODO: should we return an error if expiry is too large instead?
198            clamp_expiry(state.max_expiry, expiry)
199        }
200    };
201    event!(Level::DEBUG, expiry = %expiry.map_or_else(|| String::from("never"), |e| e.to_string()));
202
203    let delete_key = headers
204        .get("Bobashare-Delete-Key")
205        .map(|k| {
206            k.to_str().map_err(|e| UploadError::ParseHeader {
207                name: String::from("Bobashare-Delete-Key"),
208                source: anyhow::Error::new(e).context("error converting to string"),
209            })
210        })
211        .transpose()?
212        .map(ToString::to_string);
213    if delete_key.is_some() {
214        event!(Level::DEBUG, delete_key, "custom delete key was provided");
215    } else {
216        event!(Level::DEBUG, "delete_key will be randomly generated");
217    }
218
219    let mut upload = state
220        .backend
221        .create_upload(&id, &filename, mimetype, expiry, delete_key)
222        .await
223        .map_err(|e| {
224            if let CreateUploadError::AlreadyExists = e {
225                UploadError::AlreadyExists
226            } else {
227                UploadError::InternalServer(
228                    anyhow::Error::new(e).context("error while initializing upload"),
229                )
230            }
231        })?;
232    event!(
233        Level::TRACE,
234        upload = format!("{upload:?}"),
235        "created upload handle"
236    );
237
238    let mut file_writer = BufWriter::new(&mut upload.file);
239    event!(Level::DEBUG, "streaming file to disk");
240    let stream_file_task = async {
241        let mut body = body.into_data_stream();
242        loop {
243            let chunk = body.try_next().await.context("error reading body");
244            match chunk {
245                Ok(ch) => match ch {
246                    Some(c) => {
247                        event!(
248                            Level::TRACE,
249                            "writing chunk of {} bytes to file buffer",
250                            c.len()
251                        );
252                        file_writer
253                            .write_all(&c)
254                            .await
255                            .context("error writing to file")?;
256                    }
257                    None => break,
258                },
259                Err(e) => {
260                    return Err(UploadError::Cancelled(e));
261                }
262            }
263        }
264        Ok(())
265    };
266
267    let mut shutdown_rx = state.shutdown_tx.subscribe();
268    tokio::select! {
269        res = stream_file_task => {
270            if let Err(e) = res {
271                event!(Level::INFO, "upload was cancelled; it will be deleted");
272                upload
273                    .flush()
274                    .await
275                    .context("error flushing cancelled upload before deletion")?;
276                state
277                    .backend
278                    .delete_upload(id)
279                    .await
280                    .context("error deleting cancelled upload")?;
281                event!(Level::INFO, "upload was deleted successfully");
282                return Err(e);
283            }
284        },
285        _ = shutdown_rx.recv() => {
286            event!(Level::INFO, "server is shutting down; deleting lock");
287            upload.drop_lock().await.context("error deleting lock of cancelled upload")?;
288            return Err(UploadError::InternalServer(anyhow::anyhow!("server is shutting down")));
289        }
290    };
291
292    file_writer
293        .flush()
294        .await
295        .context("error flushing file buffer")?;
296
297    let metadata = upload
298        .flush()
299        .await
300        .context("error flushing upload metadata to disk")?;
301    event!(Level::DEBUG, "flushed upload metadata to disk");
302
303    // SAFETY: this shouldn't fail because `metadata.id` should be valid in a
304    // URL
305    let url = state.base_url.join(&metadata.id).unwrap().to_string();
306    let direct_url = state.raw_url.join(&metadata.id).unwrap().to_string();
307    event!(
308        Level::INFO,
309        url,
310        filename = metadata.filename,
311        mimetype = %metadata.mimetype,
312        expiry = %metadata
313            .expiry_date
314            .map_or_else(|| String::from("never"), |e| e.to_string()),
315        "successfully created upload"
316    );
317    Ok((
318        StatusCode::CREATED,
319        [
320            (header::CONTENT_LOCATION, direct_url.clone()),
321            (header::LOCATION, url.clone()),
322        ],
323        Json(UploadResponse {
324            id,
325            url,
326            direct_url,
327            filename: metadata.filename,
328            mimetype: metadata.mimetype.to_string(),
329            expiry_date: metadata.expiry_date,
330            delete_key: metadata.delete_key,
331        }),
332    ))
333}
bobashare_web/api/v1/upload.rs

bobashare_web/api/v1/
upload.rs
