support streaming content from S3, start using it in the webserver

Author: syphar

.sqlx/query-3bdc47a7b7457e290e2c63f9c22742d17a52940631caa0688d3c8b5e2c3765c8.json renamed to .sqlx/query-2fd2aad681960b30ca5149dfc7050c477667d5f022349661385026c757df88cc.json

@@ -64,8 +64,10 @@ derive_more = { version = "2.0.0", features = ["display"] }
 
 # Async
 tokio = { version = "1.0", features = ["rt-multi-thread", "signal", "macros"] }
+tokio-util = { version = "0.7.15", default-features = false, features = ["io"] }
 futures-util = "0.3.5"
 async-stream = "0.3.5"
+async-compression = { version = "0.4.25", features = ["tokio", "bzip2", "zstd", "gzip"] }
 aws-config = "1.0.0"
 aws-sdk-s3 = "1.3.0"
 aws-sdk-cloudfront = "1.3.0"

.sqlx/query-f0239a895d0ef72aff8d99f77a35656d2642564a6a3c40d742fc1b62d1c80d59.json renamed to .sqlx/query-ce9cf294c964be76b19585a35c37f4b3643b09cfea98ed8ba13e4dadb5b5e48c.json

@@ -1,9 +1,9 @@
-use super::{Blob, FileRange};
+use super::{Blob, FileRange, StreamingBlob};
 use crate::{InstanceMetrics, db::Pool, error::Result};
 use chrono::{DateTime, Utc};
 use futures_util::stream::{Stream, TryStreamExt};
 use sqlx::Acquire;
-use std::sync::Arc;
+use std::{io, sync::Arc};
 
 pub(crate) struct DatabaseBackend {
     pool: Pool,
@@ -58,38 +58,27 @@ impl DatabaseBackend {
         }
     }
 
-    pub(super) async fn get(
+    pub(super) async fn get_stream(
         &self,
         path: &str,
-        max_size: usize,
         range: Option<FileRange>,
-    ) -> Result<Blob> {
-        // The maximum size for a BYTEA (the type used for `content`) is 1GB, so this cast is safe:
-        // https://www.postgresql.org/message-id/162867790712200946i7ba8eb92v908ac595c0c35aee%40mail.gmail.com
-        let max_size = max_size.min(i32::MAX as usize) as i32;
-
+    ) -> Result<StreamingBlob> {
         struct Result {
             path: String,
             mime: String,
             date_updated: DateTime<Utc>,
             compression: Option<i32>,
             content: Option<Vec<u8>>,
-            is_too_big: bool,
         }
 
         let result = if let Some(r) = range {
-            // when we only want to get a range we can validate already if the range is small enough
-            if (r.end() - r.start() + 1) > max_size as u64 {
-                return Err(std::io::Error::other(crate::error::SizeLimitReached).into());
-            }
             let range_start = i32::try_from(*r.start())?;
 
             sqlx::query_as!(
                 Result,
                 r#"SELECT
                      path, mime, date_updated, compression,
-                     substring(content from $2 for $3) as content,
-                     FALSE as "is_too_big!"
+                     substring(content from $2 for $3) as content
                  FROM files
                  WHERE path = $1;"#,
                 path,
@@ -105,35 +94,35 @@ impl DatabaseBackend {
             sqlx::query_as!(
                 Result,
                 r#"SELECT
-                     path, mime, date_updated, compression,
-                     (CASE WHEN LENGTH(content) <= $2 THEN content ELSE NULL END) AS content,
-                     (LENGTH(content) > $2) AS "is_too_big!"
+                     path,
+                     mime,
+                     date_updated,
+                     compression,
+                     content
                  FROM files
                  WHERE path = $1;"#,
                 path,
-                max_size,
             )
             .fetch_optional(&self.pool)
             .await?
             .ok_or(super::PathNotFoundError)?
         };
 
-        if result.is_too_big {
-            return Err(std::io::Error::other(crate::error::SizeLimitReached).into());
-        }
-
         let compression = result.compression.map(|i| {
             i.try_into()
                 .expect("invalid compression algorithm stored in database")
         });
-        Ok(Blob {
+        let content = result.content.unwrap_or_default();
+        let content_len = content.len();
+        Ok(StreamingBlob {
             path: result.path,
             mime: result
                 .mime
                 .parse()
                 .unwrap_or(mime::APPLICATION_OCTET_STREAM),
             date_updated: result.date_updated,
-            content: result.content.unwrap_or_default(),
+            content: Box::new(io::Cursor::new(content)),
+            content_length: content_len,
             compression,
         })
     }