stefiosif/loft
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: generate and display thumbnails asynchronously

Browse Source
This commit is contained in:
stefiosif
2026-06-28 12:18:26 +03:00
parent 23933f47a0
commit affe9bcd14
16 changed files with 886 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
backend/src/error.rs
View File

@@ -19,6 +19,7 @@ pub enum LoftError {
NoFileProvided,
MultipartError(String),
InvalidRange,
ThumbnailGenerationError(String),
}
impl fmt::Display for LoftError {
@@ -51,6 +52,12 @@ impl From<argon2::password_hash::Error> for LoftError {
}
}
impl From<image::ImageError> for LoftError {
fn from(value: image::ImageError) -> Self {
LoftError::ThumbnailGenerationError(value.to_string())
}
}
impl std::error::Error for LoftError {}
impl IntoResponse for LoftError {
@@ -80,6 +87,10 @@ impl IntoResponse for LoftError {
error!("opendal storage error: {e}");
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
Self::ThumbnailGenerationError(e) => {
error!("thumbnail generation error: {e}");
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
Self::NoFileProvided => StatusCode::BAD_REQUEST.into_response(),
Self::MultipartError(e) => {
info!("bad request: {e}");

16
backend/src/main.rs
View File

@@ -1,6 +1,7 @@
mod ctx;
mod error;
mod model;
mod tasks;
mod web;
use std::{net::SocketAddr, time::Duration};
@@ -13,6 +14,7 @@ use axum::{
middleware,
};
use sqlx::PgPool;
use tokio::sync::mpsc::{self};
use tower_cookies::CookieManagerLayer;
use tower_governor::{
GovernorLayer, governor::GovernorConfigBuilder, key_extractor::SmartIpKeyExtractor,
@@ -31,12 +33,15 @@ use crate::{
model::{FileRepository, UserRepository},
web::{
mw_auth::{mw_ctx_resolver, mw_require_auth},
routes_file::{FileApi, routes_file},
routes_file::{FileApi, FileState, routes_file},
routes_health::{HealthApi, routes_health},
routes_login::{AuthApi, routes_auth},
},
};
const BUFFER_SIZE: usize = 100;
const BODY_LIMIT: usize = 1000 * 1000 * 1000 * 5;
#[tokio::main]
async fn main() -> Result<()> {
tracing_subscriber::registry()
@@ -51,11 +56,16 @@ async fn main() -> Result<()> {
dotenvy::dotenv().ok();
let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL must be set");
const BODY_LIMIT: usize = 1000 * 1000 * 1000 * 5;
let pool = PgPool::connect(&database_url).await?;
sqlx::migrate!().run(&pool).await?;
let file_repository = FileRepository::new(pool.clone())?;
let routes_file = routes_file(file_repository.clone())
let (tx, rx) = mpsc::channel(BUFFER_SIZE);
tokio::spawn(tasks::run_thumbnail_worker(rx, file_repository.clone()));
let file_service = FileState::new(file_repository.clone(), tx);
let routes_file = routes_file(file_service)
.route_layer(middleware::from_fn(mw_require_auth))
.layer(DefaultBodyLimit::max(BODY_LIMIT));

67
backend/src/model.rs
View File

@@ -17,6 +17,7 @@ pub struct FileRecord {
#[serde(skip_serializing)]
pub storage_key: String,
pub uploaded_at: chrono::DateTime<chrono::Utc>,
pub thumbnail_storage_path: Option<String>,
}
#[derive(Clone)]
@@ -90,6 +91,18 @@ impl FileRepository {
Ok(stream)
}
pub async fn thumbnail(
&self,
file_id: i64,
user_id: i64,
) -> Result<impl Stream<Item = std::io::Result<Bytes>> + use<>> {
let record = self.get_file(file_id, user_id).await?;
let path = format!("thumbnail/{}", &record.storage_key);
let reader = self.op.reader(&path).await?;
let stream = reader.into_bytes_stream(0..).await?;
Ok(stream)
}
pub async fn stream_part(
&self,
file_id: i64,
@@ -155,6 +168,29 @@ impl FileRepository {
Ok(records)
}
pub async fn update_thumbnail_storage_path(
&self,
path: &str,
file_record_id: i64,
user_id: i64,
) -> Result<()> {
sqlx::query!(
r#"
UPDATE file_records
SET thumbnail_storage_path = $1
WHERE id = $2
AND user_id = $3
"#,
path,
file_record_id,
user_id
)
.execute(&self.pool)
.await?;
Ok(())
}
}
#[derive(Clone, Debug, FromRow)]
@@ -383,6 +419,37 @@ mod tests {
));
}
#[tokio::test]
#[serial_test::serial]
async fn test_update_thumbnail_storage_path() {
let user_repository = user_repository().await.unwrap();
let user = user_repository
.create_user("username", "password_hash")
.await
.unwrap();
let file_repository = file_repository().await.unwrap();
truncate_file_records(&file_repository.pool).await;
let record = file_repository
.create_file_record(user.id, "a.png", 2, "a.png-uuid")
.await
.unwrap();
assert_eq!(record.thumbnail_storage_path, None);
let path = "thumbnail/a.png-uuid";
file_repository
.update_thumbnail_storage_path(path, record.id, user.id)
.await
.unwrap();
let updated = file_repository.get_file(record.id, user.id).await.unwrap();
assert_eq!(updated.thumbnail_storage_path, Some(path.to_string()));
truncate_file_records(&file_repository.pool).await;
truncate_users(&user_repository.pool).await;
}
#[tokio::test]
#[serial_test::serial]
async fn test_stream_part() {

81
backend/src/tasks.rs Normal file
View File

@@ -0,0 +1,81 @@
use image::codecs::jpeg::JpegEncoder;
use tokio::sync::mpsc::Receiver;
use tracing::{info, warn};
use crate::{
error::Result,
model::{FileRecord, FileRepository},
};
// change thumbnail task to have the file id in the param
// with the file id we can call the file repository and get the file record
// which will grand us access to the opendal storage... and also the MIME type
// so we can have a match statement after a parser that will give us the type of
// process we should do in order to create the thumbnail..
// for example an image will be processed with image crate but a video will be processed with tokio::command->ffmpeg
#[derive(Clone, Debug)]
pub struct ThumbnailTask {
record_id: i64,
user_id: i64,
}
impl ThumbnailTask {
pub fn new(record_id: i64, user_id: i64) -> Self {
Self { record_id, user_id }
}
}
pub async fn run_thumbnail_worker(
mut rx: Receiver<ThumbnailTask>,
file_repository: FileRepository,
) {
while let Some(task) = rx.recv().await {
let record_id = task.record_id;
if let Err(e) = process_task(task, &file_repository).await {
warn!("thumbnail task {} failed: {}", record_id, e);
}
}
}
async fn process_task(task: ThumbnailTask, file_repository: &FileRepository) -> Result<()> {
let record_id = task.record_id;
info!("task for file record {} received", record_id);
let record = file_repository
.get_file(task.record_id, task.user_id)
.await?;
let file_type = &record.file_type;
// TODO: create enum that maps file_type (which is actually mime_type) into file category (image, video, ...)
if file_type.contains("image") {
generate_for_image(&record, file_repository).await?;
} else {
warn!(
"thumbnail generation for file type {} not yet supported",
file_type
);
}
Ok(())
}
async fn generate_for_image(record: &FileRecord, file_repository: &FileRepository) -> Result<()> {
let storage_key = &record.storage_key;
let bytes = file_repository.op.read(storage_key).await?.to_vec();
let image = image::load_from_memory(&bytes)?;
let thumbnail = image.thumbnail(256, 256);
let rgb = thumbnail.to_rgb8();
let mut buf = Vec::new();
let mut enc = JpegEncoder::new_with_quality(&mut buf, 50);
enc.encode_image(&rgb)?;
let path = format!("thumbnail/{}", storage_key);
file_repository.op.write(&path, buf).await?;
file_repository
.update_thumbnail_storage_path(&path, record.id, record.user_id)
.await?;
Ok(())
}

110
backend/src/web/routes_file.rs
View File

@@ -7,12 +7,15 @@ use axum::{
routing::get,
};
use sqlx::types::uuid;
use tokio::sync::mpsc::Sender;
use tracing::warn;
use utoipa::OpenApi;
use crate::{
ctx::Ctx,
error::{LoftError, Result},
model::{FileRecord, FileRepository},
tasks::ThumbnailTask,
};
#[derive(OpenApi)]
@@ -23,19 +26,36 @@ use crate::{
get_file,
delete_file,
download_file,
thumbnail,
stream_part
),
components(schemas(FileRecord))
)]
pub struct FileApi;
pub fn routes_file(file_repository: FileRepository) -> Router {
#[derive(Clone)]
pub struct FileState {
pub file_repository: FileRepository,
pub tx: Sender<ThumbnailTask>,
}
impl FileState {
pub fn new(file_repository: FileRepository, tx: Sender<ThumbnailTask>) -> Self {
Self {
file_repository,
tx,
}
}
}
pub fn routes_file(file_service: FileState) -> Router {
Router::new()
.route("/files", get(list_files).post(upload_file))
.route("/files/{id}", get(get_file).delete(delete_file))
.route("/files/{id}/download", get(download_file))
.route("/files/{id}/thumbnail", get(thumbnail))
.route("/files/{id}/stream_part", get(stream_part))
.with_state(file_repository)
.with_state(file_service)
}
/// Upload a file
@@ -53,7 +73,7 @@ pub fn routes_file(file_repository: FileRepository) -> Router {
)
)]
async fn upload_file(
State(file_repository): State<FileRepository>,
State(service): State<FileState>,
ctx: Ctx,
mut multipart: Multipart,
) -> Result<(StatusCode, Json<FileRecord>)> {
@@ -63,18 +83,27 @@ async fn upload_file(
if field.name() == Some("file") {
let name = field.file_name().map(str::to_string).unwrap_or_default();
let key = format!("{name}-{}", uuid::Uuid::new_v4());
let size = file_repository.upload_file(field, &key).await?;
let size = service.file_repository.upload_file(field, &key).await?;
uploaded = Some((name, key, size));
}
}
let (name, key, size) = uploaded.ok_or(LoftError::NoFileProvided)?;
let file_record = file_repository
let record = service
.file_repository
.create_file_record(ctx.user_id(), &name, size, &key)
.await?;
Ok((StatusCode::CREATED, Json(file_record)))
if let Err(e) = service
.tx
.send(ThumbnailTask::new(record.id, ctx.user_id()))
.await
{
warn!("failed to send thumbnail task for file {}: {e}", record.id)
}
Ok((StatusCode::CREATED, Json(record)))
}
/// Get a file's metadata
@@ -92,11 +121,12 @@ async fn upload_file(
)
)]
async fn get_file(
State(file_repository): State<FileRepository>,
State(service): State<FileState>,
ctx: Ctx,
Path(file_id): Path<u64>,
) -> Result<Json<FileRecord>> {
let record = file_repository
let record = service
.file_repository
.get_file(file_id as i64, ctx.user_id())
.await?;
Ok(Json(record))
@@ -117,16 +147,43 @@ async fn get_file(
)
)]
async fn download_file(
State(file_repository): State<FileRepository>,
State(service): State<FileState>,
ctx: Ctx,
Path(file_id): Path<u64>,
) -> Result<impl IntoResponse> {
let stream = file_repository
let stream = service
.file_repository
.download_file(file_id as i64, ctx.user_id())
.await?;
Ok(Body::from_stream(stream))
}
/// Download a file thumbnail
#[utoipa::path(
get,
path = "/api/files/{id}/thumbnail",
tag = "files",
security(("cookie_auth" = [])),
params(("id" = u64, Path, description = "File id")),
responses(
(status = 200, description = "File bytes", content_type = "application/octet-stream"),
(status = 404, description = "File not found"),
(status = 401, description = "Unauthorized"),
(status = 500, description = "Internal server error")
)
)]
async fn thumbnail(
State(service): State<FileState>,
ctx: Ctx,
Path(file_id): Path<u64>,
) -> Result<impl IntoResponse> {
let stream = service
.file_repository
.thumbnail(file_id as i64, ctx.user_id())
.await?;
Ok(Body::from_stream(stream))
}
/// Stream a byte range of a file (HTTP 206)
#[utoipa::path(
get,
@@ -146,24 +203,26 @@ async fn download_file(
)
)]
async fn stream_part(
State(file_repository): State<FileRepository>,
State(service): State<FileState>,
ctx: Ctx,
headers: HeaderMap,
Path(file_id): Path<u64>,
) -> Result<impl IntoResponse> {
let file_record = file_repository
let record = service
.file_repository
.get_file(file_id as i64, ctx.user_id())
.await?;
let file_size = file_record.size as u64;
let file_size = record.size as u64;
let (start, end): (u64, u64) = parse_range(&headers, file_size)?;
let stream = file_repository
let stream = service
.file_repository
.stream_part(file_id as i64, ctx.user_id(), start, end + 1)
.await?;
let respones = Response::builder()
.status(StatusCode::PARTIAL_CONTENT)
.header(header::CONTENT_TYPE, file_record.file_type)
.header(header::CONTENT_TYPE, record.file_type)
.header(header::CONTENT_LENGTH, end - start + 1)
.header(
header::CONTENT_RANGE,
@@ -207,11 +266,12 @@ fn parse_range(headers: &HeaderMap, file_size: u64) -> Result<(u64, u64)> {
)
)]
async fn delete_file(
State(file_repository): State<FileRepository>,
State(service): State<FileState>,
ctx: Ctx,
Path(file_id): Path<u64>,
) -> Result<Json<FileRecord>> {
let file = file_repository
let file = service
.file_repository
.delete_file(file_id as i64, ctx.user_id())
.await?;
Ok(Json(file))
@@ -229,11 +289,8 @@ async fn delete_file(
(status = 500, description = "Internal server error")
)
)]
async fn list_files(
State(file_repository): State<FileRepository>,
ctx: Ctx,
) -> Result<Json<Vec<FileRecord>>> {
let files = file_repository.list_files(ctx.user_id()).await?;
async fn list_files(State(service): State<FileState>, ctx: Ctx) -> Result<Json<Vec<FileRecord>>> {
let files = service.file_repository.list_files(ctx.user_id()).await?;
Ok(Json(files))
}
@@ -251,13 +308,14 @@ mod tests {
use rand::RngExt;
use serde_json::json;
use sqlx::PgPool;
use tokio::sync::mpsc;
use tower_cookies::CookieManagerLayer;
use crate::{
model::{FileRepository, UserRepository},
web::{
mw_auth::{mw_ctx_resolver, mw_require_auth},
routes_file::routes_file,
routes_file::{FileState, routes_file},
},
};
@@ -282,8 +340,12 @@ mod tests {
async fn test_server() -> TestServer {
let user_repository = user_repository().await;
let file_repository = file_repository().await;
let (tx, _) = mpsc::channel(100);
let file_service = FileState::new(file_repository.clone(), tx);
let routes_file =
routes_file(file_repository.clone()).route_layer(middleware::from_fn(mw_require_auth));
routes_file(file_service).route_layer(middleware::from_fn(mw_require_auth));
let app = Router::new()
.nest("/api", routes_file)
.layer(middleware::from_fn_with_state(