Skip to main content

max / makenotwork

22.4 KB · 585 lines History Blame Raw
1 //! Content file export: ZIP archive of audio, covers, videos, versions, and insertions.
2 //!
3 //! Writes the ZIP to a temporary file and uploads via S3 multipart upload,
4 //! so peak memory is O(single_file) regardless of total export size.
5
6 use std::fmt::Write as _;
7 use std::io::Write;
8
9 use axum::{
10 extract::{Query, State},
11 http::header::HeaderMap,
12 response::Response,
13 };
14 use serde::Deserialize;
15 use zip::write::SimpleFileOptions;
16
17 use crate::AppStorage;
18 use crate::background::BackgroundTx;
19 use crate::email::EmailClient;
20 use sqlx::PgPool;
21
22 use crate::{
23 auth::AuthUser,
24 db,
25 error::{AppError, Result, ResultExt},
26 helpers::is_htmx_request,
27 };
28
29 use super::export_error_html;
30
31 /// Max content exports running at once. Each can move up to 2 GB through a
32 /// synchronous zip on the blocking pool; without a cap a burst could saturate
33 /// the blocking pool and stall unrelated `spawn_blocking` work. Excess exports
34 /// queue on the semaphore instead.
35 const MAX_CONCURRENT_EXPORTS: usize = 3;
36 static EXPORT_LIMITER: tokio::sync::Semaphore =
37 tokio::sync::Semaphore::const_new(MAX_CONCURRENT_EXPORTS);
38
39 /// Query parameters for the content export endpoint.
40 #[derive(Deserialize)]
41 pub(in crate::routes::api) struct ContentExportQuery {
42 /// When set, only export files from this project (useful for large
43 /// catalogs or to stay within the 2GB per-export memory limit).
44 pub project_id: Option<db::ProjectId>,
45 }
46
47 /// Export content files as a ZIP archive uploaded to S3.
48 ///
49 /// Collects audio, covers, version downloads, and insertion clips,
50 /// bundles them with a README.txt manifest, uploads to S3 as a
51 /// temporary export, and returns a presigned download link.
52 ///
53 /// Pass `?project_id=<uuid>` to limit the export to a single project
54 /// (insertions are user-scoped and always excluded from per-project exports).
55 #[tracing::instrument(skip_all, name = "exports::export_content")]
56 pub(in crate::routes::api) async fn export_content(
57 State(db): State<PgPool>,
58 State(storage): State<AppStorage>,
59 State(email): State<EmailClient>,
60 State(bg): State<BackgroundTx>,
61 headers: HeaderMap,
62 Query(query): Query<ContentExportQuery>,
63 AuthUser(user): AuthUser,
64 ) -> Result<Response> {
65 let is_htmx = is_htmx_request(&headers);
66
67 let s3 = storage.s3.as_ref().ok_or_else(|| {
68 AppError::ServiceUnavailable("File storage is not configured".to_string())
69 })?;
70
71 // Collect all S3 keys from items, versions, and insertions. These are fast
72 // indexed reads, done up front so an empty export is reported immediately;
73 // the heavy download + zip + multipart upload runs in the background (below).
74 let item_keys = db::items::get_user_s3_keys(&db, user.id).await?;
75 let version_keys = db::versions::get_user_version_s3_keys(&db, user.id).await?;
76
77 // Build the list of (s3_key, zip_path, db_size) triples. The DB-known file
78 // size lets us enforce the per-file/total caps without a per-file S3 HEAD
79 // round-trip (these columns are written at upload-confirm time). `None` only
80 // for legacy rows missing the size; those fall through to the post-download
81 // total guard.
82 let mut files: Vec<(String, String, Option<i64>)> = Vec::new();
83
84 for item in &item_keys {
85 if let Some(pid) = query.project_id
86 && item.project_id != pid
87 {
88 continue;
89 }
90 let slug = item.project_slug.as_str();
91 let title = sanitize_filename(&item.title);
92 if let Some(ref key) = item.audio_s3_key {
93 let ext = extension_from_key(key);
94 files.push((
95 key.clone(),
96 format!("projects/{slug}/{title}.{ext}"),
97 item.audio_file_size_bytes,
98 ));
99 }
100 if let Some(ref key) = item.cover_s3_key {
101 let ext = extension_from_key(key);
102 files.push((
103 key.clone(),
104 format!("projects/{slug}/{title}-cover.{ext}"),
105 item.cover_file_size_bytes,
106 ));
107 }
108 if let Some(ref key) = item.video_s3_key {
109 let ext = extension_from_key(key);
110 files.push((
111 key.clone(),
112 format!("projects/{slug}/{title}-video.{ext}"),
113 item.video_file_size_bytes,
114 ));
115 }
116 }
117
118 for ver in &version_keys {
119 if let Some(pid) = query.project_id
120 && ver.project_id != pid
121 {
122 continue;
123 }
124 if let Some(ref key) = ver.s3_key {
125 let slug = ver.project_slug.as_str();
126 let title = sanitize_filename(&ver.item_title);
127 let fname = ver.file_name.as_deref().unwrap_or("file");
128 files.push((
129 key.clone(),
130 format!(
131 "projects/{}/{}/v{}-{}",
132 slug, title, ver.version_number, fname
133 ),
134 ver.file_size_bytes,
135 ));
136 }
137 }
138
139 // Insertions are user-scoped (not project-scoped), so only include
140 // them when exporting all content (no project_id filter).
141 if query.project_id.is_none() {
142 let insertions = db::content_insertions::list_insertions(&db, user.id).await?;
143 for ins in &insertions {
144 let ext = extension_from_key(&ins.storage_key);
145 let title = sanitize_filename(&ins.title);
146 files.push((
147 ins.storage_key.clone(),
148 format!("insertions/{title}.{ext}"),
149 Some(ins.file_size),
150 ));
151 }
152 }
153
154 // Exclude confirmed-malicious objects. The download paths refuse to serve a
155 // Quarantined object even to its own creator (downloads.rs); the export must
156 // honor the same invariant, since it collects raw S3 keys with no scan gate
157 // of its own (Run 20 Security). Filtering here (not in get_user_s3_keys)
158 // keeps that shared query intact for storage accounting, which must still see
159 // every key.
160 let export_keys: Vec<String> = files.iter().map(|(k, _, _)| k.clone()).collect();
161 let quarantined = db::scanning::quarantined_s3_keys(&db, &export_keys).await?;
162 if !quarantined.is_empty() {
163 tracing::warn!(
164 user_id = %user.id, dropped = quarantined.len(),
165 "excluding quarantined objects from content export"
166 );
167 files.retain(|(key, _, _)| !quarantined.contains(key));
168 }
169
170 if files.is_empty() {
171 if is_htmx {
172 return export_error_html("No content files to export.");
173 }
174 return Err(AppError::BadRequest(
175 "No content files to export.".to_string(),
176 ));
177 }
178
179 // Hand the heavy work to the background pool: download every file, build the
180 // ZIP on disk, multipart-upload it, and email a presigned link. The request
181 // returns now instead of holding a connection + export permit for the whole
182 // multi-GB job, a wedged S3 GET used to pin both indefinitely.
183 let s3 = s3.clone();
184 let email = email.clone();
185 let username = user.username.to_string();
186 let to_email = user.email.clone();
187 let to_name = user.display_name.clone();
188 let user_id = user.id;
189 bg.spawn("content_export", async move {
190 match build_content_export(&s3, user_id, &username, files).await {
191 Ok(download_url) => {
192 if let Err(e) = email
193 .send_content_export_ready(&to_email, to_name.as_deref(), &download_url)
194 .await
195 {
196 tracing::error!(error = ?e, "failed to send content-export-ready email");
197 }
198 }
199 Err(reason) => {
200 tracing::warn!(user_id = %user_id, reason = %reason, "content export failed");
201 if let Err(e) = email
202 .send_content_export_failed(&to_email, to_name.as_deref(), Some(&reason))
203 .await
204 {
205 tracing::error!(error = ?e, "failed to send content-export-failed email");
206 }
207 }
208 }
209 });
210
211 let message = format!(
212 "Preparing your content export. We'll email a download link to {} when it's ready.",
213 user.email
214 );
215 if is_htmx {
216 return super::export_pending_html(&message);
217 }
218 Response::builder()
219 .status(axum::http::StatusCode::ACCEPTED)
220 .body(message.into())
221 .context("build export accepted response")
222 }
223
224 /// Failure modes of [`spool_s3_object_to_file`].
225 enum SpoolError {
226 /// The object exceeded the per-file cap mid-stream (never fully buffered).
227 TooLarge,
228 /// An S3 or filesystem error; the message is log-only (never user-facing).
229 Io(String),
230 }
231
232 /// Stream an S3 object to `dest`, returning the number of bytes written. Aborts
233 /// with [`SpoolError::TooLarge`] the moment the running total exceeds
234 /// `max_bytes`, so a mis-sized object can't fill the disk. Peak memory is one
235 /// streaming chunk, this is what keeps the export's footprint at O(chunk)
236 /// rather than O(file).
237 async fn spool_s3_object_to_file(
238 s3: &std::sync::Arc<dyn crate::storage::StorageBackend>,
239 s3_key: &str,
240 dest: &std::path::Path,
241 max_bytes: u64,
242 ) -> std::result::Result<u64, SpoolError> {
243 use tokio::io::AsyncWriteExt;
244
245 let mut stream = s3
246 .download_stream(s3_key)
247 .await
248 .map_err(|e| SpoolError::Io(e.to_string()))?;
249 let mut file = tokio::fs::File::create(dest)
250 .await
251 .map_err(|e| SpoolError::Io(e.to_string()))?;
252 let mut written: u64 = 0;
253 while let Some(chunk) = stream
254 .try_next()
255 .await
256 .map_err(|e| SpoolError::Io(e.to_string()))?
257 {
258 written += chunk.len() as u64;
259 if written > max_bytes {
260 return Err(SpoolError::TooLarge);
261 }
262 file.write_all(&chunk)
263 .await
264 .map_err(|e| SpoolError::Io(e.to_string()))?;
265 }
266 file.flush()
267 .await
268 .map_err(|e| SpoolError::Io(e.to_string()))?;
269 Ok(written)
270 }
271
272 /// Build the content-export ZIP off the request path: download every file,
273 /// zip to a tempfile, multipart-upload, and return a 1-hour presigned download
274 /// URL. Holds the [`EXPORT_LIMITER`] permit for its lifetime so a burst can't
275 /// saturate the blocking pool. On any failure returns a user-facing reason
276 /// string (emailed to the creator); never panics the background task.
277 async fn build_content_export(
278 s3: &std::sync::Arc<dyn crate::storage::StorageBackend>,
279 user_id: db::UserId,
280 username: &str,
281 files: Vec<(String, String, Option<i64>)>,
282 ) -> std::result::Result<String, String> {
283 // Hold a concurrency permit for the lifetime of the export so a burst can't
284 // saturate the blocking pool. Acquired here (off the request path), so a
285 // queued export holds no DB connection while it waits.
286 let _export_permit = EXPORT_LIMITER
287 .acquire()
288 .await
289 .expect("export limiter semaphore is never closed");
290
291 let s3_clone = s3.clone();
292
293 let tmp_dir = tempfile::tempdir().map_err(|e| format!("create temp dir for export: {e}"))?;
294 let zip_path = tmp_dir.path().join("export.zip");
295
296 {
297 // The `zip` crate's IO is synchronous; a single `write_all` of up to
298 // 500 MB (compression is Stored, so this is raw disk IO) would stall a
299 // tokio worker. Every blocking zip operation below runs on the blocking
300 // pool via `spawn_blocking`; the writer is moved in and handed back out
301 // each step. S3 downloads stay async, and peak memory is still
302 // O(largest_single_file), one file is in RAM at a time.
303 let create_path = zip_path.clone();
304 let mut zip =
305 tokio::task::spawn_blocking(move || -> std::result::Result<_, std::io::Error> {
306 let zip_file = std::fs::File::create(&create_path)?;
307 Ok(zip::ZipWriter::new(std::io::BufWriter::new(zip_file)))
308 })
309 .await
310 .map_err(|e| format!("join zip create task: {e}"))?
311 .map_err(|e| format!("create export zip file: {e}"))?;
312
313 let options =
314 SimpleFileOptions::default().compression_method(zip::CompressionMethod::Stored);
315
316 let mut manifest: Vec<(String, i64)> = Vec::new();
317 let mut total_size: u64 = 0;
318 const MAX_TOTAL_SIZE: u64 = 2 * 1024 * 1024 * 1024; // 2 GB
319 const MAX_FILE_SIZE: u64 = 500 * 1024 * 1024; // 500 MB per file
320 let mut skipped: Vec<String> = Vec::new();
321
322 for (s3_key, zip_path_entry, db_size) in &files {
323 // Resolve the file size BEFORE downloading so a single huge object
324 // can't blow the heap before any guard fires. Prefer the DB column
325 // written at upload-confirm; fall back to a cheap S3 HEAD for legacy
326 // rows that predate it (a `None` size). A row whose size can't be
327 // resolved (HEAD error / object gone) is skipped rather than blindly
328 // buffered, without this, a legacy `None`-size object of arbitrary
329 // size landed fully in RAM before the post-download total check could
330 // fire (PERF-1).
331 let size = match db_size {
332 Some(s) => (*s).max(0) as u64,
333 None => match s3_clone.object_size(s3_key).await {
334 Ok(Some(s)) => s.max(0) as u64,
335 Ok(None) => {
336 tracing::warn!(s3_key = %s3_key, "export: object has no size (missing?), skipping");
337 skipped.push(zip_path_entry.clone());
338 continue;
339 }
340 Err(e) => {
341 tracing::warn!(s3_key = %s3_key, error = %e, "export: HEAD failed, skipping");
342 skipped.push(zip_path_entry.clone());
343 continue;
344 }
345 },
346 };
347 if size > MAX_FILE_SIZE {
348 skipped.push(format!(
349 "{zip_path_entry} (exceeds 500 MB per-file export cap)"
350 ));
351 continue;
352 }
353 if total_size + size > MAX_TOTAL_SIZE {
354 return Err("Content export exceeds the 2 GB limit. Try exporting a single project instead.".to_string());
355 }
356
357 // Stream the object to a per-file spool on disk instead of buffering
358 // the whole (up to 500 MB) file in RAM. Peak memory is one streaming
359 // chunk, not the full object, three concurrent exports previously
360 // held ~1.5 GB of file bytes between them. The scan pipeline streams
361 // large files the same way; the exporter now matches it (Run #5 Perf
362 // S2). The MAX_FILE_SIZE guard is re-enforced mid-stream as a backstop
363 // for legacy rows whose size couldn't be resolved above.
364 let part_path = tmp_dir.path().join(format!("part-{}", manifest.len()));
365 let part_bytes =
366 match spool_s3_object_to_file(&s3_clone, s3_key, &part_path, MAX_FILE_SIZE).await {
367 Ok(n) => n,
368 Err(SpoolError::TooLarge) => {
369 skipped.push(format!(
370 "{zip_path_entry} (exceeds 500 MB per-file export cap)"
371 ));
372 let _ = tokio::fs::remove_file(&part_path).await;
373 continue;
374 }
375 Err(SpoolError::Io(e)) => {
376 tracing::warn!("Failed to download S3 key {}: {}", s3_key, e);
377 skipped.push(zip_path_entry.clone());
378 let _ = tokio::fs::remove_file(&part_path).await;
379 continue;
380 }
381 };
382 total_size += part_bytes;
383 if total_size > MAX_TOTAL_SIZE {
384 return Err("Content export exceeds the 2 GB limit. Try exporting a single project instead.".to_string());
385 }
386 let file_size = part_bytes as i64;
387 // Copy the spooled file into the zip on the blocking pool (the zip
388 // crate's IO is synchronous). The reader streams from disk via
389 // io::copy, so RAM stays at the copy buffer size, not the file size.
390 let entry = zip_path_entry.clone();
391 let copy_path = part_path.clone();
392 zip = tokio::task::spawn_blocking(
393 move || -> std::result::Result<_, zip::result::ZipError> {
394 zip.start_file(&entry, options)?;
395 let mut reader = std::fs::File::open(&copy_path)?;
396 std::io::copy(&mut reader, &mut zip)?;
397 Ok(zip)
398 },
399 )
400 .await
401 .map_err(|e| format!("join zip write task: {e}"))?
402 .map_err(|e| format!("write file into export zip: {e}"))?;
403 let _ = tokio::fs::remove_file(&part_path).await;
404 manifest.push((zip_path_entry.clone(), file_size));
405 }
406
407 if manifest.is_empty() {
408 return Err(
409 "Could not download any files from storage. Please try again later.".to_string(),
410 );
411 }
412
413 // Build README.txt as the last ZIP entry (cheap string work, async side)
414 let now = chrono::Utc::now();
415 let mut readme = format!(
416 "Makenotwork Content Export\n\
417 Creator: {}\n\
418 Exported: {}\n\
419 Files: {}\n\n\
420 Manifest:\n",
421 username,
422 now.format("%Y-%m-%d %H:%M:%S UTC"),
423 manifest.len(),
424 );
425 for (path, size) in &manifest {
426 writeln!(
427 readme,
428 " {path} ({})",
429 crate::helpers::format_file_size(*size)
430 )
431 .unwrap();
432 }
433 if !skipped.is_empty() {
434 writeln!(
435 readme,
436 "\nSkipped ({} files could not be downloaded):",
437 skipped.len()
438 )
439 .unwrap();
440 for path in &skipped {
441 writeln!(readme, " {path}").unwrap();
442 }
443 }
444 readme.push_str("\nNote: Git repositories are not included in this export.\n");
445 readme.push_str(
446 "Clone them separately: git clone https://makenot.work/git/<username>/<repo>.git\n",
447 );
448
449 // Append README, finalize the central directory, and flush the buffer to
450 // disk, all blocking, off the runtime before the upload reads the file.
451 tokio::task::spawn_blocking(move || -> std::result::Result<(), zip::result::ZipError> {
452 zip.start_file("README.txt", options)?;
453 zip.write_all(readme.as_bytes())?;
454 let buf = zip.finish()?;
455 // Flush BufWriter so all bytes hit the OS file before we upload it.
456 buf.into_inner()
457 .map_err(std::io::IntoInnerError::into_error)?;
458 Ok(())
459 })
460 .await
461 .map_err(|e| format!("join zip finalize task: {e}"))?
462 .map_err(|e| format!("finalize export zip: {e}"))?;
463 }
464
465 // Upload ZIP to S3 via multipart upload (streams from disk in 10 MB parts)
466 let timestamp = chrono::Utc::now().format("%Y%m%d-%H%M%S");
467 let export_key =
468 crate::storage::S3Client::generate_content_export_key(user_id, &timestamp.to_string());
469 if let Err(e) = s3
470 .upload_multipart(&export_key, "application/zip", &zip_path)
471 .await
472 {
473 tracing::error!(error = ?e, "Failed to upload content export ZIP to S3");
474 return Err("Failed to upload the export to storage.".to_string());
475 }
476
477 // Generate presigned download URL (1 hour)
478 let download_url = s3
479 .presign_download(&export_key, Some(3600))
480 .await
481 .map_err(|e| {
482 tracing::error!(error = ?e, "Failed to generate presigned URL for content export");
483 "Export created but the download link could not be generated.".to_string()
484 })?;
485
486 Ok(download_url)
487 }
488
489 /// Extract file extension from an S3 key (e.g. "user/item/audio/track.mp3" -> "mp3").
490 fn extension_from_key(key: &str) -> &str {
491 // Extract from the basename only: a dot earlier in the path (e.g. a user
492 // handle like `alice.dev/...`) must not be mistaken for the extension, and an
493 // extensionless key must fall back to "bin" rather than returning the whole
494 // path as a bogus extension (audit Run 13).
495 let basename = key.rsplit('/').next().unwrap_or(key);
496 match basename.rsplit_once('.') {
497 Some((stem, ext)) if !stem.is_empty() && !ext.is_empty() => ext,
498 _ => "bin",
499 }
500 }
501
502 /// Sanitize a title for use as a filename in the ZIP archive.
503 fn sanitize_filename(name: &str) -> String {
504 name.chars()
505 .map(|c| {
506 if c.is_alphanumeric() || c == '-' || c == '_' || c == ' ' {
507 c
508 } else {
509 '_'
510 }
511 })
512 .collect::<String>()
513 .trim()
514 .to_string()
515 }
516
517 #[cfg(test)]
518 mod tests {
519 use super::*;
520
521 #[test]
522 fn extension_from_key_mp3() {
523 assert_eq!(extension_from_key("user/item/audio/track.mp3"), "mp3");
524 }
525
526 #[test]
527 fn extension_from_key_nested_path() {
528 assert_eq!(extension_from_key("a/b/c/file.tar.gz"), "gz");
529 }
530
531 #[test]
532 fn extension_from_key_no_dot_returns_bin() {
533 // An extensionless key falls back to "bin", never the whole path.
534 assert_eq!(extension_from_key("user/item/audio/noext"), "bin");
535 }
536
537 #[test]
538 fn extension_from_key_dot_in_path_not_basename() {
539 // A dot earlier in the path is not the extension.
540 assert_eq!(extension_from_key("alice.dev/item/noext"), "bin");
541 assert_eq!(extension_from_key("alice.dev/item/track.wav"), "wav");
542 }
543
544 #[test]
545 fn extension_from_key_empty_returns_bin() {
546 assert_eq!(extension_from_key(""), "bin");
547 }
548
549 #[test]
550 fn extension_from_key_dot_only() {
551 // Trailing dot → empty ext → "bin".
552 assert_eq!(extension_from_key("file."), "bin");
553 }
554
555 #[test]
556 fn sanitize_filename_passthrough() {
557 assert_eq!(sanitize_filename("My Track"), "My Track");
558 }
559
560 #[test]
561 fn sanitize_filename_special_chars() {
562 assert_eq!(sanitize_filename("hello/world:2"), "hello_world_2");
563 }
564
565 #[test]
566 fn sanitize_filename_preserves_hyphens_underscores() {
567 assert_eq!(sanitize_filename("my-file_name"), "my-file_name");
568 }
569
570 #[test]
571 fn sanitize_filename_trims_whitespace() {
572 assert_eq!(sanitize_filename(" padded "), "padded");
573 }
574
575 #[test]
576 fn sanitize_filename_empty() {
577 assert_eq!(sanitize_filename(""), "");
578 }
579
580 #[test]
581 fn sanitize_filename_all_special() {
582 assert_eq!(sanitize_filename("@#$%"), "____");
583 }
584 }
585