use axum::body::Body; use axum::extract::{Path, Query, State}; use axum::http::header::CONTENT_TYPE; use axum::http::HeaderMap; use axum::Json; use futures::StreamExt; use serde::Deserialize; use sha2::Digest; use tokio::io::AsyncWriteExt; use uuid::Uuid; use crate::auth::AuthUser; use crate::error::{ApiError, ApiResult}; use crate::jobs; use crate::models::{JobKind, Photo, PhotoStatus}; use crate::owned; use crate::s3; use crate::state::AppState; #[derive(Deserialize)] pub struct UploadQuery { filename: String, } fn sanitize_filename(raw: &str) -> Result { let base = raw.rsplit(['/', '\\']).next().unwrap_or(raw); let cleaned: String = base .chars() .map(|c| { if c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_' | ' ' | '(' | ')') { c } else { '_' } }) .collect(); let cleaned = cleaned.trim().trim_start_matches('.').to_string(); if cleaned.is_empty() { return Err(ApiError::bad_request("invalid filename")); } Ok(cleaned.chars().take(150).collect()) } pub async fn upload( State(state): State, user: AuthUser, Path(album_id): Path, Query(query): Query, headers: HeaderMap, body: Body, ) -> ApiResult> { owned::album(&state, album_id, user.id).await?; let filename = sanitize_filename(&query.filename)?; let content_type = headers .get(CONTENT_TYPE) .and_then(|v| v.to_str().ok()) .unwrap_or("application/octet-stream") .to_string(); // Stream the request body to a temp file so large raws never sit in // memory, hashing as it flows: sha256 for duplicate detection, crc32 for // spool-free zip downloads. let dir = tempfile::tempdir().map_err(anyhow::Error::from)?; let path = dir.path().join("upload.bin"); let mut file = tokio::fs::File::create(&path) .await .map_err(anyhow::Error::from)?; let mut stream = body.into_data_stream(); let mut sha = sha2::Sha256::new(); let mut crc = crc32fast::Hasher::new(); let mut size: i64 = 0; while let Some(chunk) = stream.next().await { let chunk = chunk.map_err(|e| ApiError::bad_request(format!("upload aborted: {e}")))?; size += chunk.len() as i64; sha.update(&chunk); crc.update(&chunk); file.write_all(&chunk).await.map_err(anyhow::Error::from)?; } file.flush().await.map_err(anyhow::Error::from)?; drop(file); if size == 0 { return Err(ApiError::bad_request("empty upload")); } let sha256 = hex::encode(sha.finalize()); let crc32 = i64::from(crc.finalize()); // Same content already in this album? Return it — re-dragging a folder // after a partial upload just fills the gaps instead of duplicating. let existing: Option = sqlx::query_as("select * from photos where album_id = $1 and sha256 = $2") .bind(album_id) .bind(&sha256) .fetch_optional(&state.db) .await?; if let Some(existing) = existing { return Ok(Json(existing)); } // Upload to S3 first, then create the row and enqueue processing in one // transaction — a photo row can never exist without its job, and a failed // transaction (e.g. album deleted mid-upload) cleans up the S3 object. let photo_id = Uuid::new_v4(); let key = s3::original_key(photo_id, &filename); s3::put_file(&state, &key, &path, &content_type).await?; let result: Result<(sqlx::Transaction<'static, sqlx::Postgres>, Photo), sqlx::Error> = async { let mut tx = state.db.begin().await?; let photo: Photo = sqlx::query_as( "insert into photos (id, album_id, filename, content_type, size_bytes, status, sha256, crc32) values ($1, $2, $3, $4, $5, $6, $7, $8) returning *", ) .bind(photo_id) .bind(album_id) .bind(&filename) .bind(&content_type) .bind(size) .bind(PhotoStatus::Uploaded.as_str()) .bind(&sha256) .bind(crc32) .fetch_one(&mut *tx) .await?; jobs::enqueue( &mut *tx, JobKind::ProcessPhoto, serde_json::json!({ "photo_id": photo_id }), ) .await?; Ok((tx, photo)) } .await; let (tx, photo) = match result { Ok(pair) => pair, Err(e) => { // Nothing committed — safe to remove the freshly stored original. if let Err(cleanup) = s3::delete_prefix(&state, &s3::photo_prefix(photo_id)).await { tracing::error!("failed to clean up s3 after aborted upload: {cleanup:#}"); } // Concurrent identical upload beat us to the unique index — hand // back the winner instead of an error. let unique_violation = matches!( &e, sqlx::Error::Database(db) if db.code().as_deref() == Some("23505") ); if unique_violation { let winner: Option = sqlx::query_as("select * from photos where album_id = $1 and sha256 = $2") .bind(album_id) .bind(&sha256) .fetch_optional(&state.db) .await?; if let Some(winner) = winner { return Ok(Json(winner)); } } return Err(e.into()); } }; if let Err(e) = tx.commit().await { // A failed COMMIT is ambiguous (it may have been applied); deleting // the S3 object here could destroy a committed photo's original, so // leave it — an orphaned object beats data loss. tracing::error!( "commit failed after upload of photo {photo_id}; leaving s3 object in place: {e}" ); return Err(anyhow::Error::from(e).context("saving upload").into()); } Ok(Json(photo)) } /// Client-side dedup support: lets the uploader skip transferring files whose /// content already exists in the album. pub async fn by_hash( State(state): State, user: AuthUser, Path((album_id, sha256)): Path<(Uuid, String)>, ) -> ApiResult> { owned::album(&state, album_id, user.id).await?; let photo: Option = sqlx::query_as("select * from photos where album_id = $1 and sha256 = $2") .bind(album_id) .bind(&sha256) .fetch_optional(&state.db) .await?; photo.map(Json).ok_or_else(ApiError::not_found) } pub async fn delete( State(state): State, user: AuthUser, Path(photo_id): Path, ) -> ApiResult> { owned::photo(&state, photo_id, user.id).await?; let mut tx = state.db.begin().await?; let deleted = sqlx::query("delete from photos where id = $1") .bind(photo_id) .execute(&mut *tx) .await?; if deleted.rows_affected() == 0 { return Err(ApiError::not_found()); } jobs::enqueue( &mut *tx, JobKind::DeleteS3Prefix, serde_json::json!({ "prefix": s3::photo_prefix(photo_id) }), ) .await?; tx.commit().await?; Ok(Json(serde_json::json!({ "ok": true }))) } pub async fn reprocess( State(state): State, user: AuthUser, Path(photo_id): Path, ) -> ApiResult> { owned::photo(&state, photo_id, user.id).await?; let mut tx = state.db.begin().await?; let updated = sqlx::query("update photos set status = $2, error = null where id = $1") .bind(photo_id) .bind(PhotoStatus::Uploaded.as_str()) .execute(&mut *tx) .await?; if updated.rows_affected() == 0 { return Err(ApiError::not_found()); } jobs::ensure_process_photo(&mut tx, photo_id).await?; tx.commit().await?; Ok(Json(serde_json::json!({ "ok": true }))) }