feat(platform/image-processor): allow http downloads on image processor

common/src/task.rs

@@ -186,7 +186,8 @@ where
 		set_abort(abort);
 
 		// The reason this is allowed to be asserted is because we're catching the panic
-		// and returning None instead of propagating it, if the panic was caused by a task abort.
+		// and returning None instead of propagating it, if the panic was caused by a
+		// task abort.
 		match std::panic::catch_unwind(std::panic::AssertUnwindSafe(f)) {
 			Ok(r) => Some(r),
 			Err(err) => {

platform/image_processor/Cargo.toml

@@ -34,6 +34,7 @@ gifski = "1.13"
 png = "0.17"
 num_cpus = "1.16"
 bytes = "1.0"
+reqwest = { version = "0.11", default-features = false, features = ["rustls-tls"] }
 
 common = { workspace = true, features = ["default"] }
 config = { workspace = true }

platform/image_processor/src/config.rs

@@ -15,6 +15,9 @@ pub struct ImageProcessorConfig {
 
 	/// Instance ID (defaults to a random ULID)
 	pub instance_id: Ulid,
+
+	/// Allow http downloads
+	pub allow_http: bool,
 }
 
 impl Default for ImageProcessorConfig {
@@ -24,6 +27,7 @@ impl Default for ImageProcessorConfig {
 			target_bucket: S3BucketConfig::default(),
 			concurrency: num_cpus::get(),
 			instance_id: Ulid::new(),
+			allow_http: true,
 		}
 	}
 }

platform/image_processor/src/processor/error.rs

@@ -70,6 +70,12 @@ pub enum ProcessorError {
 
 	#[error("gifski encode: {0}")]
 	GifskiEncode(anyhow::Error),
+
+	#[error("http download disabled")]
+	HttpDownloadDisabled,
+
+	#[error("http download: {0}")]
+	HttpDownload(#[from] reqwest::Error),
 }
 
 pub type Result<T, E = ProcessorError> = std::result::Result<T, E>;

platform/image_processor/src/processor/job/decoder/ffmpeg.rs

@@ -41,7 +41,8 @@ impl<'data> FfmpegDecoder<'data> {
 		let input_stream_duration = input_stream.duration().unwrap_or(0);
 		let input_stream_frames = input_stream
 			.nb_frames()
-			.ok_or_else(|| ProcessorError::FfmpegDecode(anyhow!("no frame count")))?.max(1);
+			.ok_or_else(|| ProcessorError::FfmpegDecode(anyhow!("no frame count")))?
+			.max(1);
 
 		if input_stream_time_base.den == 0 || input_stream_time_base.num == 0 {
 			return Err(ProcessorError::FfmpegDecode(anyhow!("stream time base is 0")));

platform/image_processor/src/processor/job/mod.rs

@@ -72,23 +72,39 @@ pub async fn handle_job(
 
 impl<'a, G: ImageProcessorGlobal> Job<'a, G> {
 	async fn download_source(&self) -> Result<Bytes> {
-		tracing::info!(
-			"downloading {}/{}",
-			self.global.config().source_bucket.name,
-			self.job.id.to_string(),
-		);
-
-		let response = self
-			.global
-			.s3_source_bucket()
-			.get_object(&self.job.task.input_path)
-			.await
-			.map_err(ProcessorError::S3Download)?;
+		if self.job.task.input_path.starts_with("http://") || self.job.task.input_path.starts_with("https://") {
+			if !self.global.config().allow_http {
+				return Err(ProcessorError::HttpDownloadDisabled);
+			}
+
+			tracing::info!("downloading {}", self.job.task.input_path);
+
+			let response = reqwest::get(&self.job.task.input_path)
+				.await
+				.map_err(ProcessorError::HttpDownload)?;
+
+			response.error_for_status_ref().map_err(ProcessorError::HttpDownload)?;
 
-		if (200..299).contains(&response.status_code()) {
-			Ok(response.bytes().clone())
+			Ok(response.bytes().await.map_err(ProcessorError::HttpDownload)?)
 		} else {
-			Err(ProcessorError::S3Download(s3::error::S3Error::HttpFail))
+			tracing::info!(
+				"downloading {}/{}",
+				self.global.config().source_bucket.name,
+				self.job.task.input_path
+			);
+
+			let response = self
+				.global
+				.s3_source_bucket()
+				.get_object(&self.job.task.input_path)
+				.await
+				.map_err(ProcessorError::S3Download)?;
+
+			if (200..299).contains(&response.status_code()) {
+				Ok(response.bytes().clone())
+			} else {
+				Err(ProcessorError::S3Download(s3::error::S3Error::HttpFail))
+			}
 		}
 	}