Content dedup on both sides, stored checksums, spool-free zips, upload stats

- sha256+crc32 hashed during upload streaming; unique index per album
- duplicate content returns the existing photo (race-safe via 23505)
- client hashes locally (WebCrypto) and skips the transfer entirely for
  content the album already has
- zip downloads stream S3->response directly using the stored crc32;
  pre-hash photos spool once and self-heal (crc via zip, sha via reprocess)
- upload UI: overall progress bar, bytes, live speed, ETA

src/imaging.rs

@@ -62,7 +62,24 @@ pub async fn process_photo(state: &AppState, photo_id: Uuid) -> anyhow::Result<(
         .unwrap_or("bin")
         .to_lowercase();
     let src_path = dir.path().join(format!("original.{extension}"));
-    download(state, &s3::original_key(photo_id, &photo.filename), &src_path).await?;
+    let (sha256, crc32) =
+        download(state, &s3::original_key(photo_id, &photo.filename), &src_path).await?;
+    if photo.sha256.is_none() || photo.crc32.is_none() {
+        // Best-effort backfill for pre-hash photos; a duplicate in the same
+        // album trips the unique index, which is fine — skip silently.
+        if let Err(e) = sqlx::query(
+            "update photos set sha256 = coalesce(sha256, $2), crc32 = coalesce(crc32, $3)
+             where id = $1",
+        )
+        .bind(photo_id)
+        .bind(&sha256)
+        .bind(crc32)
+        .execute(&state.db)
+        .await
+        {
+            tracing::debug!("hash backfill skipped for {photo_id}: {e}");
+        }
+    }
 
     let meta = exif_metadata(&src_path).await?;
 
@@ -103,7 +120,11 @@ pub async fn process_photo(state: &AppState, photo_id: Uuid) -> anyhow::Result<(
     Ok(())
 }
 
-async fn download(state: &AppState, key: &str, path: &Path) -> anyhow::Result<()> {
+/// Download the original, hashing along the way so legacy photos (uploaded
+/// before hashes existed) can be backfilled.
+async fn download(state: &AppState, key: &str, path: &Path) -> anyhow::Result<(String, i64)> {
+    use sha2::Digest;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
     let object = state
         .s3
         .get_object()
@@ -114,8 +135,20 @@ async fn download(state: &AppState, key: &str, path: &Path) -> anyhow::Result<()
         .with_context(|| format!("fetching s3://{}/{key}", state.config.s3_bucket))?;
     let mut reader = object.body.into_async_read();
     let mut file = tokio::fs::File::create(path).await?;
-    tokio::io::copy(&mut reader, &mut file).await?;
-    Ok(())
+    let mut sha = sha2::Sha256::new();
+    let mut crc = crc32fast::Hasher::new();
+    let mut buf = vec![0u8; 128 * 1024];
+    loop {
+        let n = reader.read(&mut buf).await?;
+        if n == 0 {
+            break;
+        }
+        sha.update(&buf[..n]);
+        crc.update(&buf[..n]);
+        file.write_all(&buf[..n]).await?;
+    }
+    file.flush().await?;
+    Ok((hex::encode(sha.finalize()), i64::from(crc.finalize())))
 }
 
 #[derive(Default)]

src/models.rs

@@ -115,6 +115,10 @@ pub struct Photo {
     pub height: Option<i32>,
     pub taken_at: Option<DateTime<Utc>>,
     pub processed_at: Option<DateTime<Utc>>,
+    #[serde(skip_serializing)]
+    pub sha256: Option<String>,
+    #[serde(skip_serializing)]
+    pub crc32: Option<i64>,
     pub created_at: DateTime<Utc>,
 }
 

src/routes/mod.rs

@@ -77,6 +77,7 @@ pub fn router(state: &AppState) -> Router<AppState> {
             get(shares::list).post(shares::create),
         )
         .route("/api/albums/{id}/zip", post(zip::album_zip))
+        .route("/api/albums/{id}/photos/by-hash/{sha256}", get(photos::by_hash))
         .route("/api/photos/{id}", delete(photos::delete))
         .route("/api/photos/{id}/reprocess", post(photos::reprocess))
         .route("/api/shares/{id}", delete(shares::delete))

src/routes/photos.rs

@@ -5,6 +5,7 @@ use axum::http::HeaderMap;
 use axum::Json;
 use futures::StreamExt;
 use serde::Deserialize;
+use sha2::Digest;
 use tokio::io::AsyncWriteExt;
 use uuid::Uuid;
 
@@ -60,17 +61,23 @@ pub async fn upload(
         .unwrap_or("application/octet-stream")
         .to_string();
 
-    // Stream the request body to a temp file so large raws never sit in memory.
+    // 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)?;
@@ -78,6 +85,20 @@ pub async fn upload(
     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<Photo> =
+        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
@@ -86,30 +107,33 @@ pub async fn upload(
     let key = s3::original_key(photo_id, &filename);
     s3::put_file(&state, &key, &path, &content_type).await?;
 
-    let result: ApiResult<(sqlx::Transaction<'static, sqlx::Postgres>, Photo)> = 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)
-             values ($1, $2, $3, $4, $5, $6)
-             returning *",
-        )
-        .bind(photo_id)
-        .bind(album_id)
-        .bind(&filename)
-        .bind(&content_type)
-        .bind(size)
-        .bind(PhotoStatus::Uploaded.as_str())
-        .fetch_one(&mut *tx)
-        .await?;
-        jobs::enqueue(
-            &mut *tx,
-            JobKind::ProcessPhoto,
-            serde_json::json!({ "photo_id": photo_id }),
-        )
-        .await?;
-        Ok((tx, photo))
-    }
-    .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,
@@ -118,7 +142,24 @@ pub async fn upload(
             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:#}");
             }
-            return Err(e);
+            // 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<Photo> =
+                    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 {
@@ -133,6 +174,21 @@ pub async fn upload(
     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<AppState>,
+    Path((album_id, sha256)): Path<(Uuid, String)>,
+) -> ApiResult<Json<Photo>> {
+    let photo: Option<Photo> =
+        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<AppState>,
     Path(photo_id): Path<Uuid>,

src/routes/zip.rs

@@ -136,6 +136,9 @@ struct Entry {
     offset: u64,
     dos_time: u16,
     dos_date: u16,
+    /// Stored at upload/processing time; photos from before hashes existed
+    /// have None and take the slower spool path (which backfills it).
+    crc: Option<u32>,
 }
 
 struct ZipPlan {
@@ -176,6 +179,7 @@ fn plan_zip(photos: &[Photo]) -> Result<ZipPlan, ApiError> {
             offset: entry_offset,
             dos_time,
             dos_date,
+            crc: photo.crc32.map(|v| v as u32),
         });
     }
     let cd_offset = offset;
@@ -277,9 +281,23 @@ fn stream_zip(state: AppState, photos: Vec<Photo>, album_name: &str) -> ApiResul
         .map_err(|e| anyhow::anyhow!("building response: {e}").into())
 }
 
-/// Download an original into an anonymous temp file, computing its CRC-32 and
-/// verifying the byte count matches what the zip plan promised.
-fn spool(state: &AppState, key: String, expected_size: u64) -> JoinHandle<anyhow::Result<(tokio::fs::File, u32)>> {
+enum Fetched {
+    /// CRC already known — the body streams straight into the response.
+    Direct(Box<aws_sdk_s3::operation::get_object::GetObjectOutput>),
+    /// Pre-hash photo: spooled to a temp file to compute the CRC first.
+    Spooled(tokio::fs::File, u32),
+}
+
+/// Start fetching an original. With a known CRC this only opens the S3
+/// response (the body is consumed later, straight into the zip stream);
+/// otherwise the object is spooled to an anonymous temp file to compute the
+/// CRC, verifying the byte count the zip plan promised.
+fn fetch_entry(
+    state: &AppState,
+    key: String,
+    expected_size: u64,
+    crc_known: bool,
+) -> JoinHandle<anyhow::Result<Fetched>> {
     let state = state.clone();
     tokio::spawn(async move {
         let object = state
@@ -290,6 +308,9 @@ fn spool(state: &AppState, key: String, expected_size: u64) -> JoinHandle<anyhow
             .send()
             .await
             .map_err(|e| anyhow::anyhow!("fetching {key}: {e}"))?;
+        if crc_known {
+            return Ok(Fetched::Direct(Box::new(object)));
+        }
         let mut file = tokio::fs::File::from_std(tempfile::tempfile()?);
         let mut reader = object.body.into_async_read();
         let mut hasher = crc32fast::Hasher::new();
@@ -310,7 +331,7 @@ fn spool(state: &AppState, key: String, expected_size: u64) -> JoinHandle<anyhow
         );
         file.flush().await?;
         file.seek(std::io::SeekFrom::Start(0)).await?;
-        Ok((file, hasher.finalize()))
+        Ok(Fetched::Spooled(file, hasher.finalize()))
     })
 }
 
@@ -323,20 +344,29 @@ async fn write_zip(
     const FLAGS: u16 = 0x0800;
     let mut crcs = Vec::with_capacity(plan.entries.len());
 
-    // Prefetch: spool the next object from S3 while streaming the current one.
-    let mut pending: Option<JoinHandle<anyhow::Result<(tokio::fs::File, u32)>>> = None;
+    // Prefetch: start fetching the next object while streaming the current one.
+    let mut pending: Option<JoinHandle<anyhow::Result<Fetched>>> = None;
     for (i, entry) in plan.entries.iter().enumerate() {
         let current = match pending.take() {
             Some(handle) => handle,
-            None => spool(state, entry.s3_key.clone(), entry.size),
+            None => fetch_entry(state, entry.s3_key.clone(), entry.size, entry.crc.is_some()),
         };
         if let Some(next) = plan.entries.get(i + 1) {
-            pending = Some(spool(state, next.s3_key.clone(), next.size));
+            pending = Some(fetch_entry(
+                state,
+                next.s3_key.clone(),
+                next.size,
+                next.crc.is_some(),
+            ));
         }
-        let (mut file, crc) = current
+        let fetched = current
             .await
-            .map_err(|e| anyhow::anyhow!("spool task failed: {e}"))?
+            .map_err(|e| anyhow::anyhow!("fetch task failed: {e}"))?
             .map_err(|e| anyhow::anyhow!("photo {}: {e:#}", entry.photo_id))?;
+        let crc = match &fetched {
+            Fetched::Direct(_) => entry.crc.expect("direct fetch implies known crc"),
+            Fetched::Spooled(_, crc) => *crc,
+        };
         crcs.push(crc);
 
         let mut lfh = Vec::with_capacity(30 + entry.name.len());
@@ -353,7 +383,28 @@ async fn write_zip(
         lfh.extend_from_slice(&0u16.to_le_bytes()); // extra len
         lfh.extend_from_slice(&entry.name);
         out.write_all(&lfh).await?;
-        tokio::io::copy(&mut file, &mut out).await?;
+        match fetched {
+            Fetched::Direct(object) => {
+                let mut reader = object.body.into_async_read();
+                let copied = tokio::io::copy(&mut reader, &mut out).await?;
+                anyhow::ensure!(
+                    copied == entry.size,
+                    "{} is {copied} bytes in s3 but {} in the database",
+                    entry.s3_key,
+                    entry.size
+                );
+            }
+            Fetched::Spooled(mut file, crc) => {
+                tokio::io::copy(&mut file, &mut out).await?;
+                // Self-heal: store the freshly computed crc so the next
+                // download of this photo streams directly.
+                let _ = sqlx::query("update photos set crc32 = coalesce(crc32, $2) where id = $1")
+                    .bind(entry.photo_id)
+                    .bind(i64::from(crc))
+                    .execute(&state.db)
+                    .await;
+            }
+        }
     }
 
     // Central directory.