From e23b33e24541deb02845e9d6460388ed41928584 Mon Sep 17 00:00:00 2001 From: asonix Date: Sat, 15 Jul 2023 13:06:34 -0500 Subject: [PATCH] Enable setting content type when uploading objects --- src/backgrounded.rs | 4 ++- src/ffmpeg.rs | 7 +++++ src/generate.rs | 12 +++++-- src/ingest.rs | 4 ++- src/lib.rs | 54 +++++++++++++++++++++++++++----- src/magick.rs | 13 ++++++++ src/store.rs | 66 ++++++++++++++++++++++++++++++--------- src/store/file_store.rs | 16 ++++++++-- src/store/object_store.rs | 41 +++++++++++++++++------- 9 files changed, 176 insertions(+), 41 deletions(-) diff --git a/src/backgrounded.rs b/src/backgrounded.rs index 1028e82..86ae39e 100644 --- a/src/backgrounded.rs +++ b/src/backgrounded.rs @@ -5,6 +5,7 @@ use crate::{ }; use actix_web::web::Bytes; use futures_util::{Stream, TryStreamExt}; +use mime::APPLICATION_OCTET_STREAM; use tracing::{Instrument, Span}; pub(crate) struct Backgrounded @@ -58,7 +59,8 @@ where let stream = stream.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e)); - let identifier = store.save_stream(stream).await?; + // use octet-stream, we don't know the upload's real type yet + let identifier = store.save_stream(stream, APPLICATION_OCTET_STREAM).await?; self.identifier = Some(identifier); diff --git a/src/ffmpeg.rs b/src/ffmpeg.rs index f75926a..85e364b 100644 --- a/src/ffmpeg.rs +++ b/src/ffmpeg.rs @@ -353,6 +353,13 @@ impl ThumbnailFormat { // Self::Webp => "webp", } } + + pub(crate) fn media_type(self) -> mime::Mime { + match self { + Self::Jpeg => mime::IMAGE_JPEG, + // Self::Webp => image_webp(), + } + } } impl OutputFormat { diff --git a/src/generate.rs b/src/generate.rs index c4db291..5e88f67 100644 --- a/src/generate.rs +++ b/src/generate.rs @@ -68,14 +68,18 @@ async fn process( return Err(UploadError::MissingIdentifier.into()); }; + let thumbnail_format = thumbnail_format.unwrap_or(ThumbnailFormat::Jpeg); + let reader = crate::ffmpeg::thumbnail( store.clone(), identifier, input_format.unwrap_or(VideoFormat::Mp4), - thumbnail_format.unwrap_or(ThumbnailFormat::Jpeg), + thumbnail_format, ) .await?; - let motion_identifier = store.save_async_read(reader).await?; + let motion_identifier = store + .save_async_read(reader, thumbnail_format.media_type()) + .await?; repo.relate_motion_identifier(hash.clone(), &motion_identifier) .await?; @@ -97,7 +101,9 @@ async fn process( let details = Details::from_bytes(bytes.clone(), format.as_hint()).await?; - let identifier = store.save_bytes(bytes.clone()).await?; + let identifier = store + .save_bytes(bytes.clone(), details.content_type()) + .await?; repo.relate_details(&identifier, &details).await?; repo.relate_variant_identifier( hash, diff --git a/src/ingest.rs b/src/ingest.rs index fd53e27..564bda9 100644 --- a/src/ingest.rs +++ b/src/ingest.rs @@ -79,7 +79,9 @@ where let hasher_reader = Hasher::new(processed_reader, Sha256::new()); let hasher = hasher_reader.hasher(); - let identifier = store.save_async_read(hasher_reader).await?; + let identifier = store + .save_async_read(hasher_reader, input_type.content_type()) + .await?; drop(permit); diff --git a/src/lib.rs b/src/lib.rs index 26c03b1..992718c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1629,7 +1629,7 @@ where .await? { if repo.get(STORE_MIGRATION_MOTION).await?.is_none() { - match migrate_file(&from, &to, &identifier, skip_missing_files).await { + match migrate_file(repo, &from, &to, &identifier, skip_missing_files).await { Ok(new_identifier) => { migrate_details(repo, identifier, &new_identifier).await?; repo.relate_motion_identifier( @@ -1640,6 +1640,13 @@ where repo.set(STORE_MIGRATION_MOTION, b"1".to_vec().into()) .await?; } + Err(MigrateError::Details(e)) => { + tracing::warn!( + "Error fetching details for motion file for hash {}", + hex::encode(&hash) + ); + return Err(e.into()); + } Err(MigrateError::From(e)) if e.is_not_found() && skip_missing_files => { tracing::warn!("Skipping motion file for hash {}", hex::encode(&hash)); } @@ -1665,7 +1672,7 @@ where continue; } - match migrate_file(&from, &to, &identifier, skip_missing_files).await { + match migrate_file(repo, &from, &to, &identifier, skip_missing_files).await { Ok(new_identifier) => { migrate_details(repo, identifier, &new_identifier).await?; repo.remove_variant(hash.as_ref().to_vec().into(), variant.clone()) @@ -1680,6 +1687,13 @@ where repo.set(STORE_MIGRATION_VARIANT, new_identifier.to_bytes()?.into()) .await?; } + Err(MigrateError::Details(e)) => { + tracing::warn!( + "Error fetching details for motion file for hash {}", + hex::encode(&hash) + ); + return Err(e.into()); + } Err(MigrateError::From(e)) if e.is_not_found() && skip_missing_files => { tracing::warn!( "Skipping variant {} for hash {}", @@ -1698,12 +1712,19 @@ where } } - match migrate_file(&from, &to, &original_identifier, skip_missing_files).await { + match migrate_file(repo, &from, &to, &original_identifier, skip_missing_files).await { Ok(new_identifier) => { migrate_details(repo, original_identifier, &new_identifier).await?; repo.relate_identifier(hash.as_ref().to_vec().into(), &new_identifier) .await?; } + Err(MigrateError::Details(e)) => { + tracing::warn!( + "Error fetching details for motion file for hash {}", + hex::encode(&hash) + ); + return Err(e.into()); + } Err(MigrateError::From(e)) if e.is_not_found() && skip_missing_files => { tracing::warn!("Skipping original file for hash {}", hex::encode(&hash)); } @@ -1746,20 +1767,22 @@ where Ok(()) } -async fn migrate_file( +async fn migrate_file( + repo: &R, from: &S1, to: &S2, identifier: &S1::Identifier, skip_missing_files: bool, ) -> Result where + R: IdentifierRepo, S1: Store, S2: Store, { let mut failure_count = 0; loop { - match do_migrate_file(from, to, identifier).await { + match do_migrate_file(repo, from, to, identifier).await { Ok(identifier) => return Ok(identifier), Err(MigrateError::From(e)) if e.is_not_found() && skip_missing_files => { return Err(MigrateError::From(e)); @@ -1783,15 +1806,18 @@ where #[derive(Debug)] enum MigrateError { From(crate::store::StoreError), + Details(crate::store::StoreError), To(crate::store::StoreError), } -async fn do_migrate_file( +async fn do_migrate_file( + repo: &R, from: &S1, to: &S2, identifier: &S1::Identifier, ) -> Result where + R: IdentifierRepo, S1: Store, S2: Store, { @@ -1800,7 +1826,21 @@ where .await .map_err(MigrateError::From)?; - let new_identifier = to.save_stream(stream).await.map_err(MigrateError::To)?; + let details_opt = repo + .details(identifier) + .await + .map_err(MigrateError::Details)?; + + let content_type = if let Some(details) = details_opt { + details.content_type() + } else { + mime::APPLICATION_OCTET_STREAM + }; + + let new_identifier = to + .save_stream(stream, content_type) + .await + .map_err(MigrateError::To)?; Ok(new_identifier) } diff --git a/src/magick.rs b/src/magick.rs index cb941d0..650c729 100644 --- a/src/magick.rs +++ b/src/magick.rs @@ -179,6 +179,19 @@ impl ValidInputType { _ => None, } } + + pub(crate) fn content_type(self) -> mime::Mime { + match self { + Self::Avif => image_avif(), + Self::Gif => mime::IMAGE_GIF, + Self::Jpeg => mime::IMAGE_JPEG, + Self::Jxl => image_jxl(), + Self::Mp4 => video_mp4(), + Self::Png => mime::IMAGE_PNG, + Self::Webm => video_webm(), + Self::Webp => image_webp(), + } + } } #[derive(Debug, PartialEq, Eq)] diff --git a/src/store.rs b/src/store.rs index 180c1af..7ab8b3e 100644 --- a/src/store.rs +++ b/src/store.rs @@ -72,15 +72,27 @@ pub(crate) trait Store: Clone + Debug { async fn health_check(&self) -> Result<(), StoreError>; - async fn save_async_read(&self, reader: Reader) -> Result + async fn save_async_read( + &self, + reader: Reader, + content_type: mime::Mime, + ) -> Result where Reader: AsyncRead + Unpin + 'static; - async fn save_stream(&self, stream: S) -> Result + async fn save_stream( + &self, + stream: S, + content_type: mime::Mime, + ) -> Result where S: Stream> + Unpin + 'static; - async fn save_bytes(&self, bytes: Bytes) -> Result; + async fn save_bytes( + &self, + bytes: Bytes, + content_type: mime::Mime, + ) -> Result; async fn to_stream( &self, @@ -114,22 +126,34 @@ where T::health_check(self).await } - async fn save_async_read(&self, reader: Reader) -> Result + async fn save_async_read( + &self, + reader: Reader, + content_type: mime::Mime, + ) -> Result where Reader: AsyncRead + Unpin + 'static, { - T::save_async_read(self, reader).await + T::save_async_read(self, reader, content_type).await } - async fn save_stream(&self, stream: S) -> Result + async fn save_stream( + &self, + stream: S, + content_type: mime::Mime, + ) -> Result where S: Stream> + Unpin + 'static, { - T::save_stream(self, stream).await + T::save_stream(self, stream, content_type).await } - async fn save_bytes(&self, bytes: Bytes) -> Result { - T::save_bytes(self, bytes).await + async fn save_bytes( + &self, + bytes: Bytes, + content_type: mime::Mime, + ) -> Result { + T::save_bytes(self, bytes, content_type).await } async fn to_stream( @@ -173,22 +197,34 @@ where T::health_check(self).await } - async fn save_async_read(&self, reader: Reader) -> Result + async fn save_async_read( + &self, + reader: Reader, + content_type: mime::Mime, + ) -> Result where Reader: AsyncRead + Unpin + 'static, { - T::save_async_read(self, reader).await + T::save_async_read(self, reader, content_type).await } - async fn save_stream(&self, stream: S) -> Result + async fn save_stream( + &self, + stream: S, + content_type: mime::Mime, + ) -> Result where S: Stream> + Unpin + 'static, { - T::save_stream(self, stream).await + T::save_stream(self, stream, content_type).await } - async fn save_bytes(&self, bytes: Bytes) -> Result { - T::save_bytes(self, bytes).await + async fn save_bytes( + &self, + bytes: Bytes, + content_type: mime::Mime, + ) -> Result { + T::save_bytes(self, bytes, content_type).await } async fn to_stream( diff --git a/src/store/file_store.rs b/src/store/file_store.rs index 90e07a0..7b40a00 100644 --- a/src/store/file_store.rs +++ b/src/store/file_store.rs @@ -66,6 +66,7 @@ impl Store for FileStore { async fn save_async_read( &self, mut reader: Reader, + _content_type: mime::Mime, ) -> Result where Reader: AsyncRead + Unpin + 'static, @@ -80,15 +81,24 @@ impl Store for FileStore { Ok(self.file_id_from_path(path)?) } - async fn save_stream(&self, stream: S) -> Result + async fn save_stream( + &self, + stream: S, + content_type: mime::Mime, + ) -> Result where S: Stream> + Unpin + 'static, { - self.save_async_read(StreamReader::new(stream)).await + self.save_async_read(StreamReader::new(stream), content_type) + .await } #[tracing::instrument(skip(bytes))] - async fn save_bytes(&self, bytes: Bytes) -> Result { + async fn save_bytes( + &self, + bytes: Bytes, + _content_type: mime::Mime, + ) -> Result { let path = self.next_file().await?; if let Err(e) = self.safe_save_bytes(&path, bytes).await { diff --git a/src/store/object_store.rs b/src/store/object_store.rs index bf4c686..73c3d55 100644 --- a/src/store/object_store.rs +++ b/src/store/object_store.rs @@ -190,15 +190,24 @@ impl Store for ObjectStore { Ok(()) } - async fn save_async_read(&self, reader: Reader) -> Result + async fn save_async_read( + &self, + reader: Reader, + content_type: mime::Mime, + ) -> Result where Reader: AsyncRead + Unpin + 'static, { - self.save_stream(ReaderStream::new(reader)).await + self.save_stream(ReaderStream::new(reader), content_type) + .await } #[tracing::instrument(skip_all)] - async fn save_stream(&self, mut stream: S) -> Result + async fn save_stream( + &self, + mut stream: S, + content_type: mime::Mime, + ) -> Result where S: Stream> + Unpin + 'static, { @@ -206,7 +215,7 @@ impl Store for ObjectStore { if first_chunk.len() < CHUNK_SIZE { drop(stream); - let (req, object_id) = self.put_object_request().await?; + let (req, object_id) = self.put_object_request(content_type).await?; let response = req .send_body(first_chunk) .await @@ -221,7 +230,7 @@ impl Store for ObjectStore { let mut first_chunk = Some(first_chunk); - let (req, object_id) = self.create_multipart_request().await?; + let (req, object_id) = self.create_multipart_request(content_type).await?; let mut response = req.send().await.map_err(ObjectError::from)?; if !response.status().is_success() { @@ -329,8 +338,12 @@ impl Store for ObjectStore { } #[tracing::instrument(skip_all)] - async fn save_bytes(&self, bytes: Bytes) -> Result { - let (req, object_id) = self.put_object_request().await?; + async fn save_bytes( + &self, + bytes: Bytes, + content_type: mime::Mime, + ) -> Result { + let (req, object_id) = self.put_object_request(content_type).await?; let response = req.send_body(bytes).await.map_err(ObjectError::from)?; @@ -470,19 +483,25 @@ impl ObjectStore { Ok(self.build_request(action)) } - async fn put_object_request(&self) -> Result<(ClientRequest, ObjectId), StoreError> { + async fn put_object_request( + &self, + content_type: mime::Mime, + ) -> Result<(ClientRequest, ObjectId), StoreError> { let path = self.next_file().await?; let mut action = self.bucket.put_object(Some(&self.credentials), &path); action .headers_mut() - .insert("content-type", "application/octet-stream"); + .insert("content-type", content_type.as_ref()); Ok((self.build_request(action), ObjectId::from_string(path))) } - async fn create_multipart_request(&self) -> Result<(ClientRequest, ObjectId), StoreError> { + async fn create_multipart_request( + &self, + content_type: mime::Mime, + ) -> Result<(ClientRequest, ObjectId), StoreError> { let path = self.next_file().await?; let mut action = self @@ -491,7 +510,7 @@ impl ObjectStore { action .headers_mut() - .insert("content-type", "application/octet-stream"); + .insert("content-type", content_type.as_ref()); Ok((self.build_request(action), ObjectId::from_string(path))) }