gru/rustpad
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

s3 support commit1

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-07-23 19:40:30 +02:00
parent 8c3842157d
commit 950142a9c0
10 changed files with 282 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api.rs
+18 -26
View File
@@ -616,20 +616,19 @@ pub async fn upload_pad_file(
let (original, bytes) = file.ok_or_else(|| ApiError::bad_request("No file provided"))?;
let safe = sanitize_filename(&original);
let file_token = db::pad_file_token(&state.db, pad.id).await?;
let directory = format!("{}_{}", pad.id, file_token);
let dir = std::path::Path::new(&state.files_dir).join("pads").join(&directory);
tokio::fs::create_dir_all(&dir).await.map_err(|_| ApiError::internal("Failed to create the files directory"))?;
let mut stored = safe.clone();
let mut path = dir.join(&stored);
if path.exists() {
let mut key = crate::storage::object_key("pads", pad.id, &file_token, &stored);
if state.storage.exists(&key).await.map_err(|_| ApiError::internal("Failed to check file storage"))? {
let stem = std::path::Path::new(&safe).file_stem().and_then(|v| v.to_str()).unwrap_or("plik");
let ext = std::path::Path::new(&safe).extension().and_then(|v| v.to_str()).map(|v| format!(".{v}")).unwrap_or_default();
stored = format!("{stem}-{}{}", db::random_suffix(6), ext);
path = dir.join(&stored);
key = crate::storage::object_key("pads", pad.id, &file_token, &stored);
}
tokio::fs::write(&path, &bytes).await.map_err(|_| ApiError::internal("Failed to save the file"))?;
let url = format!("/f/{}/{}", file_token, stored);
let mime = mime_guess::from_path(&stored).first_or_octet_stream().to_string();
let cache_control = format!("public, max-age={}", state.file_cache_max_age_seconds);
state.storage.put(&key, bytes.clone().into(), &mime, &cache_control).await
.map_err(|_| ApiError::internal("Failed to save the file"))?;
db::register_pad_file(&state.db, pad.id, &stored, &url, &mime, bytes.len() as i64).await?;
Ok(Json(serde_json::json!({"name": stored, "url": url})))
}
@@ -680,20 +679,19 @@ pub async fn upload_note_file(
let (original, bytes) = file.ok_or_else(|| ApiError::bad_request("No file provided"))?;
let safe = sanitize_filename(&original);
let file_token = db::note_file_token(&state.db, note.id).await?;
let directory = format!("{}_{}", note.id, file_token);
let dir = std::path::Path::new(&state.files_dir).join("notes").join(&directory);
tokio::fs::create_dir_all(&dir).await.map_err(|_| ApiError::internal("Failed to create the files directory"))?;
let mut stored = safe.clone();
let mut path = dir.join(&stored);
if path.exists() {
let mut key = crate::storage::object_key("notes", note.id, &file_token, &stored);
if state.storage.exists(&key).await.map_err(|_| ApiError::internal("Failed to check file storage"))? {
let stem = std::path::Path::new(&safe).file_stem().and_then(|v| v.to_str()).unwrap_or("plik");
let ext = std::path::Path::new(&safe).extension().and_then(|v| v.to_str()).map(|v| format!(".{v}")).unwrap_or_default();
stored = format!("{stem}-{}{}", db::random_suffix(6), ext);
path = dir.join(&stored);
key = crate::storage::object_key("notes", note.id, &file_token, &stored);
}
tokio::fs::write(&path, &bytes).await.map_err(|_| ApiError::internal("Failed to save the file"))?;
let url = format!("/f/{}/{}", file_token, stored);
let mime = mime_guess::from_path(&stored).first_or_octet_stream().to_string();
let cache_control = format!("public, max-age={}", state.file_cache_max_age_seconds);
state.storage.put(&key, bytes.clone().into(), &mime, &cache_control).await
.map_err(|_| ApiError::internal("Failed to save the file"))?;
db::register_note_file(&state.db, note.id, &stored, &url, &mime, bytes.len() as i64).await?;
Ok(Json(serde_json::json!({"name": stored, "url": url})))
}
@@ -742,13 +740,8 @@ pub async fn delete_note_file(
.ok_or_else(ApiError::not_found_file)?;
let relative = file.url.trim_start_matches('/').split('/').collect::<Vec<_>>();
if relative.len() == 3 && relative[0] == "f" {
let directory = format!("{}_{}", note.id, relative[1]);
let path = std::path::Path::new(&state.files_dir).join("notes").join(directory).join(sanitize_filename(relative[2]));
match tokio::fs::remove_file(&path).await {
Ok(()) => {},
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {},
Err(_) => return Err(ApiError::internal("Failed to delete the file")),
}
let key = crate::storage::object_key("notes", note.id, relative[1], &sanitize_filename(relative[2]));
state.storage.delete(&key).await.map_err(|_| ApiError::internal("Failed to delete the file"))?;
}
db::delete_note_file(&state.db, note.id, file_id).await?;
Ok(Json(serde_json::json!({"ok": true})))
@@ -788,11 +781,10 @@ async fn serve_token_file(state: &SharedState, token: &str, filename: &str) -> R
db::FileOwnerKind::Pad => "pads",
db::FileOwnerKind::Note => "notes",
};
let directory = format!("{}_{}", owner.id, token);
let canonical = std::path::Path::new(&state.files_dir).join(kind).join(&directory).join(&safe);
let legacy = std::path::Path::new(&state.files_dir).join(&directory).join(&safe);
let path = if canonical.is_file() { canonical } else { legacy };
let bytes = tokio::fs::read(&path).await.map_err(|_| ApiError::not_found_file())?;
let key = crate::storage::object_key(kind, owner.id, token, &safe);
let legacy_key = crate::storage::legacy_key(owner.id, token, &safe);
let bytes = state.storage.get_local_with_legacy(&key, &legacy_key).await
.map_err(|_| ApiError::not_found_file())?;
let mime = mime_guess::from_path(&safe).first_or_octet_stream();
let mut response = bytes.into_response();
response.headers_mut().insert(
src/config.rs
+22 -1
View File
@@ -8,6 +8,7 @@ pub struct Config {
pub database_max_connections: u32,
pub static_dir: String,
pub files_dir: String,
pub storage: crate::storage::StorageConfig,
pub upload_max_size_bytes: usize,
pub asset_version: String,
pub asset_cache_max_age_seconds: u64,
@@ -31,6 +32,19 @@ impl Config {
env_var("UPLOAD_MAX_SIZE_MB", "20").parse()?;
let anonymous_access_token_ttl_days = env_positive_i64("ANONYMOUS_ACCESS_TOKEN_TTL_DAYS", 7)?;
let user_session_ttl_days = env_positive_i64("USER_SESSION_TTL_DAYS", 30)?;
let files_dir = env_var("FILES_DIR", "data/files");
let storage = match env_var("STORAGE_DRIVER", "local").trim().to_ascii_lowercase().as_str() {
"local" => crate::storage::StorageConfig::Local { root: files_dir.clone().into() },
"s3" => crate::storage::StorageConfig::S3 {
endpoint: env::var("S3_ENDPOINT").ok(),
region: env_var("S3_REGION", "us-east-1"),
bucket: required_env("S3_BUCKET")?,
access_key: required_env("S3_ACCESS_KEY")?,
secret_key: required_env("S3_SECRET_KEY")?,
force_path_style: env_bool("S3_FORCE_PATH_STYLE", false)?,
},
_ => return Err("STORAGE_DRIVER must be local or s3".into()),
};
if upload_max_size_mb == 0 {
return Err("UPLOAD_MAX_SIZE_MB must be greater than 0".into());
@@ -57,7 +71,8 @@ impl Config {
),
database_max_connections,
static_dir: env_var("STATIC_DIR", "static"),
files_dir: env_var("FILES_DIR", "data/files"),
files_dir,
storage,
upload_max_size_bytes: upload_max_size_mb
.checked_mul(1024 * 1024)
.ok_or("UPLOAD_MAX_SIZE_MB is too large")?,
@@ -108,3 +123,9 @@ fn env_positive_i64(name: &str, default: i64) -> Result<i64, Box<dyn std::error:
fn env_nonnegative_u64(name: &str, default: u64) -> Result<u64, Box<dyn std::error::Error>> {
Ok(env_var(name, &default.to_string()).parse()?)
}
fn required_env(name: &str) -> Result<String, Box<dyn std::error::Error>> {
let value = env::var(name).map_err(|_| format!("{name} is required when STORAGE_DRIVER=s3"))?;
if value.trim().is_empty() { return Err(format!("{name} cannot be empty when STORAGE_DRIVER=s3").into()); }
Ok(value)
}
src/main.rs
+5 -3
View File
@@ -6,6 +6,7 @@ mod database;
mod db;
mod queries;
mod state;
mod storage;
mod websocket;
use std::{net::SocketAddr, sync::Arc};
@@ -30,6 +31,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
database_max_connections = config.database_max_connections,
static_dir = %config.static_dir,
files_dir = %config.files_dir,
storage_driver = match &config.storage { storage::StorageConfig::Local { .. } => "local", storage::StorageConfig::S3 { .. } => "s3" },
upload_max_size_bytes = config.upload_max_size_bytes,
asset_cache_max_age_seconds = config.asset_cache_max_age_seconds,
file_cache_max_age_seconds = config.file_cache_max_age_seconds,
@@ -51,12 +53,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
run_migrations(&db).await?;
info!(database_kind = ?db.kind(), "database migrations completed");
std::fs::create_dir_all(&config.files_dir)?;
info!(files_dir = %config.files_dir, "file storage ready");
let storage = storage::Storage::from_config(config.storage.clone()).await?;
info!(storage_driver = storage.backend_name(), "file storage ready");
let state = Arc::new(AppState::new(
db,
config.asset_version.clone(),
config.files_dir.clone(),
storage,
config.upload_max_size_bytes,
config.file_cache_max_age_seconds,
config.smtp.clone(),
src/state.rs
+3 -3
View File
@@ -36,7 +36,7 @@ pub enum RoomEvent {
pub struct AppState {
pub db: Database,
pub asset_version: String,
pub files_dir: String,
pub storage: crate::storage::Storage,
pub upload_max_size_bytes: usize,
pub file_cache_max_age_seconds: u64,
pub smtp: Option<SmtpConfig>,
@@ -51,8 +51,8 @@ pub struct AppState {
}
impl AppState {
pub fn new(db: Database, asset_version: String, files_dir: String, upload_max_size_bytes: usize, file_cache_max_age_seconds: u64, smtp: Option<SmtpConfig>, registration_enabled: bool, account_confirmation_required: bool, frontend_log_level: String, anonymous_access_token_ttl_days: i64, user_session_ttl_days: i64) -> Self {
Self { db, asset_version, files_dir, upload_max_size_bytes, file_cache_max_age_seconds, smtp, registration_enabled, account_confirmation_required, frontend_log_level, anonymous_access_token_ttl_days, user_session_ttl_days, channels: RwLock::new(HashMap::new()), presence: RwLock::new(HashMap::new()), next_connection_id: AtomicU64::new(1) }
pub fn new(db: Database, asset_version: String, storage: crate::storage::Storage, upload_max_size_bytes: usize, file_cache_max_age_seconds: u64, smtp: Option<SmtpConfig>, registration_enabled: bool, account_confirmation_required: bool, frontend_log_level: String, anonymous_access_token_ttl_days: i64, user_session_ttl_days: i64) -> Self {
Self { db, asset_version, storage, upload_max_size_bytes, file_cache_max_age_seconds, smtp, registration_enabled, account_confirmation_required, frontend_log_level, anonymous_access_token_ttl_days, user_session_ttl_days, channels: RwLock::new(HashMap::new()), presence: RwLock::new(HashMap::new()), next_connection_id: AtomicU64::new(1) }
}
async fn channel_for_key(&self, key: String) -> broadcast::Sender<RoomEvent> {
if let Some(sender) = self.channels.read().await.get(&key) { return sender.clone(); }
src/storage.rs
+159
View File
@@ -0,0 +1,159 @@
use std::{path::PathBuf, sync::Arc};
use aws_config::Region;
use aws_credential_types::Credentials;
use aws_sdk_s3::{config::Builder as S3ConfigBuilder, primitives::ByteStream, Client};
use bytes::Bytes;
#[derive(Debug, Clone)]
pub enum StorageConfig {
Local { root: PathBuf },
S3 {
endpoint: Option<String>,
region: String,
bucket: String,
access_key: String,
secret_key: String,
force_path_style: bool,
},
}
#[derive(Clone)]
pub enum Storage {
Local { root: PathBuf },
S3 { client: Client, bucket: Arc<str> },
}
impl std::fmt::Debug for Storage {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Local { root } => f.debug_struct("LocalStorage").field("root", root).finish(),
Self::S3 { bucket, .. } => f.debug_struct("S3Storage").field("bucket", bucket).finish(),
}
}
}
impl Storage {
pub async fn from_config(config: StorageConfig) -> Result<Self, Box<dyn std::error::Error>> {
match config {
StorageConfig::Local { root } => {
tokio::fs::create_dir_all(&root).await?;
Ok(Self::Local { root })
}
StorageConfig::S3 { endpoint, region, bucket, access_key, secret_key, force_path_style } => {
let credentials = Credentials::new(access_key, secret_key, None, None, "rustpad-env");
let shared = aws_config::defaults(aws_config::BehaviorVersion::latest())
.region(Region::new(region.clone()))
.credentials_provider(credentials)
.load()
.await;
let mut builder = S3ConfigBuilder::from(&shared)
.region(Region::new(region))
.force_path_style(force_path_style);
if let Some(endpoint) = endpoint.filter(|value| !value.trim().is_empty()) {
builder = builder.endpoint_url(endpoint);
}
Ok(Self::S3 { client: Client::from_conf(builder.build()), bucket: Arc::from(bucket) })
}
}
}
pub fn backend_name(&self) -> &'static str {
match self { Self::Local { .. } => "local", Self::S3 { .. } => "s3" }
}
pub async fn exists(&self, key: &str) -> Result<bool, StorageError> {
match self {
Self::Local { root } => Ok(root.join(key).is_file()),
Self::S3 { client, bucket } => match client.head_object().bucket(bucket.as_ref()).key(key).send().await {
Ok(_) => Ok(true),
Err(error) if error.as_service_error().is_some_and(|service| service.is_not_found()) => Ok(false),
Err(error) => Err(StorageError::Backend(error.to_string())),
},
}
}
pub async fn put(&self, key: &str, bytes: Bytes, content_type: &str, cache_control: &str) -> Result<(), StorageError> {
match self {
Self::Local { root } => {
let path = root.join(key);
if let Some(parent) = path.parent() { tokio::fs::create_dir_all(parent).await?; }
tokio::fs::write(path, bytes).await?;
Ok(())
}
Self::S3 { client, bucket } => {
client.put_object()
.bucket(bucket.as_ref())
.key(key)
.content_type(content_type)
.cache_control(cache_control)
.body(ByteStream::from(bytes))
.send().await
.map_err(|error| StorageError::Backend(error.to_string()))?;
Ok(())
}
}
}
pub async fn get(&self, key: &str) -> Result<Bytes, StorageError> {
match self {
Self::Local { root } => Ok(Bytes::from(tokio::fs::read(root.join(key)).await?)),
Self::S3 { client, bucket } => {
let output = client.get_object().bucket(bucket.as_ref()).key(key).send().await
.map_err(|error| StorageError::Backend(error.to_string()))?;
let bytes = output.body.collect().await
.map_err(|error| StorageError::Backend(error.to_string()))?
.into_bytes();
Ok(bytes)
}
}
}
pub async fn get_local_with_legacy(&self, key: &str, legacy_key: &str) -> Result<Bytes, StorageError> {
match self {
Self::Local { root } => {
let canonical = root.join(key);
let path = if canonical.is_file() { canonical } else { root.join(legacy_key) };
Ok(Bytes::from(tokio::fs::read(path).await?))
}
Self::S3 { .. } => self.get(key).await,
}
}
pub async fn delete(&self, key: &str) -> Result<(), StorageError> {
match self {
Self::Local { root } => match tokio::fs::remove_file(root.join(key)).await {
Ok(()) => Ok(()),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(error) => Err(error.into()),
},
Self::S3 { client, bucket } => {
client.delete_object().bucket(bucket.as_ref()).key(key).send().await
.map_err(|error| StorageError::Backend(error.to_string()))?;
Ok(())
}
}
}
}
#[derive(Debug)]
pub enum StorageError {
Io(std::io::Error),
Backend(String),
}
impl std::fmt::Display for StorageError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self { Self::Io(error) => write!(f, "{error}"), Self::Backend(error) => f.write_str(error) }
}
}
impl std::error::Error for StorageError {}
impl From<std::io::Error> for StorageError { fn from(value: std::io::Error) -> Self { Self::Io(value) } }
pub fn object_key(kind: &str, owner_id: i64, token: &str, filename: &str) -> String {
format!("{kind}/{owner_id}_{token}/{filename}")
}
pub fn legacy_key(owner_id: i64, token: &str, filename: &str) -> String {
format!("{owner_id}_{token}/{filename}")
}