media: media endpoint PoC

Supported features:
 - Streaming upload
 - Content-addressed storage
 - Metadata
   - MIME type (taken from Content-Type)
   - Length (I could use stat() for this one tho)
   - filename (for Content-Disposition: attachment, WIP)

1 files changed, 94 insertions, 32 deletions

diff --git a/kittybox-rs/src/media/mod.rs b/kittybox-rs/src/media/mod.rs
index 0d26f92..1bf3958 100644
--- a/kittybox-rs/src/media/mod.rs
+++ b/kittybox-rs/src/media/mod.rs
@@ -1,42 +1,104 @@
 use axum::{
-    extract::{Extension, Host, Multipart},
-    response::{IntoResponse, Json, Response},
+    extract::{Extension, Host, multipart::{Multipart, MultipartError}, Path},
+    response::{IntoResponse, Response}, headers::HeaderValue,
 };
-use bytes::buf::Buf;
-use futures_util::StreamExt;
+use crate::{micropub::{MicropubError, ErrorType}, indieauth::User};
 
 pub mod storage;
-use storage::{MediaStore, MediaStoreError};
+use storage::{MediaStore, MediaStoreError, Metadata, ErrorKind};
+pub use storage::file::FileStore;
 
-/*pub fn upload() -> impl Filter<Extract = (impl Reply,), Error = Rejection> + Clone {
-    warp::post()
-        .and(crate::util::require_host())
-        .and(warp::multipart::form().max_length(1024 * 1024 * 150 /*mb*/))
-        .and_then(|host, mut form: FormData| async move {
-            // TODO get rid of the double unwrap() here
-            let file: Part = form.next().await.unwrap().unwrap();
-            log::debug!(
-                "Uploaded: {:?}, type: {:?}",
-                file.filename(),
-                file.content_type()
-            );
+impl From<MultipartError> for MicropubError {
+    fn from(err: MultipartError) -> Self {
+        Self {
+            error: ErrorType::InvalidRequest,
+            error_description: format!("multipart/form-data error: {}", err)
+        }
+    }
+}
+impl From<MediaStoreError> for MicropubError {
+    fn from(err: MediaStoreError) -> Self {
+        Self {
+            error: ErrorType::InternalServerError,
+            error_description: format!("{}", err)
+        }
+    }
+}
 
-            let mut data = file.stream();
-            while let Some(buf) = data.next().await {
-                // TODO save it into a file
-                log::debug!("buffer length: {:?}", buf.map(|b| b.remaining()));
-            }
-            Ok::<_, warp::Rejection>(warp::reply::with_header(
-                warp::reply::with_status("", warp::http::StatusCode::CREATED),
-                "Location",
-                "./awoo.png",
-            ))
-        })
-}*/
+#[tracing::instrument(skip(blobstore))]
 pub async fn upload<S: MediaStore>(
+    mut upload: Multipart,
+    Extension(blobstore): Extension<S>,
+    user: User
+) -> Response {
+    if !user.check_scope("media") {
+        return MicropubError {
+            error: ErrorType::NotAuthorized,
+            error_description: "Interacting with the media storage requires the \"media\" scope.".to_owned()
+        }.into_response();
+    }
+    let host = user.me.host().unwrap().to_string() + &user.me.port().map(|i| format!(":{}", i)).unwrap_or_default();
+    let field = match upload.next_field().await {
+        Ok(Some(field)) => field,
+        Ok(None) => {
+            return MicropubError {
+                error: ErrorType::InvalidRequest,
+                error_description: "Send multipart/form-data with one field named file".to_owned()
+            }.into_response();
+        },
+        Err(err) => {
+            return MicropubError::from(err).into_response();
+        },
+    };
+    let metadata: Metadata = (&field).into();
+    match blobstore.write_streaming(&host, metadata, field).await {
+        Ok(filename) => IntoResponse::into_response((
+            axum::http::StatusCode::CREATED,
+            [
+                ("Location", user.me.join(
+                    &format!(".kittybox/media/uploads/{}", filename)
+                ).unwrap().as_str())
+            ]
+        )),
+        Err(err) => MicropubError::from(err).into_response()
+    }
+}
+
+#[tracing::instrument(skip(blobstore))]
+pub async fn serve<S: MediaStore>(
     Host(host): Host,
-    upload: Multipart,
-    Extension(db): Extension<S>,
+    Path(path): Path<String>,
+    Extension(blobstore): Extension<S>
 ) -> Response {
-    todo!()
+    use axum::http::StatusCode;
+    tracing::debug!("Searching for file...");
+    match blobstore.read_streaming(&host, path.as_str()).await {
+        Ok((metadata, stream)) => {
+            tracing::debug!("Metadata: {:?}", metadata);
+            let mut r = Response::builder();
+            {
+                let headers = r.headers_mut().unwrap();
+                headers.insert(
+                    "Content-Type",
+                    HeaderValue::from_str(&metadata.content_type).unwrap()
+                );
+                if let Some(length) = metadata.length {
+                    headers.insert(
+                        "Content-Length",
+                        HeaderValue::from_str(&length.to_string()).unwrap()
+                    );
+                }
+            }
+            r.body(axum::body::StreamBody::new(stream)).unwrap().into_response()
+        },
+        Err(err) => match err.kind() {
+            ErrorKind::NotFound => {
+                IntoResponse::into_response(StatusCode::NOT_FOUND)
+            },
+            _ => {
+                tracing::error!("{}", err);
+                IntoResponse::into_response(StatusCode::INTERNAL_SERVER_ERROR)
+            }
+        }
+    }
 }