diff --git a/Cargo.lock b/Cargo.lock
index c0a5e73..3ac1d15 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2562,6 +2562,7 @@ dependencies = [
"cookie",
"crc32fast",
"futures",
+ "hex",
"image",
"rand 0.8.7",
"reqwest",
diff --git a/Cargo.toml b/Cargo.toml
index 8c52e9b..34bc187 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -25,6 +25,7 @@ axum = { version = "0.8", features = ["macros"] }
axum-extra = { version = "0.10", features = ["cookie", "cookie-signed"] }
chrono = { version = "0.4", features = ["serde"] }
futures = "0.3"
+hex = "0.4"
image = "0.25"
rand = "0.8"
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
diff --git a/frontend/src/api.js b/frontend/src/api.js
index d935ba4..aee167c 100644
--- a/frontend/src/api.js
+++ b/frontend/src/api.js
@@ -73,6 +73,13 @@ export function postDownload(url, ids = '') {
form.remove()
}
+// SHA-256 of a File, matching the server's content hash — used to skip
+// uploading bytes the album already has.
+export async function sha256Hex(file) {
+ const digest = await crypto.subtle.digest('SHA-256', await file.arrayBuffer())
+ return [...new Uint8Array(digest)].map((b) => b.toString(16).padStart(2, '0')).join('')
+}
+
export function uploadFile(url, file, onProgress) {
return new Promise((resolve, reject) => {
const xhr = new XMLHttpRequest()
diff --git a/frontend/src/pages/AlbumPage.jsx b/frontend/src/pages/AlbumPage.jsx
index fee6032..816b742 100644
--- a/frontend/src/pages/AlbumPage.jsx
+++ b/frontend/src/pages/AlbumPage.jsx
@@ -1,21 +1,57 @@
import { useCallback, useEffect, useRef, useState } from 'react'
import { Link, useNavigate, useParams } from 'react-router-dom'
-import { api, postDownload, uploadFile } from '../api'
+import { api, postDownload, sha256Hex, uploadFile } from '../api'
import Gallery from '../components/Gallery'
import Lightbox from '../components/Lightbox'
-import SelectionBar from '../components/SelectionBar'
+import SelectionBar, { fmtBytes } from '../components/SelectionBar'
import Stars from '../components/Stars'
import useSelection from '../useSelection'
+function fmtEta(seconds) {
+ if (!isFinite(seconds) || seconds < 0) return ''
+ if (seconds < 60) return `${Math.ceil(seconds)}s`
+ if (seconds < 3600) return `${Math.ceil(seconds / 60)} min`
+ return `${Math.floor(seconds / 3600)}h ${Math.ceil((seconds % 3600) / 60)} min`
+}
+
const UPLOAD_CONCURRENCY = 3
function UploadZone({ albumId, onUploaded }) {
const [queue, setQueue] = useState([])
const [dragging, setDragging] = useState(false)
+ const [speed, setSpeed] = useState(0)
const inputRef = useRef(null)
const running = useRef(0)
const pending = useRef([])
const lastRefresh = useRef(0)
+ const loadedRef = useRef(0)
+
+ const totalBytes = queue.reduce((sum, item) => sum + item.file.size, 0)
+ const loadedBytes = queue.reduce(
+ (sum, item) =>
+ sum + (item.status === 'done' ? item.file.size : (item.progress || 0) * item.file.size),
+ 0,
+ )
+ loadedRef.current = loadedBytes
+ const active = queue.some((item) =>
+ ['uploading', 'queued', 'checking'].includes(item.status),
+ )
+
+ // Sample throughput once a second (EMA-smoothed) while uploads run.
+ useEffect(() => {
+ if (!active) {
+ setSpeed(0)
+ return
+ }
+ let last = { loaded: loadedRef.current, time: Date.now() }
+ const timer = setInterval(() => {
+ const now = Date.now()
+ const instant = (loadedRef.current - last.loaded) / ((now - last.time) / 1000)
+ last = { loaded: loadedRef.current, time: now }
+ setSpeed((prev) => (prev > 0 ? prev * 0.7 + instant * 0.3 : instant))
+ }, 1000)
+ return () => clearInterval(timer)
+ }, [active])
// Refresh the album at most every 5s during a bulk upload (the processing
// poll keeps it fresh anyway), plus once when the queue drains.
@@ -31,23 +67,27 @@ function UploadZone({ albumId, onUploaded }) {
while (running.current < UPLOAD_CONCURRENCY && pending.current.length > 0) {
const item = pending.current.shift()
running.current += 1
- setQueue((q) =>
- q.map((x) => (x.key === item.key ? { ...x, status: 'uploading' } : x)),
- )
- const url = `/api/albums/${albumId}/photos?filename=${encodeURIComponent(item.file.name)}`
- uploadFile(url, item.file, (p) =>
- setQueue((q) => q.map((x) => (x.key === item.key ? { ...x, progress: p } : x))),
- )
- .then(() =>
- setQueue((q) =>
- q.map((x) => (x.key === item.key ? { ...x, status: 'done', progress: 1 } : x)),
- ),
- )
- .catch((e) =>
- setQueue((q) =>
- q.map((x) => (x.key === item.key ? { ...x, status: 'error', error: e.message } : x)),
- ),
- )
+ const update = (patch) =>
+ setQueue((q) => q.map((x) => (x.key === item.key ? { ...x, ...patch } : x)))
+ const transfer = async () => {
+ // Hash locally first: content the album already has is skipped
+ // without transferring a single byte.
+ update({ status: 'checking' })
+ try {
+ const hash = await sha256Hex(item.file)
+ await api(`/api/albums/${albumId}/photos/by-hash/${hash}`)
+ update({ status: 'skipped', progress: 1 })
+ return
+ } catch {
+ // 404 (not there yet) or hashing unavailable — upload normally.
+ }
+ update({ status: 'uploading' })
+ const url = `/api/albums/${albumId}/photos?filename=${encodeURIComponent(item.file.name)}`
+ await uploadFile(url, item.file, (p) => update({ progress: p }))
+ update({ status: 'done', progress: 1 })
+ }
+ transfer()
+ .catch((e) => update({ status: 'error', error: e.message }))
.finally(() => {
running.current -= 1
refresh()
@@ -95,6 +135,22 @@ function UploadZone({ albumId, onUploaded }) {
}}
/>
Drop RAWs or JPGs here, or click to select
+ {queue.length > 0 && (
+ e.stopPropagation()}>
+
+
+ {queue.filter((i) => i.status === 'done' || i.status === 'skipped').length} /{' '}
+ {queue.length} files ·{' '}
+ {fmtBytes(loadedBytes)} of {fmtBytes(totalBytes)}
+ {active && speed > 0 && (
+ <>
+ {' · '}
+ {fmtBytes(speed)}/s · ~{fmtEta((totalBytes - loadedBytes) / speed)} left
+
+ )}
+
+
e.stopPropagation()}>
{queue.map((item) => (
@@ -102,6 +158,10 @@ function UploadZone({ albumId, onUploaded }) {
{item.file.name}
{item.status === 'error' ? (
{item.error}
+ ) : item.status === 'skipped' ? (
+ already uploaded
+ ) : item.status === 'checking' ? (
+ checking…
) : (
)}
diff --git a/frontend/src/styles.css b/frontend/src/styles.css
index 5b7fb44..09564ac 100644
--- a/frontend/src/styles.css
+++ b/frontend/src/styles.css
@@ -307,6 +307,18 @@ input:focus {
.upload-zone p {
margin: 0;
}
+.upload-summary {
+ display: flex;
+ flex-direction: column;
+ gap: 0.3rem;
+ margin-top: 1rem;
+ cursor: default;
+ font-size: 0.9rem;
+}
+.upload-total {
+ width: 100%;
+ accent-color: var(--accent);
+}
.upload-list {
list-style: none;
margin: 1rem 0 0;
diff --git a/migrations/0002_content_hash.sql b/migrations/0002_content_hash.sql
new file mode 100644
index 0000000..4ad285e
--- /dev/null
+++ b/migrations/0002_content_hash.sql
@@ -0,0 +1,10 @@
+-- Content hashes: sha256 powers duplicate-upload detection (same content in
+-- the same album is returned instead of copied); crc32 lets zip downloads
+-- stream originals straight from S3 without a local spool pass.
+-- Both are null for photos uploaded before this migration; they self-heal on
+-- reprocess (sha256 + crc32) and on zip download (crc32).
+alter table photos add column sha256 text;
+alter table photos add column crc32 bigint;
+
+create unique index photos_album_sha_uidx on photos (album_id, sha256)
+ where sha256 is not null;
diff --git a/src/imaging.rs b/src/imaging.rs
index d55c1fb..c34c796 100644
--- a/src/imaging.rs
+++ b/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)]
diff --git a/src/models.rs b/src/models.rs
index 25fe538..c7c1102 100644
--- a/src/models.rs
+++ b/src/models.rs
@@ -115,6 +115,10 @@ pub struct Photo {
pub height: Option,
pub taken_at: Option>,
pub processed_at: Option>,
+ #[serde(skip_serializing)]
+ pub sha256: Option,
+ #[serde(skip_serializing)]
+ pub crc32: Option,
pub created_at: DateTime,
}
diff --git a/src/routes/mod.rs b/src/routes/mod.rs
index ba184c8..0b1113e 100644
--- a/src/routes/mod.rs
+++ b/src/routes/mod.rs
@@ -77,6 +77,7 @@ pub fn router(state: &AppState) -> Router {
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))
diff --git a/src/routes/photos.rs b/src/routes/photos.rs
index e3a0cfd..50d6e2e 100644
--- a/src/routes/photos.rs
+++ b/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 =
+ 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 =
+ 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,
+ Path((album_id, sha256)): Path<(Uuid, String)>,
+) -> ApiResult> {
+ 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,
Path(photo_id): Path,
diff --git a/src/routes/zip.rs b/src/routes/zip.rs
index cca5a41..9088404 100644
--- a/src/routes/zip.rs
+++ b/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,
}
struct ZipPlan {
@@ -176,6 +179,7 @@ fn plan_zip(photos: &[Photo]) -> Result {
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, 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> {
+enum Fetched {
+ /// CRC already known — the body streams straight into the response.
+ Direct(Box),
+ /// 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> {
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 JoinHandle>> = None;
+ // Prefetch: start fetching the next object while streaming the current one.
+ let mut pending: Option>> = 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.