GITGUD

neiam neiam /dms

public · Issues · Pulls · Labels · Forks · Compare · Actions failure · Badges · Packages

⭐

Log in to mark this repository.

Add zip upload: ingest every image in an archive as its own upload

5a7f679 · gmorell · 2026-08-29 21:17

13 files +691 -52
Message 
{commit_body(@commit)}

Files changed

modified Cargo.lock
+40 −0
@@ -1338,6 +1338,7 @@ dependencies = [
1338 1338 "tracing",
1339 1339 "tracing-subscriber",
1340 1340 "uuid",
1341 + "zip",
1341 1342 ]
1342 1343
1343 1344 [[package]]
@@ -1528,6 +1529,7 @@ checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c"
1528 1529 dependencies = [
1529 1530 "crc32fast",
1530 1531 "miniz_oxide",
1532 + "zlib-rs",
1531 1533 ]
1532 1534
1533 1535 [[package]]
@@ -5090,6 +5092,12 @@ version = "0.2.5"
5090 5092 source = "registry+https://github.com/rust-lang/crates.io-index"
5091 5093 checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
5092 5094
5095 +[[package]]
5096 +name = "typed-path"
5097 +version = "0.12.3"
5098 +source = "registry+https://github.com/rust-lang/crates.io-index"
5099 +checksum = "8e28f89b80c87b8fb0cf04ab448d5dd0dd0ade2f8891bae878de66a75a28600e"
5100 +
5093 5101 [[package]]
5094 5102 name = "typeid"
5095 5103 version = "1.0.3"
@@ -5880,6 +5888,38 @@ dependencies = [
5880 5888 "syn 2.0.106",
5881 5889 ]
5882 5890
5891 +[[package]]
5892 +name = "zip"
5893 +version = "8.6.0"
5894 +source = "registry+https://github.com/rust-lang/crates.io-index"
5895 +checksum = "2d04a6b5381502aa6087c94c669499eb1602eb9c5e8198e534de571f7154809b"
5896 +dependencies = [
5897 + "crc32fast",
5898 + "flate2",
5899 + "indexmap",
5900 + "memchr",
5901 + "typed-path",
5902 + "zopfli",
5903 +]
5904 +
5905 +[[package]]
5906 +name = "zlib-rs"
5907 +version = "0.6.7"
5908 +source = "registry+https://github.com/rust-lang/crates.io-index"
5909 +checksum = "34b31d188d9d685a4f9c7b46d6e36631b07058d2cfe190267adce54dc230bf12"
5910 +
5911 +[[package]]
5912 +name = "zopfli"
5913 +version = "0.8.3"
5914 +source = "registry+https://github.com/rust-lang/crates.io-index"
5915 +checksum = "f05cd8797d63865425ff89b5c4a48804f35ba0ce8d125800027ad6017d2b5249"
5916 +dependencies = [
5917 + "bumpalo",
5918 + "crc32fast",
5919 + "log",
5920 + "simd-adler32",
5921 +]
5922 +
5883 5923 [[package]]
5884 5924 name = "zune-core"
5885 5925 version = "0.5.1"
modified Cargo.toml
+4 −1
@@ -19,7 +19,7 @@ path = "src/lib.rs"
19 19 [dependencies]
20 20 axum = { version = "0.8.5", features = ["multipart"] }
21 21 log = "0.4.28"
22 −tokio = "1.47.1"
22 +tokio = { version = "1.47.1", features = ["rt-multi-thread", "macros", "sync"] }
23 23 tower-http = { version = "0.6.6", features = ["fs", "trace"] }
24 24 tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
25 25 anyhow = "1.0.100"
@@ -62,6 +62,9 @@ rten = "0.21"
62 62 rten-imageproc = "0.21"
63 63 qrcode = { version = "0.14", default-features = false, features = ["svg"] }
64 64 reqwest = { version = "0.12", default-features = false, features = ["json", "multipart", "rustls-tls"] }
65 +# Zip ingest. Only the codecs real-world image archives use — "stored" is
66 +# always available, deflate covers everything Finder/Explorer produce.
67 +zip = { version = "8", default-features = false, features = ["deflate"] }
65 68
66 69 [build-dependencies]
67 70 minijinja-embed = "2.12"
modified README.md
+36 −0
@@ -104,6 +104,8 @@ All env-driven — defaults make sense for local dev.
104 104 | `BIND_ADDRESS` | `0.0.0.0:3007` | Webservice listener. |
105 105 | `STORAGE_ROOT` | `./storage` | Content-addressed file root. |
106 106 | `MAX_UPLOAD_BYTES` | `104857600` (100 MiB) | Per-upload cap. Bigger than this and the user gets a friendly redirect; the body limit layer matches. |
107 +| `MAX_ZIP_ENTRIES` | `1000` | Most images read out of one zip. Past it the import stops and says so. |
108 +| `MAX_ZIP_TOTAL_BYTES` | `1073741824` (1 GiB) | Total *decompressed* bytes read from one zip. A zip is bounded on the wire by `MAX_UPLOAD_BYTES`, but its contents are not — this is the zip-bomb guard. |
107 109 | `SESSION_DURATION_HOURS` | `24` | Browser session cookie lifetime. |
108 110 | `REGISTRATION_TOKEN_DURATION_DAYS` | `7` | Default invite expiry. |
109 111 | `AUTO_MIGRATE` | `true` | Run pending migrations on webservice boot. Disable in environments that prefer a separate migration step. |
@@ -156,6 +158,40 @@ reset-password alice`.
156 158 Inside the running container the binary is `dms-user-admin`; `just user-in-pod
157 159 "reset-password alice --generate"` wraps `podman compose exec`.
158 160
161 +## Ingest
162 +
163 +Three ways in, all converging on the same tail — content-address the bytes,
164 +dedupe per user against `(file_hash, user_id)`, insert the row, attach any
165 +user tags, enqueue the default job set.
166 +
167 +| Path | Web | API |
168 +|---|---|---|
169 +| One file | `POST /uploads` (multipart `file`) | `POST /api/uploads` |
170 +| A link | `POST /uploads/url` (form `url`) | `POST /api/uploads/url` |
171 +| A zip of images | `POST /uploads/zip` (multipart `file`) | `POST /api/uploads/zip` |
172 +
173 +Pasting anywhere on the site works too: an image on the clipboard attaches to
174 +the quick-upload form (or uploads directly), and a pasted link is fetched
175 +server-side so animated GIFs and videos keep their original bytes.
176 +
177 +Zip import walks the archive on a blocking thread and streams each image
178 +through a small channel, so peak memory is the zip plus a few images rather
179 +than the whole decompressed archive. Per entry:
180 +
181 +- Only entries whose name resolves to an image type are stored. Everything
182 + else is counted and skipped.
183 +- Directories, `__MACOSX/` resource forks, and dotfiles are ignored. Nested
184 + paths flatten to the basename — storage is content-addressed, so the
185 + archive's layout doesn't survive.
186 +- `MAX_UPLOAD_BYTES` caps each entry, enforced against the bytes actually
187 + read rather than the size the central directory claims.
188 +- `MAX_ZIP_ENTRIES` and `MAX_ZIP_TOTAL_BYTES` bound the whole archive.
189 +
190 +Every image inherits the description and tags submitted with the zip, and one
191 +AMQP connection dispatches jobs for the entire batch. A failing entry is
192 +counted, not fatal — the response reports imported / duplicate / skipped /
193 +failed counts, and says explicitly when a limit truncated the read.
194 +
159 195 ## Search
160 196
161 197 Two surfaces, same engine.
modified src/bin/webservice.rs
+2 −0
@@ -88,6 +88,7 @@ async fn main() -> Result<(), anyhow::Error> {
88 88 .route("/api-tokens/{id}/delete", post(handlers::delete_api_token))
89 89 .route("/uploads", get(handlers::list_uploads).post(handlers::upload_file))
90 90 .route("/uploads/url", post(handlers::upload_from_url))
91 + .route("/uploads/zip", post(handlers::upload_zip))
91 92 .route("/uploads/{id}", get(handlers::show_upload))
92 93 .route("/uploads/{id}/jobs", post(handlers::enqueue_job))
93 94 .route("/uploads/{id}/tags", post(handlers::add_tags))
@@ -107,6 +108,7 @@ async fn main() -> Result<(), anyhow::Error> {
107 108 .route("/profile", get(api_profile))
108 109 .route("/uploads", get(handlers::api_list_uploads).post(handlers::api_upload_file))
109 110 .route("/uploads/url", post(handlers::api_upload_from_url))
111 + .route("/uploads/zip", post(handlers::api_upload_zip))
110 112 .route("/uploads/{id}", get(handlers::api_show_upload))
111 113 .route("/uploads/{id}/tags", post(handlers::api_add_tags))
112 114 .route("/search", get(handlers::api_search_uploads))
modified src/config.rs
+13 −0
@@ -12,6 +12,11 @@ pub struct Config {
12 12 pub worker_prefetch: u16,
13 13 pub worker_max_attempts: i32,
14 14 pub max_upload_bytes: usize,
15 + /// Zip ingest guards. A zip is bounded by `max_upload_bytes` on the wire,
16 + /// but its *decompressed* contents are not — these cap the expansion so a
17 + /// zip bomb can't fill the archive.
18 + pub max_zip_entries: usize,
19 + pub max_zip_total_bytes: usize,
15 20 pub auto_migrate: bool,
16 21 pub registration_enabled: bool,
17 22 pub describe: DescribeConfig,
@@ -64,6 +69,14 @@ impl Config {
64 69 .unwrap_or_else(|_| "104857600".to_string()) // 100 MiB
65 70 .parse()
66 71 .unwrap_or(100 * 1024 * 1024),
72 + max_zip_entries: env::var("MAX_ZIP_ENTRIES")
73 + .ok()
74 + .and_then(|v| v.parse().ok())
75 + .unwrap_or(1000),
76 + max_zip_total_bytes: env::var("MAX_ZIP_TOTAL_BYTES")
77 + .ok()
78 + .and_then(|v| v.parse().ok())
79 + .unwrap_or(1024 * 1024 * 1024), // 1 GiB decompressed
67 80 auto_migrate: env::var("AUTO_MIGRATE")
68 81 .map(|v| matches!(v.to_ascii_lowercase().as_str(), "1" | "true" | "yes" | "on"))
69 82 .unwrap_or(true),
modified src/handlers/mod.rs
+4 −2
@@ -13,6 +13,8 @@ pub use failed::{
13 13 api_requeue_all_dlqs, api_requeue_dlq_kind,
14 14 };
15 15 pub use uploads::{
16 − list_uploads, upload_file, upload_from_url, show_upload, enqueue_job, add_tags, search_uploads,
17 − api_list_uploads, api_upload_file, api_upload_from_url, api_show_upload, api_add_tags, api_search_uploads,
16 + list_uploads, upload_file, upload_from_url, upload_zip, show_upload, enqueue_job, add_tags,
17 + search_uploads,
18 + api_list_uploads, api_upload_file, api_upload_from_url, api_upload_zip, api_show_upload,
19 + api_add_tags, api_search_uploads,
18 20 };
modified src/handlers/uploads.rs
+581 −49
@@ -44,17 +44,22 @@ fn human_bytes(n: usize) -> String {
44 44 }
45 45 }
46 46
47 −fn upload_redirect_error(msg: &str) -> Redirect {
48 − let encoded: String = msg
49 − .as_bytes()
50 − .iter()
51 − .map(|&b| match b {
47 +fn urlencode(s: &str) -> String {
48 + s.bytes()
49 + .map(|b| match b {
52 50 b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => (b as char).to_string(),
53 51 b' ' => "+".to_string(),
54 52 _ => format!("%{:02X}", b),
55 53 })
56 − .collect();
57 − Redirect::to(&format!("/uploads?error={}", encoded))
54 + .collect()
55 +}
56 +
57 +fn upload_redirect_error(msg: &str) -> Redirect {
58 + Redirect::to(&format!("/uploads?error={}", urlencode(msg)))
59 +}
60 +
61 +fn upload_redirect_notice(msg: &str) -> Redirect {
62 + Redirect::to(&format!("/uploads?notice={}", urlencode(msg)))
58 63 }
59 64
60 65 const PAGE_SIZE: u64 = 24;
@@ -163,6 +168,43 @@ pub async fn upload_from_url(
163 168 }
164 169 }
165 170
171 +/// POST /uploads/zip — ingest every image inside an uploaded zip archive.
172 +/// Each entry becomes its own upload row and runs the usual job pipeline, so a
173 +/// batch of photos can be dropped in without picking them one by one.
174 +pub async fn upload_zip(
175 + State(state): State<AppState>,
176 + user: UserContext,
177 + headers: HeaderMap,
178 + multipart: Multipart,
179 +) -> Response {
180 + if let Some(len) = headers
181 + .get(CONTENT_LENGTH)
182 + .and_then(|v| v.to_str().ok())
183 + .and_then(|s| s.parse::<usize>().ok())
184 + {
185 + if len > state.config.max_upload_bytes {
186 + return upload_redirect_error(&format!(
187 + "Zip too large ({}). Maximum is {}.",
188 + human_bytes(len),
189 + human_bytes(state.config.max_upload_bytes)
190 + ))
191 + .into_response();
192 + }
193 + }
194 +
195 + match ingest_zip(&state, &user, multipart).await {
196 + Ok(summary) => upload_redirect_notice(&summary.message()).into_response(),
197 + Err(IngestError::TooLarge { actual, limit }) => upload_redirect_error(&format!(
198 + "Zip too large ({}). Maximum is {}.",
199 + human_bytes(actual),
200 + human_bytes(limit)
201 + ))
202 + .into_response(),
203 + Err(IngestError::BadRequest(msg)) => upload_redirect_error(&msg).into_response(),
204 + Err(IngestError::Status(code)) => code.into_response(),
205 + }
206 +}
207 +
166 208 #[derive(Debug, Deserialize)]
167 209 pub struct AddTagsForm {
168 210 pub tags: Option<String>,
@@ -451,6 +493,62 @@ pub async fn api_upload_from_url(
451 493 }
452 494 }
453 495
496 +/// POST /api/uploads/zip — JSON sibling of `upload_zip`. Returns the ids that
497 +/// were created plus the counts for everything that was not.
498 +pub async fn api_upload_zip(
499 + State(state): State<AppState>,
500 + user: UserContext,
501 + headers: HeaderMap,
502 + multipart: Multipart,
503 +) -> Response {
504 + if let Some(len) = headers
505 + .get(CONTENT_LENGTH)
506 + .and_then(|v| v.to_str().ok())
507 + .and_then(|s| s.parse::<usize>().ok())
508 + {
509 + if len > state.config.max_upload_bytes {
510 + return (
511 + StatusCode::PAYLOAD_TOO_LARGE,
512 + Json(serde_json::json!({
513 + "error": "payload_too_large",
514 + "max_bytes": state.config.max_upload_bytes,
515 + "received_bytes": len,
516 + })),
517 + )
518 + .into_response();
519 + }
520 + }
521 +
522 + match ingest_zip(&state, &user, multipart).await {
523 + Ok(summary) => Json(serde_json::json!({
524 + "imported": summary.imported,
525 + "imported_count": summary.imported.len(),
526 + "duplicates": summary.duplicates,
527 + "skipped_non_image": summary.skipped_non_image,
528 + "skipped_too_large": summary.skipped_too_large,
529 + "failed": summary.failed,
530 + "truncated": summary.truncated,
531 + "message": summary.message(),
532 + }))
533 + .into_response(),
534 + Err(IngestError::TooLarge { actual, limit }) => (
535 + StatusCode::PAYLOAD_TOO_LARGE,
536 + Json(serde_json::json!({
537 + "error": "payload_too_large",
538 + "max_bytes": limit,
539 + "received_bytes": actual,
540 + })),
541 + )
542 + .into_response(),
543 + Err(IngestError::BadRequest(msg)) => (
544 + StatusCode::BAD_REQUEST,
545 + Json(serde_json::json!({"error": msg})),
546 + )
547 + .into_response(),
548 + Err(IngestError::Status(code)) => code.into_response(),
549 + }
550 +}
551 +
454 552 pub async fn api_show_upload(
455 553 State(state): State<AppState>,
456 554 user: UserContext,
@@ -495,6 +593,22 @@ impl From<StatusCode> for IngestError {
495 593 fn from(c: StatusCode) -> Self { IngestError::Status(c) }
496 594 }
497 595
596 +/// Map a multipart read failure onto our friendly variants. Axum surfaces the
597 +/// body-limit rejection as an opaque string, so match on it to turn "body too
598 +/// large" into the same TooLarge the explicit byte checks produce.
599 +fn multipart_err(state: &AppState, err: String, message: &str) -> IngestError {
600 + error!(error = %err, context = message, "multipart read failed");
601 + let lower = err.to_ascii_lowercase();
602 + if lower.contains("length limit") || lower.contains("too large") {
603 + IngestError::TooLarge {
604 + actual: state.config.max_upload_bytes + 1,
605 + limit: state.config.max_upload_bytes,
606 + }
607 + } else {
608 + IngestError::BadRequest(message.to_string())
609 + }
610 +}
611 +
498 612 async fn ingest_multipart(
499 613 state: &AppState,
500 614 user: &UserContext,
@@ -512,17 +626,7 @@ async fn ingest_multipart(
512 626 Ok(Some(f)) => f,
513 627 Ok(None) => break,
514 628 Err(e) => {
515 − // Map Axum's "body too large" rejection into our friendly
516 − // TooLarge variant so the user sees a nice message.
517 − let s = e.to_string();
518 − error!(error = %s, "multipart next_field failed");
519 − if s.to_ascii_lowercase().contains("length limit") || s.to_ascii_lowercase().contains("too large") {
520 − return Err(IngestError::TooLarge {
521 − actual: state.config.max_upload_bytes + 1,
522 − limit: state.config.max_upload_bytes,
523 − });
524 − }
525 − return Err(IngestError::BadRequest("Malformed multipart upload.".into()));
629 + return Err(multipart_err(state, e.to_string(), "Malformed multipart upload."));
526 630 }
527 631 };
528 632 let name = field.name().unwrap_or("").to_string();
@@ -530,18 +634,10 @@ async fn ingest_multipart(
530 634 "file" => {
531 635 original_name = field.file_name().map(|s| s.to_string());
532 636 content_type = field.content_type().map(|s| s.to_string());
533 − let bytes = field.bytes().await.map_err(|e| {
534 − let s = e.to_string();
535 − error!(error = %s, "multipart read file bytes failed");
536 − if s.to_ascii_lowercase().contains("length limit") || s.to_ascii_lowercase().contains("too large") {
537 − IngestError::TooLarge {
538 − actual: state.config.max_upload_bytes + 1,
539 − limit: state.config.max_upload_bytes,
540 − }
541 − } else {
542 − IngestError::BadRequest("Could not read uploaded file.".into())
543 − }
544 − })?;
637 + let bytes = field
638 + .bytes()
639 + .await
640 + .map_err(|e| multipart_err(state, e.to_string(), "Could not read uploaded file."))?;
545 641 if bytes.len() > state.config.max_upload_bytes {
546 642 return Err(IngestError::TooLarge {
547 643 actual: bytes.len(),
@@ -594,13 +690,28 @@ struct StoredUpload {
594 690 tags_raw: Option<String>,
595 691 }
596 692
597 −/// Shared tail of every ingest path: persist to storage, dedupe per user,
598 −/// insert the row, attach tags, and enqueue processing jobs.
693 +/// Shared tail of every single-file ingest path: persist, then enqueue the
694 +/// default job set. Duplicates short-circuit without re-enqueuing.
599 695 async fn store_upload(
600 696 state: &AppState,
601 697 user: &UserContext,
602 698 data: StoredUpload,
603 699 ) -> Result<Uuid, IngestError> {
700 + let (upload, duplicate) = persist_upload(state, user, data).await?;
701 + if !duplicate {
702 + enqueue_default_jobs(state, &upload).await;
703 + }
704 + Ok(upload.id)
705 +}
706 +
707 +/// Persist to storage, dedupe per user, insert the row and attach tags.
708 +/// Returns the row plus whether it was an existing duplicate — callers decide
709 +/// how to dispatch jobs (one-at-a-time, or batched for a zip import).
710 +async fn persist_upload(
711 + state: &AppState,
712 + user: &UserContext,
713 + data: StoredUpload,
714 +) -> Result<(uploads::Model, bool), IngestError> {
604 715 let StoredUpload {
605 716 bytes,
606 717 original_name,
@@ -638,7 +749,7 @@ async fn store_upload(
638 749 .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
639 750 {
640 751 info!(upload_id = %existing.id, "duplicate upload short-circuited");
641 − return Ok(existing.id);
752 + return Ok((existing, true));
642 753 }
643 754
644 755 let active = uploads::ActiveModel {
@@ -674,8 +785,7 @@ async fn store_upload(
674 785 }
675 786 }
676 787
677 − enqueue_default_jobs(state, &inserted).await;
678 − Ok(inserted.id)
788 + Ok((inserted, false))
679 789 }
680 790
681 791 /// Fetch a remote URL server-side and hand the bytes to [`store_upload`].
@@ -785,33 +895,334 @@ async fn ingest_url(
785 895 .await
786 896 }
787 897
788 −async fn enqueue_default_jobs(state: &AppState, upload: &uploads::Model) {
898 +// ============================================================================
899 +// Zip ingest — one archive of images in, many uploads out
900 +// ============================================================================
901 +
902 +/// Tally of a zip import. Everything the import chose *not* to store is
903 +/// counted so the user gets an honest report instead of a silent truncation.
904 +#[derive(Debug, Default)]
905 +pub struct ZipImportSummary {
906 + pub imported: Vec<Uuid>,
907 + /// Entries whose bytes this user had already uploaded.
908 + pub duplicates: usize,
909 + pub skipped_non_image: usize,
910 + pub skipped_too_large: usize,
911 + pub failed: usize,
912 + /// Hit MAX_ZIP_ENTRIES or MAX_ZIP_TOTAL_BYTES — the tail of the archive
913 + /// was never read.
914 + pub truncated: bool,
915 +}
916 +
917 +impl ZipImportSummary {
918 + pub fn message(&self) -> String {
919 + let n = self.imported.len();
920 + let mut msg = format!(
921 + "Imported {} image{} from the zip",
922 + n,
923 + if n == 1 { "" } else { "s" }
924 + );
925 + let mut notes: Vec<String> = Vec::new();
926 + if self.duplicates > 0 {
927 + notes.push(format!("{} already in the archive", self.duplicates));
928 + }
929 + if self.skipped_non_image > 0 {
930 + notes.push(format!("{} non-image entries skipped", self.skipped_non_image));
931 + }
932 + if self.skipped_too_large > 0 {
933 + notes.push(format!("{} over the size limit", self.skipped_too_large));
934 + }
935 + if self.failed > 0 {
936 + notes.push(format!("{} failed", self.failed));
937 + }
938 + if notes.is_empty() {
939 + msg.push('.');
940 + } else {
941 + msg = format!("{} ({}).", msg, notes.join(", "));
942 + }
943 + if self.truncated {
944 + msg.push_str(" Archive limits reached — the rest of the zip was not read.");
945 + }
946 + msg
947 + }
948 +}
949 +
950 +/// One image lifted out of a zip, on its way to [`persist_upload`].
951 +struct ZipEntryFile {
952 + name: String,
953 + content_type: Option<String>,
954 + bytes: Vec<u8>,
955 +}
956 +
957 +/// Counters the blocking scan owns; merged into the summary once it finishes.
958 +#[derive(Debug, Default)]
959 +struct ZipScan {
960 + skipped_non_image: usize,
961 + skipped_too_large: usize,
962 + truncated: bool,
963 +}
964 +
965 +/// Walk the archive on a blocking thread, handing each image to `tx` as it is
966 +/// decompressed. Streaming through a small channel — rather than extracting up
967 +/// front — keeps peak memory at the zip plus a handful of images, not the
968 +/// whole decompressed archive.
969 +fn scan_zip_blocking(
970 + zip_bytes: Vec<u8>,
971 + max_entry_bytes: usize,
972 + max_total_bytes: usize,
973 + max_entries: usize,
974 + tx: tokio::sync::mpsc::Sender<ZipEntryFile>,
975 +) -> Result<ZipScan, String> {
976 + use std::io::Read;
977 +
978 + let mut archive = zip::ZipArchive::new(std::io::Cursor::new(zip_bytes))
979 + .map_err(|e| format!("Could not read that zip file: {e}"))?;
980 +
981 + let mut scan = ZipScan::default();
982 + let mut total_bytes: usize = 0;
983 + let mut taken: usize = 0;
984 +
985 + for i in 0..archive.len() {
986 + if taken >= max_entries {
987 + scan.truncated = true;
988 + break;
989 + }
990 + let mut entry = match archive.by_index(i) {
991 + Ok(e) => e,
992 + Err(e) => {
993 + error!(error = %e, index = i, "zip entry unreadable");
994 + scan.skipped_non_image += 1;
995 + continue;
996 + }
997 + };
998 + if entry.is_dir() {
999 + continue;
1000 + }
1001 + // enclosed_name() rejects absolute paths and `..` traversal. Storage is
1002 + // content-addressed so the name never reaches the filesystem, but an
1003 + // untrustworthy name is reason enough to leave the entry alone.
1004 + let Some(path) = entry.enclosed_name() else {
1005 + scan.skipped_non_image += 1;
1006 + continue;
1007 + };
1008 + // Finder resource forks ride along in most macOS-made zips; so do
1009 + // dotfiles. Neither is something the user meant to upload, so they
1010 + // don't count against the skipped tally.
1011 + if path.components().any(|c| c.as_os_str() == "__MACOSX") {
1012 + continue;
1013 + }
1014 + let name = match path.file_name().and_then(|n| n.to_str()) {
1015 + Some(n) if !n.starts_with('.') => n.to_string(),
1016 + _ => continue,
1017 + };
1018 +
1019 + let content_type = mime_guess::from_path(&name).first().map(|m| m.to_string());
1020 + if !content_type
1021 + .as_deref()
1022 + .map(|ct| ct.starts_with("image/"))
1023 + .unwrap_or(false)
1024 + {
1025 + scan.skipped_non_image += 1;
1026 + continue;
1027 + }
1028 +
1029 + let declared = entry.size() as usize;
1030 + if declared > max_entry_bytes {
1031 + scan.skipped_too_large += 1;
1032 + continue;
1033 + }
1034 + if total_bytes.saturating_add(declared) > max_total_bytes {
1035 + scan.truncated = true;
1036 + break;
1037 + }
1038 +
1039 + // Read through a Take: the central directory's size is a claim, not a
1040 + // guarantee, and a crafted zip can keep producing bytes past it.
1041 + let mut buf = Vec::with_capacity(declared);
1042 + if let Err(e) = (&mut entry)
1043 + .take(max_entry_bytes as u64 + 1)
1044 + .read_to_end(&mut buf)
1045 + {
1046 + error!(error = %e, entry = %name, "zip entry decompression failed");
1047 + scan.skipped_non_image += 1;
1048 + continue;
1049 + }
1050 + if buf.len() > max_entry_bytes {
1051 + scan.skipped_too_large += 1;
1052 + continue;
1053 + }
1054 +
1055 + total_bytes += buf.len();
1056 + taken += 1;
1057 + // A send error means the consumer went away — stop decompressing.
1058 + if tx
1059 + .blocking_send(ZipEntryFile { name, content_type, bytes: buf })
1060 + .is_err()
1061 + {
1062 + break;
1063 + }
1064 + }
1065 +
1066 + Ok(scan)
1067 +}
1068 +
1069 +/// Ingest every image in an uploaded zip. `tags` and `item_description` from
1070 +/// the form, when present, are applied to every image in the archive.
1071 +async fn ingest_zip(
1072 + state: &AppState,
1073 + user: &UserContext,
1074 + mut multipart: Multipart,
1075 +) -> Result<ZipImportSummary, IngestError> {
1076 + let mut zip_bytes: Option<Vec<u8>> = None;
1077 + let mut item_description: Option<String> = None;
1078 + let mut tags_raw: Option<String> = None;
1079 +
1080 + loop {
1081 + let field = match multipart.next_field().await {
1082 + Ok(Some(f)) => f,
1083 + Ok(None) => break,
1084 + Err(e) => return Err(multipart_err(state, e.to_string(), "Malformed multipart upload.")),
1085 + };
1086 + match field.name().unwrap_or("") {
1087 + "file" => {
1088 + let bytes = field
1089 + .bytes()
1090 + .await
1091 + .map_err(|e| multipart_err(state, e.to_string(), "Could not read uploaded file."))?;
1092 + if bytes.len() > state.config.max_upload_bytes {
1093 + return Err(IngestError::TooLarge {
1094 + actual: bytes.len(),
1095 + limit: state.config.max_upload_bytes,
1096 + });
1097 + }
1098 + zip_bytes = Some(bytes.to_vec());
1099 + }
1100 + "item_description" => {
1101 + item_description = field.text().await.ok().filter(|s| !s.trim().is_empty());
1102 + }
1103 + "tags" => {
1104 + tags_raw = field.text().await.ok();
1105 + }
1106 + _ => {
1107 + let _ = field.bytes().await;
1108 + }
1109 + }
1110 + }
1111 +
1112 + let zip_bytes = zip_bytes.ok_or_else(|| IngestError::BadRequest("No zip file selected.".into()))?;
1113 +
1114 + let (tx, mut rx) = tokio::sync::mpsc::channel::<ZipEntryFile>(4);
1115 + let max_entry_bytes = state.config.max_upload_bytes;
1116 + let max_total_bytes = state.config.max_zip_total_bytes;
1117 + let max_entries = state.config.max_zip_entries;
1118 + let scanner = tokio::task::spawn_blocking(move || {
1119 + scan_zip_blocking(zip_bytes, max_entry_bytes, max_total_bytes, max_entries, tx)
1120 + });
1121 +
1122 + let mut summary = ZipImportSummary::default();
1123 + let mut stored: Vec<uploads::Model> = Vec::new();
1124 + while let Some(entry) = rx.recv().await {
1125 + let name = entry.name.clone();
1126 + let result = persist_upload(
1127 + state,
1128 + user,
1129 + StoredUpload {
1130 + bytes: entry.bytes,
1131 + original_name: entry.name,
1132 + content_type: entry.content_type,
1133 + item_name: None,
1134 + item_description: item_description.clone(),
1135 + tags_raw: tags_raw.clone(),
1136 + },
1137 + )
1138 + .await;
1139 + match result {
1140 + Ok((_, true)) => summary.duplicates += 1,
1141 + Ok((upload, false)) => {
1142 + summary.imported.push(upload.id);
1143 + stored.push(upload);
1144 + }
1145 + // One bad entry shouldn't abort the import — count it and carry on.
1146 + Err(_) => {
1147 + error!(entry = %name, "zip entry ingest failed");
1148 + summary.failed += 1;
1149 + }
1150 + }
1151 + }
1152 +
1153 + let scan = match scanner.await {
1154 + Ok(Ok(scan)) => scan,
1155 + // A malformed archive fails before the first entry, so nothing has been
1156 + // stored yet and this is safe to surface as a plain bad request.
1157 + Ok(Err(msg)) => return Err(IngestError::BadRequest(msg)),
1158 + Err(e) => {
1159 + error!(error = %e, "zip scan task panicked");
1160 + return Err(IngestError::Status(StatusCode::INTERNAL_SERVER_ERROR));
1161 + }
1162 + };
1163 + summary.skipped_non_image = scan.skipped_non_image;
1164 + summary.skipped_too_large = scan.skipped_too_large;
1165 + summary.truncated = scan.truncated;
1166 +
1167 + // One AMQP connection for the whole archive.
1168 + enqueue_default_jobs_many(state, &stored).await;
1169 +
1170 + info!(
1171 + imported = summary.imported.len(),
1172 + duplicates = summary.duplicates,
1173 + skipped_non_image = summary.skipped_non_image,
1174 + skipped_too_large = summary.skipped_too_large,
1175 + failed = summary.failed,
1176 + truncated = summary.truncated,
1177 + "zip import finished"
1178 + );
1179 + Ok(summary)
1180 +}
1181 +
1182 +fn default_job_kinds(upload: &uploads::Model) -> Vec<JobKind> {
789 1183 let is_image = upload
790 1184 .content_type
791 1185 .as_deref()
792 1186 .map(|ct| ct.starts_with("image/"))
793 1187 .unwrap_or(false);
794 1188
795 − let mut kinds: Vec<JobKind> = Vec::new();
796 1189 if is_image {
797 − kinds.push(JobKind::Thumbnail);
798 − kinds.push(JobKind::Ocr);
799 − kinds.push(JobKind::Describe);
1190 + vec![JobKind::Thumbnail, JobKind::Ocr, JobKind::Describe]
1191 + } else {
1192 + Vec::new()
800 1193 }
801 − if kinds.is_empty() {
1194 +}
1195 +
1196 +async fn enqueue_default_jobs(state: &AppState, upload: &uploads::Model) {
1197 + enqueue_default_jobs_many(state, std::slice::from_ref(upload)).await;
1198 +}
1199 +
1200 +/// Dispatch the default job set for a batch of fresh uploads over a single
1201 +/// AMQP connection. A zip import can land hundreds of rows at once; connecting
1202 +/// per upload would be one handshake per image.
1203 +async fn enqueue_default_jobs_many(state: &AppState, batch: &[uploads::Model]) {
1204 + let planned: Vec<(&uploads::Model, Vec<JobKind>)> = batch
1205 + .iter()
1206 + .map(|u| (u, default_job_kinds(u)))
1207 + .filter(|(_, kinds)| !kinds.is_empty())
1208 + .collect();
1209 + if planned.is_empty() {
802 1210 return;
803 1211 }
804 1212
805 1213 // Create the pending rows BEFORE publishing so the aggregate computed by
806 1214 // workers sees the full set up front.
807 − for kind in &kinds {
808 − if let Err(e) = jobs::upsert_pending(state.db.as_ref(), upload.id, *kind).await {
809 − error!(error = %e, ?kind, "upsert pending job row failed");
1215 + for (upload, kinds) in &planned {
1216 + for kind in kinds {
1217 + if let Err(e) = jobs::upsert_pending(state.db.as_ref(), upload.id, *kind).await {
1218 + error!(error = %e, ?kind, "upsert pending job row failed");
1219 + }
810 1220 }
811 1221 }
812 1222 // Bring uploads.processing_status in line with the pending rows.
1223 + let ids: Vec<Uuid> = planned.iter().map(|(u, _)| u.id).collect();
813 1224 let _ = uploads::Entity::update_many()
814 − .filter(uploads::Column::Id.eq(upload.id))
1225 + .filter(uploads::Column::Id.is_in(ids))
815 1226 .col_expr(uploads::Column::ProcessingStatus, sea_orm::sea_query::Expr::value(jobs::UPLOAD_STATUS_PROCESSING))
816 1227 .exec(state.db.as_ref())
817 1228 .await;
@@ -824,14 +1235,135 @@ async fn enqueue_default_jobs(state: &AppState, upload: &uploads::Model) {
824 1235 }
825 1236 };
826 1237
827 − for kind in kinds {
828 − let msg = JobMessage::new(kind, upload.id, upload.user_id);
829 − if let Err(e) = publish_job(&conn, &msg).await {
830 − error!(kind = ?kind, error = %e, "publish failed");
1238 + for (upload, kinds) in planned {
1239 + for kind in kinds {
1240 + let msg = JobMessage::new(kind, upload.id, upload.user_id);
1241 + if let Err(e) = publish_job(&conn, &msg).await {
1242 + error!(kind = ?kind, error = %e, "publish failed");
1243 + }
831 1244 }
832 1245 }
833 1246 }
834 1247
1248 +#[cfg(test)]
1249 +mod zip_unit_tests {
1250 + use super::*;
1251 + use std::io::Write;
1252 +
1253 + /// Build an in-memory zip from (name, bytes) pairs.
1254 + fn make_zip(entries: &[(&str, &[u8])]) -> Vec<u8> {
1255 + let mut w = zip::ZipWriter::new(std::io::Cursor::new(Vec::new()));
1256 + let opts = zip::write::SimpleFileOptions::default()
1257 + .compression_method(zip::CompressionMethod::Stored);
1258 + for (name, bytes) in entries {
1259 + w.start_file(*name, opts).expect("start_file");
1260 + w.write_all(bytes).expect("write");
1261 + }
1262 + w.finish().expect("finish").into_inner()
1263 + }
1264 +
1265 + /// Run the scan and drain the channel. Capacity comfortably exceeds the
1266 + /// entry count so the blocking sends never park.
1267 + fn scan(zip_bytes: Vec<u8>, max_entry: usize, max_total: usize, max_entries: usize)
1268 + -> (Vec<ZipEntryFile>, ZipScan)
1269 + {
1270 + let (tx, mut rx) = tokio::sync::mpsc::channel::<ZipEntryFile>(64);
1271 + let scan = scan_zip_blocking(zip_bytes, max_entry, max_total, max_entries, tx)
1272 + .expect("scan should succeed");
1273 + let mut got = Vec::new();
1274 + while let Ok(entry) = rx.try_recv() {
1275 + got.push(entry);
1276 + }
1277 + (got, scan)
1278 + }
1279 +
1280 + #[test]
1281 + fn picks_images_and_skips_the_rest() {
1282 + let zip = make_zip(&[
1283 + ("a.png", b"fake-png"),
1284 + ("notes.txt", b"hello"),
1285 + ("sub/dir/b.jpg", b"fake-jpg"),
1286 + ("__MACOSX/._a.png", b"resource fork"),
1287 + (".hidden.png", b"dotfile"),
1288 + ]);
1289 + let (got, scan) = scan(zip, 1024, 1024 * 1024, 100);
1290 +
1291 + let names: Vec<&str> = got.iter().map(|e| e.name.as_str()).collect();
1292 + assert_eq!(names, vec!["a.png", "b.jpg"], "nested paths flatten to the basename");
1293 + assert_eq!(got[0].content_type.as_deref(), Some("image/png"));
1294 + // notes.txt is the only genuine skip — Finder forks and dotfiles are
1295 + // junk the user never chose to upload, so they aren't tallied.
1296 + assert_eq!(scan.skipped_non_image, 1);
1297 + assert_eq!(scan.skipped_too_large, 0);
1298 + assert!(!scan.truncated);
1299 + }
1300 +
1301 + #[test]
1302 + fn oversize_entries_are_skipped_not_fatal() {
1303 + let zip = make_zip(&[("small.png", b"tiny"), ("big.png", &[0u8; 512])]);
1304 + let (got, scan) = scan(zip, 64, 1024 * 1024, 100);
1305 +
1306 + assert_eq!(got.len(), 1);
1307 + assert_eq!(got[0].name, "small.png");
1308 + assert_eq!(scan.skipped_too_large, 1);
1309 + }
1310 +
1311 + #[test]
1312 + fn entry_cap_truncates() {
1313 + let zip = make_zip(&[("a.png", b"one"), ("b.png", b"two"), ("c.png", b"three")]);
1314 + let (got, scan) = scan(zip, 1024, 1024 * 1024, 2);
1315 +
1316 + assert_eq!(got.len(), 2);
1317 + assert!(scan.truncated, "hitting MAX_ZIP_ENTRIES must be reported");
1318 + }
1319 +
1320 + #[test]
1321 + fn total_byte_cap_truncates() {
1322 + let zip = make_zip(&[("a.png", &[0u8; 100]), ("b.png", &[0u8; 100])]);
1323 + // Room for the first entry only.
1324 + let (got, scan) = scan(zip, 1024, 150, 100);
1325 +
1326 + assert_eq!(got.len(), 1);
1327 + assert!(scan.truncated);
1328 + }
1329 +
1330 + #[test]
1331 + fn garbage_is_not_a_zip() {
1332 + let (tx, _rx) = tokio::sync::mpsc::channel::<ZipEntryFile>(4);
1333 + let err = scan_zip_blocking(b"definitely not a zip".to_vec(), 1024, 1024, 10, tx)
1334 + .expect_err("garbage must be rejected");
1335 + assert!(err.starts_with("Could not read that zip file"), "got: {err}");
1336 + }
1337 +
1338 + #[test]
1339 + fn summary_message_reports_every_bucket() {
1340 + let summary = ZipImportSummary {
1341 + imported: vec![Uuid::now_v7()],
1342 + duplicates: 2,
1343 + skipped_non_image: 3,
1344 + skipped_too_large: 1,
1345 + failed: 1,
1346 + truncated: true,
1347 + };
1348 + let msg = summary.message();
1349 + assert!(msg.starts_with("Imported 1 image from the zip ("), "got: {msg}");
1350 + assert!(msg.contains("2 already in the archive"));
1351 + assert!(msg.contains("3 non-image entries skipped"));
1352 + assert!(msg.contains("1 over the size limit"));
1353 + assert!(msg.contains("1 failed"));
1354 + assert!(msg.contains("the rest of the zip was not read"));
1355 + }
1356 +
1357 + #[test]
1358 + fn summary_message_stays_plain_when_clean() {
1359 + let summary = ZipImportSummary {
1360 + imported: vec![Uuid::now_v7(), Uuid::now_v7()],
1361 + ..Default::default()
1362 + };
1363 + assert_eq!(summary.message(), "Imported 2 images from the zip.");
1364 + }
1365 +}
1366 +
835 1367 // ============================================================================
836 1368 // Dedicated search page
837 1369 // ============================================================================
modified templates/uploads/list.html
+11 −0
@@ -54,6 +54,17 @@
54 54 <button class="btn btn-primary btn-sm" type="submit">Upload</button>
55 55 </form>
56 56
57 + <form method="post" action="/uploads/zip" enctype="multipart/form-data" class="card bg-base-200 p-4 space-y-2">
58 + <div class="flex items-center justify-between">
59 + <h3 class="font-bold">Bulk upload (zip)</h3>
60 + <span class="text-xs opacity-60">Every image inside is ingested and processed</span>
61 + </div>
62 + <input class="file-input file-input-bordered file-input-sm w-full" type="file" name="file" accept=".zip,application/zip,application/x-zip-compressed" required/>
63 + <textarea class="textarea textarea-bordered textarea-sm w-full" name="item_description" placeholder="Optional description — applied to every image"></textarea>
64 + <input class="input input-bordered input-sm w-full" type="text" name="tags" placeholder="Optional tags — applied to every image"/>
65 + <button class="btn btn-primary btn-sm" type="submit">Upload zip</button>
66 + </form>
67 +
57 68 {% if uploads|length == 0 %}
58 69 <p class="opacity-70">No uploads matched.</p>
59 70 {% else %}

Parents: 48f426e