Chase Douglas
2 months ago
22 changed files with 1446 additions and 209 deletions
@ -0,0 +1,24 @@ |
|||
use std::io::{Error, ErrorKind}; |
|||
|
|||
// Cache the AWS SDK config, as recommended by the AWS SDK documentation. The
|
|||
// initial load is async, so we spawn a thread to load it and then join it to
|
|||
// get the result in a blocking fashion.
|
|||
static AWS_SDK_CONFIG: std::sync::LazyLock<std::io::Result<aws_config::SdkConfig>> = std::sync::LazyLock::new(|| { |
|||
std::thread::spawn(|| { |
|||
let rt = tokio::runtime::Builder::new_current_thread() |
|||
.enable_all() |
|||
.build()?; |
|||
|
|||
std::io::Result::Ok(rt.block_on(aws_config::load_defaults(aws_config::BehaviorVersion::latest()))) |
|||
}) |
|||
.join() |
|||
.map_err(|e| Error::new(ErrorKind::Other, format!("Failed to load AWS config for DSQL connection: {e:#?}")))? |
|||
.map_err(|e| Error::new(ErrorKind::Other, format!("Failed to load AWS config for DSQL connection: {e}"))) |
|||
}); |
|||
|
|||
pub(crate) fn aws_sdk_config() -> std::io::Result<&'static aws_config::SdkConfig> { |
|||
(*AWS_SDK_CONFIG).as_ref().map_err(|e| match e.get_ref() { |
|||
Some(inner) => Error::new(e.kind(), inner), |
|||
None => Error::from(e.kind()), |
|||
}) |
|||
} |
|||
@ -0,0 +1,141 @@ |
|||
use std::{io::{Error, ErrorKind}, path::{Path, PathBuf}, time::SystemTime}; |
|||
|
|||
use rocket::fs::TempFile; |
|||
use tokio::{fs::{File, OpenOptions}, io::{AsyncReadExt, AsyncWriteExt}}; |
|||
|
|||
use super::PersistentFSBackend; |
|||
|
|||
pub(crate) struct LocalFSBackend(String); |
|||
|
|||
impl AsRef<Path> for LocalFSBackend { |
|||
fn as_ref(&self) -> &Path { |
|||
self.0.as_ref() |
|||
} |
|||
} |
|||
|
|||
impl PersistentFSBackend for LocalFSBackend { |
|||
fn new<P: AsRef<Path>>(path: P) -> std::io::Result<Self> { |
|||
Ok(Self(path |
|||
.as_ref() |
|||
.to_str() |
|||
.ok_or_else(|| |
|||
Error::new( |
|||
ErrorKind::InvalidInput, |
|||
"Data folder path {path:?} is not valid UTF-8" |
|||
) |
|||
)? |
|||
.to_string() |
|||
)) |
|||
} |
|||
|
|||
async fn read(self) -> std::io::Result<Vec<u8>> { |
|||
let mut file = File::open(self).await?; |
|||
let mut buffer = Vec::new(); |
|||
file.read_to_end(&mut buffer).await?; |
|||
Ok(buffer) |
|||
} |
|||
|
|||
async fn write(self, buf: &[u8]) -> std::io::Result<()> { |
|||
let mut file = OpenOptions::new().create(true).truncate(true).write(true).open(self).await?; |
|||
file.write_all(buf).await?; |
|||
Ok(()) |
|||
} |
|||
|
|||
async fn path_exists(self) -> std::io::Result<bool> { |
|||
match tokio::fs::metadata(self).await { |
|||
Ok(_) => Ok(true), |
|||
Err(e) => match e.kind() { |
|||
ErrorKind::NotFound => Ok(false), |
|||
_ => Err(e), |
|||
}, |
|||
} |
|||
} |
|||
|
|||
async fn file_exists(self) -> std::io::Result<bool> { |
|||
match tokio::fs::metadata(self).await { |
|||
Ok(metadata) => Ok(metadata.is_file()), |
|||
Err(e) => match e.kind() { |
|||
ErrorKind::NotFound => Ok(false), |
|||
_ => Err(e), |
|||
}, |
|||
} |
|||
} |
|||
|
|||
async fn path_is_dir(self) -> std::io::Result<bool> { |
|||
match tokio::fs::metadata(self).await { |
|||
Ok(metadata) => Ok(metadata.is_dir()), |
|||
Err(e) => match e.kind() { |
|||
ErrorKind::NotFound => Ok(false), |
|||
_ => Err(e), |
|||
}, |
|||
} |
|||
} |
|||
|
|||
async fn canonicalize(self) -> std::io::Result<PathBuf> { |
|||
tokio::fs::canonicalize(self).await |
|||
} |
|||
|
|||
async fn create_dir_all(self) -> std::io::Result<()> { |
|||
tokio::fs::create_dir_all(self).await |
|||
} |
|||
|
|||
async fn persist_temp_file(self, mut temp_file: TempFile<'_>) -> std::io::Result<()> { |
|||
if temp_file.persist_to(&self).await.is_err() { |
|||
temp_file.move_copy_to(self).await?; |
|||
} |
|||
|
|||
Ok(()) |
|||
} |
|||
|
|||
async fn remove_file(self) -> std::io::Result<()> { |
|||
tokio::fs::remove_file(self).await |
|||
} |
|||
|
|||
async fn remove_dir_all(self) -> std::io::Result<()> { |
|||
tokio::fs::remove_dir_all(self).await |
|||
} |
|||
|
|||
async fn last_modified(self) -> std::io::Result<SystemTime> { |
|||
tokio::fs::symlink_metadata(self) |
|||
.await? |
|||
.modified() |
|||
} |
|||
|
|||
async fn download_url(self, local_host: &str) -> std::io::Result<String> { |
|||
use std::sync::LazyLock; |
|||
use crate::{ |
|||
auth::{encode_jwt, generate_file_download_claims, generate_send_claims}, |
|||
db::models::{AttachmentId, CipherId, SendId, SendFileId}, |
|||
CONFIG |
|||
}; |
|||
|
|||
let LocalFSBackend(path) = self; |
|||
|
|||
static ATTACHMENTS_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{}/", CONFIG.attachments_folder())); |
|||
static SENDS_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{}/", CONFIG.sends_folder())); |
|||
|
|||
if path.starts_with(&*ATTACHMENTS_PREFIX) { |
|||
let attachment_parts = path.trim_start_matches(&*ATTACHMENTS_PREFIX).split('/').collect::<Vec<&str>>(); |
|||
|
|||
let [cipher_uuid, attachment_id] = attachment_parts[..] else { |
|||
return Err(Error::new(ErrorKind::InvalidInput, format!("Attachment path {path:?} does not match a known download URL path pattern"))); |
|||
}; |
|||
|
|||
let token = encode_jwt(&generate_file_download_claims(CipherId::from(cipher_uuid.to_string()), AttachmentId(attachment_id.to_string()))); |
|||
|
|||
Ok(format!("{}/attachments/{}/{}?token={}", local_host, cipher_uuid, attachment_id, token)) |
|||
} else if path.starts_with(&*SENDS_PREFIX) { |
|||
let send_parts = path.trim_start_matches(&*SENDS_PREFIX).split('/').collect::<Vec<&str>>(); |
|||
|
|||
let [send_id, file_id] = send_parts[..] else { |
|||
return Err(Error::new(ErrorKind::InvalidInput, format!("Send path {path:?} does not match a known download URL path pattern"))); |
|||
}; |
|||
|
|||
let token = encode_jwt(&generate_send_claims(&SendId::from(send_id.to_string()), &SendFileId::from(file_id.to_string()))); |
|||
|
|||
Ok(format!("{}/api/sends/{}/{}?t={}", local_host, send_id, file_id, token)) |
|||
} else { |
|||
Err(Error::new(ErrorKind::InvalidInput, "Data folder path {path:?} does not match a known download URL path pattern")) |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,316 @@ |
|||
mod local; |
|||
|
|||
#[cfg(s3)] |
|||
mod s3; |
|||
|
|||
use std::{io::{Error, ErrorKind}, path::{Path, PathBuf}, time::SystemTime}; |
|||
|
|||
use rocket::fs::TempFile; |
|||
|
|||
enum FSType { |
|||
Local(local::LocalFSBackend), |
|||
|
|||
#[cfg(s3)] |
|||
S3(s3::S3FSBackend), |
|||
} |
|||
|
|||
pub(crate) trait PersistentFSBackend: Sized { |
|||
fn new<P: AsRef<Path>>(path: P) -> std::io::Result<Self>; |
|||
async fn read(self) -> std::io::Result<Vec<u8>>; |
|||
async fn write(self, buf: &[u8]) -> std::io::Result<()>; |
|||
async fn path_exists(self) -> std::io::Result<bool>; |
|||
async fn file_exists(self) -> std::io::Result<bool>; |
|||
async fn path_is_dir(self) -> std::io::Result<bool>; |
|||
async fn canonicalize(self) -> std::io::Result<PathBuf>; |
|||
async fn create_dir_all(self) -> std::io::Result<()>; |
|||
async fn persist_temp_file(self, temp_file: TempFile<'_>) -> std::io::Result<()>; |
|||
async fn remove_file(self) -> std::io::Result<()>; |
|||
async fn remove_dir_all(self) -> std::io::Result<()>; |
|||
async fn last_modified(self) -> std::io::Result<SystemTime>; |
|||
async fn download_url(self, local_host: &str) -> std::io::Result<String>; |
|||
} |
|||
|
|||
impl PersistentFSBackend for FSType { |
|||
fn new<P: AsRef<Path>>(path: P) -> std::io::Result<Self> { |
|||
#[cfg(s3)] |
|||
if path.as_ref().starts_with("s3://") { |
|||
return Ok(FSType::S3(s3::S3FSBackend::new(path)?)); |
|||
} |
|||
|
|||
Ok(FSType::Local(local::LocalFSBackend::new(path)?)) |
|||
} |
|||
|
|||
async fn read(self) -> std::io::Result<Vec<u8>> { |
|||
match self { |
|||
FSType::Local(fs) => fs.read().await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.read().await, |
|||
} |
|||
} |
|||
|
|||
async fn write(self, buf: &[u8]) -> std::io::Result<()> { |
|||
match self { |
|||
FSType::Local(fs) => fs.write(buf).await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.write(buf).await, |
|||
} |
|||
} |
|||
|
|||
async fn path_exists(self) -> std::io::Result<bool> { |
|||
match self { |
|||
FSType::Local(fs) => fs.path_exists().await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.path_exists().await, |
|||
} |
|||
} |
|||
|
|||
async fn file_exists(self) -> std::io::Result<bool> { |
|||
match self { |
|||
FSType::Local(fs) => fs.file_exists().await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.file_exists().await, |
|||
} |
|||
} |
|||
|
|||
async fn path_is_dir(self) -> std::io::Result<bool> { |
|||
match self { |
|||
FSType::Local(fs) => fs.path_is_dir().await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.path_is_dir().await, |
|||
} |
|||
} |
|||
|
|||
async fn canonicalize(self) -> std::io::Result<PathBuf> { |
|||
match self { |
|||
FSType::Local(fs) => fs.canonicalize().await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.canonicalize().await, |
|||
} |
|||
} |
|||
|
|||
async fn create_dir_all(self) -> std::io::Result<()> { |
|||
match self { |
|||
FSType::Local(fs) => fs.create_dir_all().await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.create_dir_all().await, |
|||
} |
|||
} |
|||
|
|||
async fn persist_temp_file(self, temp_file: TempFile<'_>) -> std::io::Result<()> { |
|||
match self { |
|||
FSType::Local(fs) => fs.persist_temp_file(temp_file).await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.persist_temp_file(temp_file).await, |
|||
} |
|||
} |
|||
|
|||
async fn remove_file(self) -> std::io::Result<()> { |
|||
match self { |
|||
FSType::Local(fs) => fs.remove_file().await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.remove_file().await, |
|||
} |
|||
} |
|||
|
|||
async fn remove_dir_all(self) -> std::io::Result<()> { |
|||
match self { |
|||
FSType::Local(fs) => fs.remove_dir_all().await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.remove_dir_all().await, |
|||
} |
|||
} |
|||
|
|||
async fn last_modified(self) -> std::io::Result<SystemTime> { |
|||
match self { |
|||
FSType::Local(fs) => fs.last_modified().await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.last_modified().await, |
|||
} |
|||
} |
|||
|
|||
async fn download_url(self, local_host: &str) -> std::io::Result<String> { |
|||
match self { |
|||
FSType::Local(fs) => fs.download_url(local_host).await, |
|||
#[cfg(s3)] |
|||
FSType::S3(fs) => fs.download_url(local_host).await, |
|||
} |
|||
} |
|||
} |
|||
|
|||
/// Reads the contents of a file at the given path.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `path` - A reference to the path of the file to read.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<Vec<u8>>` - A result containing a vector of bytes with the
|
|||
/// file contents if successful, or an I/O error.
|
|||
pub(crate) async fn read<P: AsRef<Path>>(path: P) -> std::io::Result<Vec<u8>> { |
|||
FSType::new(path)?.read().await |
|||
} |
|||
|
|||
/// Writes data to a file at the given path.
|
|||
///
|
|||
/// If the file does not exist, it will be created. If it does exist, it will be
|
|||
/// overwritten.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `path` - A reference to the path of the file to write.
|
|||
/// * `buf` - A byte slice containing the data to write to the file.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<()>` - A result indicating success or an I/O error.
|
|||
pub(crate) async fn write<P: AsRef<Path>>(path: P, buf: &[u8]) -> std::io::Result<()> { |
|||
FSType::new(path)?.write(buf).await |
|||
} |
|||
|
|||
/// Checks whether a path exists.
|
|||
///
|
|||
/// This function returns `true` in all cases where the path exists, including
|
|||
/// as a file, directory, or symlink.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `path` - A reference to the path to check.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<bool>` - A result containing a boolean value indicating
|
|||
/// whether the path exists.
|
|||
pub(crate) async fn path_exists<P: AsRef<Path>>(path: P) -> std::io::Result<bool> { |
|||
FSType::new(path)?.path_exists().await |
|||
} |
|||
|
|||
/// Checks whether a regular file exists at the given path.
|
|||
///
|
|||
/// This function returns `false` if the path is a symlink.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `path` - A reference to the path to check.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<bool>` - A result containing a boolean value indicating
|
|||
/// whether a regular file exists at the given path.
|
|||
pub(crate) async fn file_exists<P: AsRef<Path>>(path: P) -> std::io::Result<bool> { |
|||
FSType::new(path)?.file_exists().await |
|||
} |
|||
|
|||
/// Checks whether a directory exists at the given path.
|
|||
///
|
|||
/// This function returns `false` if the path is a symlink.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `path` - A reference to the path to check.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<bool>` - A result containing a boolean value indicating
|
|||
/// whether a directory exists at the given path.
|
|||
pub(crate) async fn path_is_dir<P: AsRef<Path>>(path: P) -> std::io::Result<bool> { |
|||
FSType::new(path)?.path_is_dir().await |
|||
} |
|||
|
|||
/// Canonicalizes the given path.
|
|||
///
|
|||
/// This function resolves the given path to an absolute path, eliminating any
|
|||
/// symbolic links and relative path components.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `path` - A reference to the path to canonicalize.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<PathBuf>` - A result containing the canonicalized path if successful,
|
|||
/// or an I/O error.
|
|||
pub(crate) async fn canonicalize<P: AsRef<Path>>(path: P) -> std::io::Result<PathBuf> { |
|||
FSType::new(path)?.canonicalize().await |
|||
} |
|||
|
|||
/// Creates a directory and all its parent components as needed.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `path` - A reference to the path of the directory to create.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<()>` - A result indicating success or an I/O error.
|
|||
pub(crate) async fn create_dir_all<P: AsRef<Path>>(path: P) -> std::io::Result<()> { |
|||
FSType::new(path)?.create_dir_all().await |
|||
} |
|||
|
|||
/// Persists a temporary file to a permanent location.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `temp_file` - The temporary file to persist.
|
|||
/// * `path` - A reference to the path where the file should be persisted.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<()>` - A result indicating success or an I/O error.
|
|||
pub(crate) async fn persist_temp_file<P: AsRef<Path>>(temp_file: TempFile<'_>, path: P) -> std::io::Result<()> { |
|||
FSType::new(path)?.persist_temp_file(temp_file).await |
|||
} |
|||
|
|||
/// Removes a file at the given path.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `path` - A reference to the path of the file to remove.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<()>` - A result indicating success or an I/O error.
|
|||
pub(crate) async fn remove_file<P: AsRef<Path>>(path: P) -> std::io::Result<()> { |
|||
FSType::new(path)?.remove_file().await |
|||
} |
|||
|
|||
/// Removes a directory and all its contents at the given path.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `path` - A reference to the path of the directory to remove.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<()>` - A result indicating success or an I/O error.
|
|||
pub(crate) async fn remove_dir_all<P: AsRef<Path>>(path: P) -> std::io::Result<()> { |
|||
FSType::new(path)?.remove_dir_all().await |
|||
} |
|||
|
|||
pub(crate) async fn file_is_expired<P: AsRef<Path>>(path: P, ttl: u64) -> Result<bool, Error> { |
|||
let path = path.as_ref(); |
|||
|
|||
let modified = FSType::new(path)?.last_modified().await?; |
|||
|
|||
let age = SystemTime::now().duration_since(modified) |
|||
.map_err(|e| Error::new( |
|||
ErrorKind::InvalidData, |
|||
format!("Failed to determine file age for {path:?} from last modified timestamp '{modified:#?}': {e:?}" |
|||
)))?; |
|||
|
|||
Ok(ttl > 0 && ttl <= age.as_secs()) |
|||
} |
|||
|
|||
/// Generates a pre-signed url to download attachment and send files.
|
|||
///
|
|||
/// # Arguments
|
|||
///
|
|||
/// * `path` - A reference to the path of the file to read.
|
|||
/// * `local_host` - This API server host.
|
|||
///
|
|||
/// # Returns
|
|||
///
|
|||
/// * `std::io::Result<String>` - A result containing the url if successful, or an I/O error.
|
|||
pub(crate) async fn download_url<P: AsRef<Path>>(path: P, local_host: &str) -> std::io::Result<String> { |
|||
FSType::new(path)?.download_url(local_host).await |
|||
} |
|||
@ -0,0 +1,316 @@ |
|||
use std::{io::{Error, ErrorKind}, path::{Path, PathBuf}, time::SystemTime}; |
|||
|
|||
use aws_sdk_s3::{client::Client, primitives::ByteStream, types::StorageClass::IntelligentTiering}; |
|||
use rocket::{fs::TempFile, http::ContentType}; |
|||
use tokio::{fs::File, io::AsyncReadExt}; |
|||
use url::Url; |
|||
|
|||
use crate::aws::aws_sdk_config; |
|||
|
|||
use super::PersistentFSBackend; |
|||
|
|||
pub(crate) struct S3FSBackend { |
|||
path: PathBuf, |
|||
bucket: String, |
|||
key: String, |
|||
} |
|||
|
|||
fn s3_client() -> std::io::Result<Client> { |
|||
static AWS_S3_CLIENT: std::sync::LazyLock<std::io::Result<Client>> = std::sync::LazyLock::new(|| { |
|||
Ok(Client::new(aws_sdk_config()?)) |
|||
}); |
|||
|
|||
(*AWS_S3_CLIENT) |
|||
.as_ref() |
|||
.map(|client| client.clone()) |
|||
.map_err(|e| match e.get_ref() { |
|||
Some(inner) => Error::new(e.kind(), inner), |
|||
None => Error::from(e.kind()), |
|||
}) |
|||
} |
|||
|
|||
impl PersistentFSBackend for S3FSBackend { |
|||
fn new<P: AsRef<Path>>(path: P) -> std::io::Result<Self> { |
|||
let path = path.as_ref(); |
|||
|
|||
let url = Url::parse(path.to_str().ok_or_else(|| Error::new(ErrorKind::InvalidInput, "Invalid path"))?) |
|||
.map_err(|e| Error::new(ErrorKind::InvalidInput, format!("Invalid data folder S3 URL {path:?}: {e}")))?; |
|||
|
|||
let bucket = url.host_str() |
|||
.ok_or_else(|| Error::new(ErrorKind::InvalidInput, format!("Missing Bucket name in data folder S3 URL {path:?}")))? |
|||
.to_string(); |
|||
|
|||
let key = url.path().trim_start_matches('/').to_string(); |
|||
|
|||
Ok(S3FSBackend { |
|||
path: path.to_path_buf(), |
|||
bucket, |
|||
key, |
|||
}) |
|||
} |
|||
|
|||
async fn read(self) -> std::io::Result<Vec<u8>> { |
|||
let S3FSBackend { path, key, bucket } = self; |
|||
|
|||
let result = s3_client()? |
|||
.get_object() |
|||
.bucket(bucket) |
|||
.key(key) |
|||
.send() |
|||
.await; |
|||
|
|||
match result { |
|||
Ok(response) => { |
|||
let mut buffer = Vec::new(); |
|||
response.body.into_async_read().read_to_end(&mut buffer).await?; |
|||
Ok(buffer) |
|||
} |
|||
Err(e) => { |
|||
if let Some(service_error) = e.as_service_error() { |
|||
if service_error.is_no_such_key() { |
|||
Err(Error::new(ErrorKind::NotFound, format!("Data folder S3 object {path:?} not found"))) |
|||
} else { |
|||
Err(Error::other(format!("Failed to request data folder S3 object {path:?}: {e:?}"))) |
|||
} |
|||
} else { |
|||
Err(Error::other(format!("Failed to request data folder S3 object {path:?}: {e:?}"))) |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
async fn write(self, buf: &[u8]) -> std::io::Result<()> { |
|||
let S3FSBackend { path, key, bucket } = self; |
|||
|
|||
let content_type = Path::new(&key) |
|||
.extension() |
|||
.and_then(|ext| ext.to_str()) |
|||
.and_then(|ext| ContentType::from_extension(ext)) |
|||
.and_then(|t| Some(t.to_string())); |
|||
|
|||
s3_client()? |
|||
.put_object() |
|||
.bucket(bucket) |
|||
.set_content_type(content_type) |
|||
.key(key) |
|||
.storage_class(IntelligentTiering) |
|||
.body(ByteStream::from(buf.to_vec())) |
|||
.send() |
|||
.await |
|||
.map_err(|e| Error::other(format!("Failed to write to data folder S3 object {path:?}: {e:?}")))?; |
|||
|
|||
Ok(()) |
|||
} |
|||
|
|||
async fn path_exists(self) -> std::io::Result<bool> { |
|||
Ok(true) |
|||
} |
|||
|
|||
async fn file_exists(self) -> std::io::Result<bool> { |
|||
let S3FSBackend { path, key, bucket } = self; |
|||
|
|||
match s3_client()? |
|||
.head_object() |
|||
.bucket(bucket) |
|||
.key(key) |
|||
.send() |
|||
.await { |
|||
Ok(_) => Ok(true), |
|||
Err(e) => { |
|||
if let Some(service_error) = e.as_service_error() { |
|||
if service_error.is_not_found() { |
|||
Ok(false) |
|||
} else { |
|||
Err(Error::other(format!("Failed to request data folder S3 object {path:?}: {e:?}"))) |
|||
} |
|||
} else { |
|||
Err(Error::other(format!("Failed to request data folder S3 object {path:?}: {e:?}"))) |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
async fn path_is_dir(self) -> std::io::Result<bool> { |
|||
Ok(true) |
|||
} |
|||
|
|||
async fn canonicalize(self) -> std::io::Result<PathBuf> { |
|||
Ok(self.path) |
|||
} |
|||
|
|||
async fn create_dir_all(self) -> std::io::Result<()> { |
|||
Ok(()) |
|||
} |
|||
|
|||
async fn persist_temp_file(self, temp_file: TempFile<'_>) -> std::io::Result<()> { |
|||
let S3FSBackend { path, key, bucket } = self; |
|||
|
|||
// We want to stream the TempFile directly to S3 without copying it into
|
|||
// another memory buffer. The official AWS SDK makes it easy to stream
|
|||
// from a `tokio::fs::File`, but does not have a reasonable way to stream
|
|||
// from an `impl AsyncBufRead`.
|
|||
//
|
|||
// A TempFile's contents may be saved in memory or on disk. We use the
|
|||
// SDK to stream the file if we can access it on disk, otherwise we fall
|
|||
// back to a second copy in memory.
|
|||
let file = match temp_file.path() { |
|||
Some(path) => File::open(path).await.ok(), |
|||
None => None, |
|||
}; |
|||
|
|||
let byte_stream = match file { |
|||
Some(file) => ByteStream::read_from().file(file).build().await.ok(), |
|||
None => None, |
|||
}; |
|||
|
|||
let byte_stream = match byte_stream { |
|||
Some(byte_stream) => byte_stream, |
|||
None => { |
|||
// TODO: Implement a mechanism to stream the file directly to S3
|
|||
// without buffering it again in memory. This would require
|
|||
// chunking it into a multi-part upload. See example here:
|
|||
// https://imfeld.dev/writing/rust_s3_streaming_upload
|
|||
let mut read_stream = temp_file.open().await?; |
|||
let mut buf = Vec::with_capacity(temp_file.len() as usize); |
|||
read_stream.read_to_end(&mut buf).await?; |
|||
ByteStream::from(buf) |
|||
} |
|||
}; |
|||
|
|||
let content_type = temp_file |
|||
.content_type() |
|||
.map(|t| t.to_string()) |
|||
.or_else(|| |
|||
temp_file.name() |
|||
.and_then(|name| Path::new(name).extension()) |
|||
.and_then(|ext| ext.to_str()) |
|||
.and_then(|ext| ContentType::from_extension(ext)) |
|||
.and_then(|t| Some(t.to_string())) |
|||
); |
|||
|
|||
s3_client()? |
|||
.put_object() |
|||
.bucket(bucket) |
|||
.key(key) |
|||
.storage_class(IntelligentTiering) |
|||
.set_content_type(content_type) |
|||
.body(byte_stream) |
|||
.send() |
|||
.await |
|||
.map_err(|e| Error::other(format!("Failed to write to data folder S3 object {path:?}: {e:?}")))?; |
|||
|
|||
Ok(()) |
|||
} |
|||
|
|||
async fn remove_file(self) -> std::io::Result<()> { |
|||
let S3FSBackend { path, key, bucket } = self; |
|||
|
|||
s3_client()? |
|||
.delete_object() |
|||
.bucket(bucket) |
|||
.key(key) |
|||
.send() |
|||
.await |
|||
.map_err(|e| Error::other(format!("Failed to delete data folder S3 object {path:?}: {e:?}")))?; |
|||
|
|||
Ok(()) |
|||
} |
|||
|
|||
async fn remove_dir_all(self) -> std::io::Result<()> { |
|||
use aws_sdk_s3::types::{Delete, ObjectIdentifier}; |
|||
|
|||
let S3FSBackend { path, key: prefix, bucket } = self; |
|||
|
|||
let s3_client = s3_client()?; |
|||
|
|||
let mut list_response = s3_client |
|||
.list_objects_v2() |
|||
.bucket(bucket.clone()) |
|||
.prefix(format!("{prefix}/")) |
|||
.into_paginator() |
|||
.send(); |
|||
|
|||
while let Some(list_result) = list_response.next().await { |
|||
let list_result = list_result |
|||
.map_err(|e| Error::other(format!("Failed to list data folder S3 objects with prefix {path:?}/ intended for deletion: {e:?}")))?; |
|||
|
|||
let objects = list_result |
|||
.contents |
|||
.ok_or_else(|| Error::other(format!("Failed to list data folder S3 objects with prefix {path:?}/ intended for deletion: Missing contents")))?; |
|||
|
|||
let keys = objects.into_iter() |
|||
.map(|object| object.key |
|||
.ok_or_else(|| Error::other(format!("Failed to list data folder S3 objects with prefix {path:?}/ intended for deletion: An object is missing its key"))) |
|||
) |
|||
.collect::<std::io::Result<Vec<_>>>()?; |
|||
|
|||
let mut delete = Delete::builder().quiet(true); |
|||
|
|||
for key in keys { |
|||
delete = delete.objects( |
|||
ObjectIdentifier::builder() |
|||
.key(key) |
|||
.build() |
|||
.map_err(|e| Error::other(format!("Failed to delete data folder S3 objects with prefix {path:?}/: {e:?}")))? |
|||
); |
|||
} |
|||
|
|||
let delete = delete |
|||
.build() |
|||
.map_err(|e| Error::other(format!("Failed to delete data folder S3 objects with prefix {path:?}/: {e:?}")))?; |
|||
|
|||
s3_client |
|||
.delete_objects() |
|||
.bucket(bucket.clone()) |
|||
.delete(delete) |
|||
.send() |
|||
.await |
|||
.map_err(|e| Error::other(format!("Failed to delete data folder S3 objects with prefix {path:?}/: {e:?}")))?; |
|||
} |
|||
|
|||
Ok(()) |
|||
} |
|||
|
|||
async fn last_modified(self) -> std::io::Result<SystemTime> { |
|||
let S3FSBackend { path, key, bucket } = self; |
|||
|
|||
let response = s3_client()? |
|||
.head_object() |
|||
.bucket(bucket) |
|||
.key(key) |
|||
.send() |
|||
.await |
|||
.map_err(|e| match e.as_service_error() { |
|||
Some(service_error) if service_error.is_not_found() => Error::new(ErrorKind::NotFound, format!("Failed to get metadata for data folder S3 object {path:?}: Object does not exist")), |
|||
Some(service_error) => Error::other(format!("Failed to get metadata for data folder S3 object {path:?}: {service_error:?}")), |
|||
None => Error::other(format!("Failed to get metadata for data folder S3 object {path:?}: {e:?}")), |
|||
})?; |
|||
|
|||
let last_modified = response.last_modified |
|||
.ok_or_else(|| Error::new(ErrorKind::NotFound, format!("Failed to get metadata for data folder S3 object {path:?}: Missing last modified data")))?; |
|||
|
|||
SystemTime::try_from(last_modified) |
|||
.map_err(|e| Error::new(ErrorKind::InvalidData, format!("Failed to parse last modified date for data folder S3 object {path:?}: {e:?}"))) |
|||
} |
|||
|
|||
async fn download_url(self, _local_host: &str) -> std::io::Result<String> { |
|||
use std::time::Duration; |
|||
use aws_sdk_s3::presigning::PresigningConfig; |
|||
|
|||
let S3FSBackend { path, key, bucket } = self; |
|||
|
|||
s3_client()? |
|||
.get_object() |
|||
.bucket(bucket) |
|||
.key(key) |
|||
.presigned( |
|||
PresigningConfig::expires_in(Duration::from_secs(5 * 60)) |
|||
.map_err(|e| Error::other( |
|||
format!("Failed to generate presigned config for GetObject URL for data folder S3 object {path:?}: {e:?}") |
|||
))? |
|||
) |
|||
.await |
|||
.map(|presigned| presigned.uri().to_string()) |
|||
.map_err(|e| Error::other(format!("Failed to generate presigned URL for GetObject for data folder S3 object {path:?}: {e:?}"))) |
|||
} |
|||
} |
|||
Loading…
Reference in new issue