1use crate::settings::SettingsReadGuard;
2use aws_sdk_s3::{
3 Client as S3Client,
4 config::{Config as S3Config, Credentials, Region, retry::RetryConfig, timeout::TimeoutConfig},
5 primitives::ByteStream,
6 types::{CompletedMultipartUpload, CompletedPart},
7};
8use compact_str::ToCompactString;
9use serde::{Deserialize, Serialize};
10use std::{path::Path, sync::Arc};
11use tokio::io::{AsyncReadExt, AsyncWriteExt};
12use tokio_util::bytes::{Bytes, BytesMut};
13use utoipa::ToSchema;
14
15#[derive(ToSchema, Deserialize, Serialize)]
16pub struct StorageAsset {
17 pub name: compact_str::CompactString,
18 pub url: String,
19 pub size: u64,
20 pub is_directory: bool,
21 pub created: chrono::DateTime<chrono::Utc>,
22}
23
24const SEARCH_SCAN_LIMIT: usize = 10_000;
25
26pub fn content_type(path: impl AsRef<str>) -> &'static str {
28 let extension = match Path::new(path.as_ref())
29 .extension()
30 .and_then(|e| e.to_str())
31 {
32 Some(extension) => extension.to_ascii_lowercase(),
33 None => return "application/octet-stream",
34 };
35
36 match extension.as_str() {
37 "png" => "image/png",
38 "jpg" | "jpeg" => "image/jpeg",
39 "gif" => "image/gif",
40 "webp" => "image/webp",
41 "avif" => "image/avif",
42 "bmp" => "image/bmp",
43 "ico" => "image/x-icon",
44 "svg" => "image/svg+xml",
45 "woff" => "font/woff",
46 "woff2" => "font/woff2",
47 "ttf" => "font/ttf",
48 "otf" => "font/otf",
49 "mp4" => "video/mp4",
50 "webm" => "video/webm",
51 "ogg" => "audio/ogg",
52 "mp3" => "audio/mpeg",
53 "wav" => "audio/wav",
54 "pdf" => "application/pdf",
55 "xml" => "application/xml",
56 "txt" | "csv" => "text/plain",
57 _ => "application/octet-stream",
58 }
59}
60
61#[inline]
63pub fn content_type_requires_attachment(content_type: &str) -> bool {
64 matches!(content_type, "image/svg+xml" | "application/xml")
65}
66
67fn get_s3_client(
68 access_key: &str,
69 secret_key: &str,
70 region: &str,
71 endpoint: &str,
72 path_style: bool,
73) -> Result<S3Client, anyhow::Error> {
74 let credentials = Credentials::new(access_key, secret_key, None, None, "calagopus-static");
75
76 let timeout_config = TimeoutConfig::builder()
77 .connect_timeout(std::time::Duration::from_secs(10))
78 .build();
79
80 let config = S3Config::builder()
81 .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
82 .credentials_provider(credentials)
83 .region(Region::new(region.to_string()))
84 .endpoint_url(endpoint)
85 .force_path_style(path_style)
86 .timeout_config(timeout_config)
87 .retry_config(RetryConfig::standard())
88 .build();
89
90 Ok(S3Client::from_conf(config))
91}
92
93pub struct StorageUrlRetriever<'a> {
94 settings: SettingsReadGuard<'a>,
95}
96
97impl<'a> StorageUrlRetriever<'a> {
98 pub fn new(settings: SettingsReadGuard<'a>) -> Self {
99 Self { settings }
100 }
101
102 pub fn get_settings(&self) -> &super::settings::AppSettings {
103 &self.settings
104 }
105
106 pub fn get_url(&self, path: impl AsRef<str>) -> String {
107 match &self.settings.storage_driver {
108 super::settings::StorageDriver::Filesystem { .. } => {
109 format!(
110 "{}/{}",
111 self.settings.app.url.trim_end_matches('/'),
112 path.as_ref()
113 )
114 }
115 super::settings::StorageDriver::S3 { public_url, .. } => {
116 format!("{}/{}", public_url.trim_end_matches('/'), path.as_ref())
117 }
118 }
119 }
120}
121
122pub struct Storage {
123 settings: Arc<super::settings::Settings>,
124}
125
126impl Storage {
127 pub fn new(settings: Arc<super::settings::Settings>) -> Self {
128 Self { settings }
129 }
130
131 pub async fn retrieve_urls(&self) -> Result<StorageUrlRetriever<'_>, anyhow::Error> {
132 let settings = self.settings.get().await?;
133
134 Ok(StorageUrlRetriever::new(settings))
135 }
136
137 pub async fn remove(&self, path: Option<impl AsRef<str>>) -> Result<(), anyhow::Error> {
138 let path = match path {
139 Some(path) => path,
140 None => return Ok(()),
141 };
142 let path = path.as_ref();
143
144 if path.is_empty() || path.contains("..") || path.starts_with('/') {
145 return Err(anyhow::anyhow!("invalid path"));
146 }
147
148 let settings = self.settings.get().await?;
149
150 tracing::debug!(path, "removing file");
151
152 match &settings.storage_driver {
153 super::settings::StorageDriver::Filesystem { path: base_path } => {
154 let base_filesystem =
155 match crate::cap::CapFilesystem::async_new(base_path.into()).await {
156 Ok(base_filesystem) => base_filesystem,
157 Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(()),
158 Err(err) => return Err(err.into()),
159 };
160 drop(settings);
161
162 if let Err(err) = base_filesystem.async_remove_file(&path).await
163 && err
164 .downcast_ref::<std::io::Error>()
165 .is_none_or(|e| e.kind() != std::io::ErrorKind::NotFound)
166 {
167 return Err(err);
168 }
169
170 if let Some(parent) = Path::new(path).parent().map(|p| p.to_path_buf()) {
171 tokio::spawn(async move {
172 tokio::time::sleep(std::time::Duration::from_secs(10)).await;
173
174 let mut directory = match base_filesystem.async_read_dir(&parent).await {
175 Ok(directory) => directory,
176 Err(_) => return,
177 };
178
179 if directory.next_entry().await.is_none() {
180 base_filesystem.async_remove_dir(parent).await.ok();
181 }
182 });
183 }
184 }
185 super::settings::StorageDriver::S3 {
186 access_key,
187 secret_key,
188 bucket,
189 region,
190 endpoint,
191 path_style,
192 ..
193 } => {
194 let s3_client =
195 get_s3_client(access_key, secret_key, region, endpoint, *path_style)?;
196 let bucket = bucket.clone();
197 drop(settings);
198
199 s3_client
200 .delete_object()
201 .bucket(bucket)
202 .key(path)
203 .send()
204 .await?;
205 }
206 }
207
208 Ok(())
209 }
210
211 pub async fn store(
212 &self,
213 path: impl AsRef<str>,
214 data: impl tokio::io::AsyncRead + Unpin,
215 content_type: impl AsRef<str>,
216 ) -> Result<u64, anyhow::Error> {
217 self.store_with_cache_control(path, data, content_type, None)
218 .await
219 }
220
221 pub async fn store_with_cache_control(
225 &self,
226 path: impl AsRef<str>,
227 data: impl tokio::io::AsyncRead + Unpin,
228 content_type: impl AsRef<str>,
229 cache_control: Option<&str>,
230 ) -> Result<u64, anyhow::Error> {
231 let path = path.as_ref();
232 let content_type = content_type.as_ref();
233
234 if path.is_empty() || path.contains("..") || path.starts_with('/') {
235 return Err(anyhow::anyhow!("invalid path"));
236 }
237
238 let settings = self.settings.get().await?;
239
240 tracing::debug!(path, content_type, "storing file");
241
242 match &settings.storage_driver {
243 super::settings::StorageDriver::Filesystem { path: base_path } => {
244 tokio::fs::create_dir_all(base_path).await?;
245
246 let base_filesystem =
247 crate::cap::CapFilesystem::async_new(base_path.into()).await?;
248 drop(settings);
249
250 if let Some(parent) = Path::new(path).parent() {
251 base_filesystem.async_create_dir_all(parent).await?;
252 }
253
254 let mut file = base_filesystem.async_create(path).await?;
255 let mut data = data;
256 let bytes = tokio::io::copy(&mut data, &mut file).await?;
257
258 file.shutdown().await?;
259 Ok(bytes)
260 }
261 super::settings::StorageDriver::S3 {
262 access_key,
263 secret_key,
264 bucket,
265 region,
266 endpoint,
267 path_style,
268 ..
269 } => {
270 let s3_client =
271 get_s3_client(access_key, secret_key, region, endpoint, *path_style)?;
272 let bucket = bucket.clone();
273 drop(settings);
274
275 upload_multipart(&s3_client, &bucket, path, content_type, cache_control, data).await
276 }
277 }
278 }
279
280 pub async fn list(
281 &self,
282 base: impl AsRef<str>,
283 directory: impl AsRef<str>,
284 page: usize,
285 per_page: usize,
286 ) -> Result<crate::models::Pagination<StorageAsset>, anyhow::Error> {
287 let base = base.as_ref();
288 let directory = directory.as_ref();
289
290 if base.is_empty() || base.contains("..") || base.starts_with('/') {
291 return Err(anyhow::anyhow!("invalid base path"));
292 }
293 if !directory.is_empty()
294 && (directory.contains("..") || directory.starts_with('/') || directory.ends_with('/'))
295 {
296 return Err(anyhow::anyhow!("invalid directory path"));
297 }
298
299 let settings = self.settings.get().await?;
300
301 match &settings.storage_driver {
302 super::settings::StorageDriver::Filesystem { path: base_path } => {
303 let dir_path = if directory.is_empty() {
304 Path::new(base_path).join(base)
305 } else {
306 Path::new(base_path).join(base).join(directory)
307 };
308
309 let base_filesystem = match crate::cap::CapFilesystem::async_new(dir_path).await {
310 Ok(base_filesystem) => base_filesystem,
311 Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
312 return Ok(crate::models::Pagination {
313 total: 0,
314 per_page: per_page as i64,
315 page: page as i64,
316 data: Vec::new(),
317 });
318 }
319 Err(err) => return Err(err.into()),
320 };
321 drop(settings);
322
323 let mut dir_reader = base_filesystem.async_read_dir("").await?;
324 let mut raw_dirs: Vec<String> = Vec::new();
325 let mut raw_files: Vec<String> = Vec::new();
326
327 while let Some(Ok((is_dir, name))) = dir_reader.next_entry().await {
328 if is_dir {
329 raw_dirs.push(name);
330 } else {
331 raw_files.push(name);
332 }
333 }
334
335 raw_dirs.sort_unstable();
336 raw_files.sort_unstable();
337
338 let total = (raw_dirs.len() + raw_files.len()) as i64;
339 let start = (page - 1) * per_page;
340
341 let storage_url_retriever = self.retrieve_urls().await?;
342
343 let mut entries = Vec::new();
344
345 for (is_dir, name) in raw_dirs
346 .into_iter()
347 .map(|n| (true, n))
348 .chain(raw_files.into_iter().map(|n| (false, n)))
349 .skip(start)
350 .take(per_page)
351 {
352 let full_name = if directory.is_empty() {
353 name.clone()
354 } else {
355 format!("{directory}/{name}")
356 };
357
358 let (size, created) = if is_dir {
359 (0u64, chrono::DateTime::<chrono::Utc>::default())
360 } else {
361 let metadata = match base_filesystem.async_metadata(&name).await {
362 Ok(m) => m,
363 Err(_) => continue,
364 };
365 let created = metadata
366 .created()
367 .or_else(|_| metadata.modified())?
368 .into_std()
369 .into();
370 (metadata.len(), created)
371 };
372
373 entries.push(StorageAsset {
374 url: storage_url_retriever.get_url(format!("{base}/{full_name}")),
375 name: full_name.to_compact_string(),
376 size,
377 is_directory: is_dir,
378 created,
379 });
380 }
381
382 Ok(crate::models::Pagination {
383 total,
384 per_page: per_page as i64,
385 page: page as i64,
386 data: entries,
387 })
388 }
389 super::settings::StorageDriver::S3 {
390 access_key,
391 secret_key,
392 bucket,
393 region,
394 endpoint,
395 path_style,
396 ..
397 } => {
398 let s3_client =
399 get_s3_client(access_key, secret_key, region, endpoint, *path_style)?;
400 let bucket_name = bucket.clone();
401 drop(settings);
402
403 let s3_prefix = if directory.is_empty() {
404 format!("{base}/")
405 } else {
406 format!("{base}/{directory}/")
407 };
408 let strip_prefix = format!("{base}/");
409
410 let storage_url_retriever = self.retrieve_urls().await?;
411
412 let mut dirs = Vec::new();
413 let mut files = Vec::new();
414
415 let mut paginator = s3_client
416 .list_objects_v2()
417 .bucket(&*bucket_name)
418 .prefix(&s3_prefix)
419 .delimiter("/")
420 .into_paginator()
421 .send();
422
423 while let Some(result) = paginator.next().await {
424 let page = result?;
425
426 for cp in page.common_prefixes() {
427 let Some(prefix) = cp.prefix() else { continue };
428 let name = prefix
429 .strip_prefix(&strip_prefix)
430 .unwrap_or(prefix)
431 .trim_end_matches('/')
432 .to_compact_string();
433 dirs.push(StorageAsset {
434 url: storage_url_retriever.get_url(prefix),
435 name,
436 size: 0,
437 is_directory: true,
438 created: chrono::DateTime::<chrono::Utc>::default(),
439 });
440 }
441
442 for entry in page.contents() {
443 let Some(key) = entry.key() else { continue };
444 if key == s3_prefix {
445 continue;
446 }
447 let name = key
448 .strip_prefix(&strip_prefix)
449 .unwrap_or(key)
450 .to_compact_string();
451 let size = entry.size().unwrap_or(0).max(0) as u64;
452 let created = entry
453 .last_modified()
454 .and_then(|dt| {
455 chrono::DateTime::<chrono::Utc>::from_timestamp(
456 dt.secs(),
457 dt.subsec_nanos(),
458 )
459 })
460 .unwrap_or_default();
461
462 files.push(StorageAsset {
463 url: storage_url_retriever.get_url(key),
464 name,
465 size,
466 is_directory: false,
467 created,
468 });
469 }
470 }
471
472 let total = (dirs.len() + files.len()) as i64;
473 let start = (page - 1) * per_page;
474
475 Ok(crate::models::Pagination {
476 total,
477 per_page: per_page as i64,
478 page: page as i64,
479 data: dirs
480 .into_iter()
481 .chain(files)
482 .skip(start)
483 .take(per_page)
484 .collect(),
485 })
486 }
487 }
488 }
489
490 pub async fn search(
491 &self,
492 base: impl AsRef<str>,
493 directory: impl AsRef<str>,
494 search: impl AsRef<str>,
495 limit: usize,
496 ) -> Result<Vec<StorageAsset>, anyhow::Error> {
497 let base = base.as_ref();
498 let directory = directory.as_ref();
499 let search = search.as_ref().trim().to_lowercase();
500
501 if base.is_empty() || base.contains("..") || base.starts_with('/') {
502 return Err(anyhow::anyhow!("invalid base path"));
503 }
504 if !directory.is_empty()
505 && (directory.contains("..") || directory.starts_with('/') || directory.ends_with('/'))
506 {
507 return Err(anyhow::anyhow!("invalid directory path"));
508 }
509 if search.is_empty() {
510 return Err(anyhow::anyhow!("empty search"));
511 }
512
513 let settings = self.settings.get().await?;
514
515 let mut dirs = Vec::new();
516 let mut files = Vec::new();
517 let mut scanned = 0;
518
519 match &settings.storage_driver {
520 super::settings::StorageDriver::Filesystem { path: base_path } => {
521 let dir_path = if directory.is_empty() {
522 Path::new(base_path).join(base)
523 } else {
524 Path::new(base_path).join(base).join(directory)
525 };
526
527 let base_filesystem = match crate::cap::CapFilesystem::async_new(dir_path).await {
528 Ok(base_filesystem) => base_filesystem,
529 Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
530 return Ok(Vec::new());
531 }
532 Err(err) => return Err(err.into()),
533 };
534 drop(settings);
535
536 let storage_url_retriever = self.retrieve_urls().await?;
537 let mut walker = base_filesystem.async_walk_dir("").await?;
538
539 while let Some(Ok((is_dir, path))) = walker.next_entry().await {
540 scanned += 1;
541 if scanned > SEARCH_SCAN_LIMIT {
542 break;
543 }
544
545 let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
546 continue;
547 };
548 if !name.to_lowercase().contains(&search) {
549 continue;
550 }
551
552 let relative = path.to_string_lossy();
553 let full_name = if directory.is_empty() {
554 relative.to_compact_string()
555 } else {
556 format!("{directory}/{relative}").to_compact_string()
557 };
558
559 let (size, created) = if is_dir {
560 (0u64, chrono::DateTime::<chrono::Utc>::default())
561 } else {
562 let metadata = match base_filesystem.async_metadata(&path).await {
563 Ok(m) => m,
564 Err(_) => continue,
565 };
566 let created = metadata
567 .created()
568 .or_else(|_| metadata.modified())?
569 .into_std()
570 .into();
571 (metadata.len(), created)
572 };
573
574 let asset = StorageAsset {
575 url: storage_url_retriever.get_url(format!("{base}/{full_name}")),
576 name: full_name,
577 size,
578 is_directory: is_dir,
579 created,
580 };
581
582 if is_dir { &mut dirs } else { &mut files }.push(asset);
583 }
584 }
585 super::settings::StorageDriver::S3 {
586 access_key,
587 secret_key,
588 bucket,
589 region,
590 endpoint,
591 path_style,
592 ..
593 } => {
594 let s3_client =
595 get_s3_client(access_key, secret_key, region, endpoint, *path_style)?;
596 let bucket_name = bucket.clone();
597 drop(settings);
598
599 let s3_prefix = if directory.is_empty() {
600 format!("{base}/")
601 } else {
602 format!("{base}/{directory}/")
603 };
604 let strip_prefix = format!("{base}/");
605
606 let storage_url_retriever = self.retrieve_urls().await?;
607 let mut seen_dirs = std::collections::HashSet::new();
608
609 let mut paginator = s3_client
610 .list_objects_v2()
611 .bucket(&*bucket_name)
612 .prefix(&s3_prefix)
613 .into_paginator()
614 .send();
615
616 'scan: while let Some(result) = paginator.next().await {
617 let objects = result?;
618
619 for entry in objects.contents() {
620 let Some(key) = entry.key() else { continue };
621 if key.ends_with('/') {
622 continue;
623 }
624
625 scanned += 1;
626 if scanned > SEARCH_SCAN_LIMIT {
627 break 'scan;
628 }
629
630 let relative = key.strip_prefix(&strip_prefix).unwrap_or(key);
631
632 let mut segment_start = 0;
633 for (index, char) in relative.char_indices() {
634 if char != '/' {
635 continue;
636 }
637
638 let dir_path = &relative[..index];
639 let name = &relative[segment_start..index];
640 segment_start = index + 1;
641
642 if dir_path.len() <= directory.len()
643 || !name.to_lowercase().contains(&search)
644 || !seen_dirs.insert(dir_path.to_compact_string())
645 {
646 continue;
647 }
648
649 dirs.push(StorageAsset {
650 url: storage_url_retriever.get_url(format!("{base}/{dir_path}/")),
651 name: dir_path.to_compact_string(),
652 size: 0,
653 is_directory: true,
654 created: chrono::DateTime::<chrono::Utc>::default(),
655 });
656 }
657
658 if !relative[segment_start..].to_lowercase().contains(&search) {
659 continue;
660 }
661
662 files.push(StorageAsset {
663 url: storage_url_retriever.get_url(key),
664 name: relative.to_compact_string(),
665 size: entry.size().unwrap_or(0).max(0) as u64,
666 created: entry
667 .last_modified()
668 .and_then(|dt| {
669 chrono::DateTime::<chrono::Utc>::from_timestamp(
670 dt.secs(),
671 dt.subsec_nanos(),
672 )
673 })
674 .unwrap_or_default(),
675 is_directory: false,
676 });
677 }
678 }
679 }
680 }
681
682 dirs.sort_unstable_by(|a: &StorageAsset, b: &StorageAsset| a.name.cmp(&b.name));
683 files.sort_unstable_by(|a: &StorageAsset, b: &StorageAsset| a.name.cmp(&b.name));
684
685 Ok(dirs.into_iter().chain(files).take(limit).collect())
686 }
687}
688
689const PART_SIZE: usize = 16 * 1024 * 1024;
690
691async fn upload_multipart(
692 client: &S3Client,
693 bucket: &str,
694 key: &str,
695 content_type: &str,
696 cache_control: Option<&str>,
697 mut data: impl tokio::io::AsyncRead + Unpin,
698) -> Result<u64, anyhow::Error> {
699 let first_part = read_part(&mut data, PART_SIZE).await?;
700
701 if first_part.len() < PART_SIZE {
702 let total = first_part.len() as u64;
703 client
704 .put_object()
705 .bucket(bucket)
706 .key(key)
707 .content_type(content_type)
708 .set_cache_control(cache_control.map(str::to_string))
709 .body(ByteStream::from(first_part))
710 .send()
711 .await?;
712 return Ok(total);
713 }
714
715 let create = client
716 .create_multipart_upload()
717 .bucket(bucket)
718 .key(key)
719 .content_type(content_type)
720 .set_cache_control(cache_control.map(str::to_string))
721 .send()
722 .await?;
723
724 let upload_id = create
725 .upload_id()
726 .ok_or_else(|| anyhow::anyhow!("S3 did not return an upload_id"))?
727 .to_string();
728
729 let result = run_multipart(client, bucket, key, &upload_id, &mut data, first_part).await;
730
731 match result {
732 Ok(total) => Ok(total),
733 Err(err) => {
734 if let Err(abort_err) = client
735 .abort_multipart_upload()
736 .bucket(bucket)
737 .key(key)
738 .upload_id(&upload_id)
739 .send()
740 .await
741 {
742 tracing::warn!(
743 bucket,
744 key,
745 upload_id,
746 "failed to abort multipart upload after error: {:#?}",
747 abort_err
748 );
749 }
750 Err(err)
751 }
752 }
753}
754
755async fn run_multipart(
756 client: &S3Client,
757 bucket: &str,
758 key: &str,
759 upload_id: &str,
760 data: &mut (impl tokio::io::AsyncRead + Unpin),
761 first_part: Bytes,
762) -> Result<u64, anyhow::Error> {
763 let mut completed = Vec::new();
764 let mut total: u64 = 0;
765 let mut part_number: i32 = 1;
766 let mut current = first_part;
767
768 loop {
769 let part_len = current.len() as u64;
770 let resp = client
771 .upload_part()
772 .bucket(bucket)
773 .key(key)
774 .upload_id(upload_id)
775 .part_number(part_number)
776 .body(ByteStream::from(current))
777 .send()
778 .await?;
779
780 completed.push(
781 CompletedPart::builder()
782 .part_number(part_number)
783 .set_e_tag(resp.e_tag().map(|s| s.to_string()))
784 .build(),
785 );
786 total += part_len;
787 part_number += 1;
788
789 let next = read_part(data, PART_SIZE).await?;
790 if next.is_empty() {
791 break;
792 }
793 current = next;
794 }
795
796 let completed_upload = CompletedMultipartUpload::builder()
797 .set_parts(Some(completed))
798 .build();
799
800 client
801 .complete_multipart_upload()
802 .bucket(bucket)
803 .key(key)
804 .upload_id(upload_id)
805 .multipart_upload(completed_upload)
806 .send()
807 .await?;
808
809 Ok(total)
810}
811
812async fn read_part(
813 data: &mut (impl tokio::io::AsyncRead + Unpin),
814 cap: usize,
815) -> Result<Bytes, std::io::Error> {
816 let mut buf = BytesMut::with_capacity(cap);
817 while buf.len() < cap {
818 let n = data.read_buf(&mut buf).await?;
819 if n == 0 {
820 break;
821 }
822 }
823 Ok(buf.freeze())
824}