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

S3 support #1

Merged
gru merged 6 commits from s3_support into master 2026-07-24 11:57:41 +02:00
Conversation 0 Commits 6 Files Changed 30
+2387 -712
12 changed files with 2387 additions and 712 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 33b4667e42 - Show all commits
dev.sh
+18 -8
View File
@@ -1,5 +1,6 @@
#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")"
mkdir -p data/db data/files
@@ -11,15 +12,24 @@ export FILES_DIR="${FILES_DIR:-$(pwd)/data/files}"
export UPLOAD_MAX_SIZE_MB="${UPLOAD_MAX_SIZE_MB:-20}"
export STATIC_DIR="${STATIC_DIR:-$(pwd)/static}"
export RUST_LOG="${RUST_LOG:-rustpad=debug,tower_http=info}"
# A new value on every run prevents stale HTML/JS cache issues.
# Generate a new asset version on each run to prevent stale HTML and JavaScript.
export ASSET_VERSION="${ASSET_VERSION:-dev-$(date +%s)}"
if command -v cargo >/dev/null 2>&1; then
exec cargo run
elif command -v docker >/dev/null 2>&1; then
export IMAGE_TAG="${IMAGE_TAG:-dev}"
exec docker compose up --build --force-recreate --remove-orphans
else
echo "Brak cargo i docker. Zainstaluj Rust 1.85+ albo Docker." >&2
exit 1
echo "Cleaning RustPad build artifacts..."
cargo clean --package rustpad
echo "Starting RustPad..."
exec cargo run --package rustpad
fi
if command -v docker >/dev/null 2>&1; then
export IMAGE_TAG="${IMAGE_TAG:-dev}"
echo "Starting RustPad with Docker..."
exec docker compose up --build --force-recreate --remove-orphans
fi
echo "Neither Cargo nor Docker was found. Install Rust 1.85+ or Docker." >&2
exit 1
src/api.rs
+393 -89
View File
@@ -1,12 +1,12 @@
use axum::{
extract::{Multipart, Path, State},
http::{header, HeaderMap, HeaderValue, StatusCode},
response::{IntoResponse, Response},
Json,
extract::{Multipart, Path, State},
http::{HeaderMap, HeaderValue, StatusCode, header},
response::{IntoResponse, Response},
};
use serde::{Deserialize, Serialize};
use chrono::{Duration, Utc};
use rand_core::{OsRng, RngCore};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use slug::slugify;
@@ -141,8 +141,18 @@ pub async fn create_workspace(
let slug = unique_workspace_slug(&state, title).await?;
let workspace = db::create_workspace(&state.db, &slug, title, password).await?;
if let Some(user) = crate::auth::optional_user(&state, &headers).await.map_err(|e| ApiError::forbidden(&e.message))? {
sqlx::query(queries::get(state.db.kind(), queries::USER_ATTACH_WORKSPACE)).bind(user.id).bind(&workspace.slug).execute(state.db.pool()).await?;
if let Some(user) = crate::auth::optional_user(&state, &headers)
.await
.map_err(|e| ApiError::forbidden(&e.message))?
{
sqlx::query(queries::get(
state.db.kind(),
queries::USER_ATTACH_WORKSPACE,
))
.bind(user.id)
.bind(&workspace.slug)
.execute(state.db.pool())
.await?;
}
Ok((
@@ -169,7 +179,13 @@ pub async fn open_workspace(
Path(workspace_slug): Path<String>,
Json(payload): Json<PasswordRequest>,
) -> Result<Json<WorkspaceOpenResponse>, ApiError> {
let workspace = authorized_workspace(&state, &workspace_slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
let workspace = authorized_workspace(
&state,
&workspace_slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
let notes = db::list_notes(&state.db, workspace.id)
.await?
.into_iter()
@@ -195,16 +211,37 @@ pub async fn create_note(
Path(workspace_slug): Path<String>,
Json(payload): Json<CreateNoteRequest>,
) -> Result<(StatusCode, Json<NoteListItem>), ApiError> {
let workspace = authorized_workspace(&state, &workspace_slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
let workspace = authorized_workspace(
&state,
&workspace_slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
let title = validate_name(&payload.name, "Note name")?;
let base = slugify(title);
if base.is_empty() {
return Err(ApiError::bad_request("The name cannot be converted into a valid address"));
return Err(ApiError::bad_request(
"The name cannot be converted into a valid address",
));
}
let slug = unique_note_slug(&state, workspace.id, &base).await?;
let created_by = payload.created_by.as_deref().map(str::trim).filter(|v| !v.is_empty()).map(|v| v.chars().take(40).collect::<String>());
let note = db::create_note(&state.db, workspace.id, &slug, title, payload.protect, created_by.as_deref()).await?;
let created_by = payload
.created_by
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
.map(|v| v.chars().take(40).collect::<String>());
let note = db::create_note(
&state.db,
workspace.id,
&slug,
title,
payload.protect,
created_by.as_deref(),
)
.await?;
Ok((
StatusCode::CREATED,
Json(NoteListItem {
@@ -248,7 +285,14 @@ pub async fn history(
Path((workspace_slug, note_slug)): Path<(String, String)>,
Json(payload): Json<PasswordRequest>,
) -> Result<Json<Vec<db::Revision>>, ApiError> {
let (workspace, note) = authorized_note(&state, &workspace_slug, &note_slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
let (workspace, note) = authorized_note(
&state,
&workspace_slug,
&note_slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
let _ = workspace;
let revisions = db::list_revisions(&state.db, note.id)
.await?
@@ -266,15 +310,29 @@ pub async fn restore(
Path((workspace_slug, note_slug)): Path<(String, String)>,
Json(payload): Json<RestoreRequest>,
) -> Result<Json<serde_json::Value>, ApiError> {
let (workspace, note) = authorized_note(&state, &workspace_slug, &note_slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
let (workspace, note) = authorized_note(
&state,
&workspace_slug,
&note_slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
let content: Option<String> = sqlx::query_scalar(queries::get(state.db.kind(), queries::Q028))
.bind(payload.revision_id)
.bind(note.id)
.fetch_optional(state.db.pool())
.await?;
let content = content.ok_or_else(ApiError::not_found_revision)?;
let (revision_id, updated_at) =
db::save_revision(&state.db, note.id, workspace.id, &content, Some("restore"), "[]").await?;
let (revision_id, updated_at) = db::save_revision(
&state.db,
note.id,
workspace.id,
&content,
Some("restore"),
"[]",
)
.await?;
let update = NoteUpdate {
content,
revision_id,
@@ -282,7 +340,10 @@ pub async fn restore(
author: Some("restore".into()),
owner_map: "[]".into(),
};
let _ = state.note_channel(&workspace_slug, &note_slug).await.send(RoomEvent::Document(update));
let _ = state
.note_channel(&workspace_slug, &note_slug)
.await
.send(RoomEvent::Document(update));
Ok(Json(serde_json::json!({"ok": true})))
}
@@ -296,8 +357,13 @@ pub async fn authorized_workspace(
.await?
.ok_or_else(ApiError::not_found_workspace)?;
let token_access = verify_resource_access_token(state, "workspace", slug, access_token).await?;
if workspace.is_private != 0 && !token_access { return Err(ApiError::forbidden("This workspace is private.")); }
if workspace.password_hash.is_some() && !db::verify_workspace_password(&workspace, password) && !token_access {
if workspace.is_private != 0 && !token_access {
return Err(ApiError::forbidden("This workspace is private."));
}
if workspace.password_hash.is_some()
&& !db::verify_workspace_password(&workspace, password)
&& !token_access
{
return Err(ApiError::unauthorized());
}
Ok(workspace)
@@ -353,7 +419,9 @@ fn validate_password(password: Option<&str>) -> Result<Option<&str>, ApiError> {
async fn unique_workspace_slug(state: &SharedState, title: &str) -> Result<String, ApiError> {
let base = slugify(title);
if base.is_empty() {
return Err(ApiError::bad_request("The name cannot be converted into a valid address"));
return Err(ApiError::bad_request(
"The name cannot be converted into a valid address",
));
}
let needs_suffix = base.chars().count() < MIN_WORKSPACE_SLUG_LENGTH
@@ -376,7 +444,10 @@ async fn unique_note_slug(
workspace_id: i64,
base: &str,
) -> Result<String, ApiError> {
if db::find_note(&state.db, workspace_id, base).await?.is_none() {
if db::find_note(&state.db, workspace_id, base)
.await?
.is_none()
{
return Ok(base.to_owned());
}
for _ in 0..8 {
@@ -391,7 +462,6 @@ async fn unique_note_slug(
Err(ApiError::internal("Failed to create a unique address"))
}
#[derive(Debug, Deserialize)]
pub struct CreatePadRequest {
name: String,
@@ -424,12 +494,21 @@ pub async fn create_pad(
let password = validate_password(payload.password.as_deref())?;
let base = slugify(title);
if base.is_empty() {
return Err(ApiError::bad_request("The name cannot be converted into a valid address"));
return Err(ApiError::bad_request(
"The name cannot be converted into a valid address",
));
}
let slug = unique_pad_slug(&state, &base).await?;
let pad = db::create_pad(&state.db, &slug, title, password).await?;
if let Some(user) = crate::auth::optional_user(&state, &headers).await.map_err(|e| ApiError::forbidden(&e.message))? {
sqlx::query(queries::get(state.db.kind(), queries::USER_ATTACH_PAD)).bind(user.id).bind(&pad.slug).execute(state.db.pool()).await?;
if let Some(user) = crate::auth::optional_user(&state, &headers)
.await
.map_err(|e| ApiError::forbidden(&e.message))?
{
sqlx::query(queries::get(state.db.kind(), queries::USER_ATTACH_PAD))
.bind(user.id)
.bind(&pad.slug)
.execute(state.db.pool())
.await?;
}
Ok((
StatusCode::CREATED,
@@ -462,10 +541,18 @@ pub async fn publish_pad_page(
Path(slug): Path<String>,
Json(payload): Json<PublishRequest>,
) -> Result<Json<PublishResponse>, ApiError> {
let pad = authorized_pad(&state, &slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
let pad = authorized_pad(
&state,
&slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
let token = db::publish_pad(&state.db, pad.id).await?;
db::set_pad_public_task_updates(&state.db, pad.id, payload.allow_task_updates).await?;
Ok(Json(PublishResponse { url: format!("/s/{token}") }))
Ok(Json(PublishResponse {
url: format!("/s/{token}"),
}))
}
pub async fn publish_note_page(
@@ -473,10 +560,19 @@ pub async fn publish_note_page(
Path((workspace_slug, note_slug)): Path<(String, String)>,
Json(payload): Json<PublishRequest>,
) -> Result<Json<PublishResponse>, ApiError> {
let (_, note) = authorized_note(&state, &workspace_slug, &note_slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
let (_, note) = authorized_note(
&state,
&workspace_slug,
&note_slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
let token = db::publish_note(&state.db, note.id).await?;
db::set_note_public_task_updates(&state.db, note.id, payload.allow_task_updates).await?;
Ok(Json(PublishResponse { url: format!("/s/{token}") }))
Ok(Json(PublishResponse {
url: format!("/s/{token}"),
}))
}
pub async fn public_page(
@@ -499,9 +595,17 @@ pub async fn update_public_task(
Path(token): Path<String>,
Json(payload): Json<PublicTaskUpdateRequest>,
) -> Result<Json<PublicPageResponse>, ApiError> {
let current = db::find_published_page(&state.db, &token).await?.ok_or_else(ApiError::not_found_note)?;
if !current.allow_task_updates { return Err(ApiError::forbidden("Task updates are disabled for this page")); }
let page = db::update_public_task(&state.db, &token, payload.source_line, payload.checked).await?.ok_or_else(ApiError::not_found_note)?;
let current = db::find_published_page(&state.db, &token)
.await?
.ok_or_else(ApiError::not_found_note)?;
if !current.allow_task_updates {
return Err(ApiError::forbidden(
"Task updates are disabled for this page",
));
}
let page = db::update_public_task(&state.db, &token, payload.source_line, payload.checked)
.await?
.ok_or_else(ApiError::not_found_note)?;
Ok(Json(PublicPageResponse {
title: page.title,
content: page.content,
@@ -515,7 +619,13 @@ pub async fn pad_history(
Path(slug): Path<String>,
Json(payload): Json<PasswordRequest>,
) -> Result<Json<Vec<db::Revision>>, ApiError> {
let pad = authorized_pad(&state, &slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
let pad = authorized_pad(
&state,
&slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
let revisions = db::list_pad_revisions(&state.db, pad.id)
.await?
.into_iter()
@@ -532,20 +642,28 @@ pub async fn pad_restore(
Path(slug): Path<String>,
Json(payload): Json<RestoreRequest>,
) -> Result<Json<serde_json::Value>, ApiError> {
let pad = authorized_pad(&state, &slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
let pad = authorized_pad(
&state,
&slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
let content: Option<String> = sqlx::query_scalar(queries::get(state.db.kind(), queries::Q029))
.bind(payload.revision_id)
.bind(pad.id)
.fetch_optional(state.db.pool())
.await?;
let content = content.ok_or_else(ApiError::not_found_revision)?;
let owner_map: Option<String> = sqlx::query_scalar(queries::get(state.db.kind(), queries::Q030))
let owner_map: Option<String> =
sqlx::query_scalar(queries::get(state.db.kind(), queries::Q030))
.bind(payload.revision_id)
.bind(pad.id)
.fetch_optional(state.db.pool())
.await?;
let owner_map = owner_map.unwrap_or_else(|| "[]".into());
let (revision_id, updated_at) = db::save_pad_revision(&state.db, pad.id, &content, Some("restore"), &owner_map).await?;
let (revision_id, updated_at) =
db::save_pad_revision(&state.db, pad.id, &content, Some("restore"), &owner_map).await?;
let update = NoteUpdate {
content,
revision_id,
@@ -553,7 +671,10 @@ pub async fn pad_restore(
author: Some("restore".into()),
owner_map,
};
let _ = state.pad_channel(&slug).await.send(RoomEvent::Document(update));
let _ = state
.pad_channel(&slug)
.await
.send(RoomEvent::Document(update));
Ok(Json(serde_json::json!({"ok": true})))
}
@@ -567,7 +688,9 @@ async fn authorized_pad(
.await?
.ok_or_else(ApiError::not_found_note)?;
let token_access = verify_resource_access_token(state, "pad", slug, access_token).await?;
if pad.is_private != 0 && !token_access { return Err(ApiError::forbidden("This note is private.")); }
if pad.is_private != 0 && !token_access {
return Err(ApiError::forbidden("This note is private."));
}
if pad.password_hash.is_some() && !db::verify_pad_password(&pad, password) && !token_access {
return Err(ApiError::unauthorized());
}
@@ -587,8 +710,6 @@ async fn unique_pad_slug(state: &SharedState, base: &str) -> Result<String, ApiE
Err(ApiError::internal("Failed to create a unique address"))
}
pub async fn upload_pad_file(
State(state): State<SharedState>,
Path(slug): Path<String>,
@@ -597,15 +718,32 @@ pub async fn upload_pad_file(
let mut password: Option<String> = None;
let mut access_token: Option<String> = None;
let mut file: Option<(String, Vec<u8>)> = None;
while let Some(field) = multipart.next_field().await.map_err(|_| ApiError::bad_request("Invalid form data"))? {
while let Some(field) = multipart
.next_field()
.await
.map_err(|_| ApiError::bad_request("Invalid form data"))?
{
let name = field.name().unwrap_or_default().to_owned();
if name == "password" {
password = Some(field.text().await.map_err(|_| ApiError::bad_request("Invalid password"))?);
password = Some(
field
.text()
.await
.map_err(|_| ApiError::bad_request("Invalid password"))?,
);
} else if name == "access_token" {
access_token = Some(field.text().await.map_err(|_| ApiError::bad_request("Invalid access token"))?);
access_token = Some(
field
.text()
.await
.map_err(|_| ApiError::bad_request("Invalid access token"))?,
);
} else if name == "file" {
let filename = field.file_name().unwrap_or("plik").to_owned();
let bytes = field.bytes().await.map_err(|_| ApiError::bad_request("Failed to read the file"))?;
let bytes = field
.bytes()
.await
.map_err(|_| ApiError::bad_request("Failed to read the file"))?;
if bytes.len() > state.upload_max_size_bytes {
return Err(ApiError::payload_too_large(state.upload_max_size_bytes));
}
@@ -618,16 +756,33 @@ pub async fn upload_pad_file(
let file_token = db::pad_file_token(&state.db, pad.id).await?;
let mut stored = safe.clone();
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();
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);
key = crate::storage::object_key("pads", pad.id, &file_token, &stored);
}
let url = format!("/f/{}/{}", file_token, stored);
let mime = mime_guess::from_path(&stored).first_or_octet_stream().to_string();
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
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})))
@@ -638,14 +793,24 @@ pub async fn pad_files(
Path(slug): Path<String>,
Json(payload): Json<PasswordRequest>,
) -> Result<Json<Vec<db::NoteFile>>, ApiError> {
let pad = authorized_pad(&state, &slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
let pad = authorized_pad(
&state,
&slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
let mut files = db::list_pad_files(&state.db, pad.id).await?;
for file in &mut files {
let attached = pad.content.contains(&file.url);
if attached != file.is_attached {
db::set_pad_file_attached(&state.db, file.id, attached).await?;
file.is_attached = attached;
file.detached_at = if attached { None } else { Some(chrono::Utc::now().to_rfc3339()) };
file.detached_at = if attached {
None
} else {
Some(chrono::Utc::now().to_rfc3339())
};
}
file.created_at = db::normalize_timestamp(&file.created_at);
}
@@ -660,50 +825,101 @@ pub async fn upload_note_file(
let mut password: Option<String> = None;
let mut access_token: Option<String> = None;
let mut file: Option<(String, Vec<u8>)> = None;
while let Some(field) = multipart.next_field().await.map_err(|_| ApiError::bad_request("Invalid form data"))? {
while let Some(field) = multipart
.next_field()
.await
.map_err(|_| ApiError::bad_request("Invalid form data"))?
{
let name = field.name().unwrap_or_default().to_owned();
if name == "password" {
password = Some(field.text().await.map_err(|_| ApiError::bad_request("Invalid password"))?);
password = Some(
field
.text()
.await
.map_err(|_| ApiError::bad_request("Invalid password"))?,
);
} else if name == "access_token" {
access_token = Some(field.text().await.map_err(|_| ApiError::bad_request("Invalid access token"))?);
access_token = Some(
field
.text()
.await
.map_err(|_| ApiError::bad_request("Invalid access token"))?,
);
} else if name == "file" {
let filename = field.file_name().unwrap_or("plik").to_owned();
let bytes = field.bytes().await.map_err(|_| ApiError::bad_request("Failed to read the file"))?;
let bytes = field
.bytes()
.await
.map_err(|_| ApiError::bad_request("Failed to read the file"))?;
if bytes.len() > state.upload_max_size_bytes {
return Err(ApiError::payload_too_large(state.upload_max_size_bytes));
}
file = Some((filename, bytes.to_vec()));
}
}
let (_workspace, note) = authorized_note(&state, &workspace_slug, &note_slug, password.as_deref(), access_token.as_deref()).await?;
let (_workspace, note) = authorized_note(
&state,
&workspace_slug,
&note_slug,
password.as_deref(),
access_token.as_deref(),
)
.await?;
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 mut stored = safe.clone();
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();
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);
key = crate::storage::object_key("notes", note.id, &file_token, &stored);
}
let url = format!("/f/{}/{}", file_token, stored);
let mime = mime_guess::from_path(&stored).first_or_octet_stream().to_string();
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
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})))
}
pub async fn delete_note(
State(state): State<SharedState>,
Path((workspace_slug, note_slug)): Path<(String, String)>,
Json(payload): Json<PasswordRequest>,
) -> Result<Json<serde_json::Value>, ApiError> {
let (_workspace, note) = authorized_note(&state, &workspace_slug, &note_slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
if note.protected { return Err(ApiError::bad_request("This note is protected and cannot be deleted")); }
let (_workspace, note) = authorized_note(
&state,
&workspace_slug,
&note_slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
if note.protected {
return Err(ApiError::bad_request(
"This note is protected and cannot be deleted",
));
}
db::delete_note(&state.db, note.id).await?;
Ok(Json(serde_json::json!({"ok": true})))
}
@@ -713,14 +929,25 @@ pub async fn note_files(
Path((workspace_slug, note_slug)): Path<(String, String)>,
Json(payload): Json<PasswordRequest>,
) -> Result<Json<Vec<db::NoteFile>>, ApiError> {
let (_workspace, note) = authorized_note(&state, &workspace_slug, &note_slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
let (_workspace, note) = authorized_note(
&state,
&workspace_slug,
&note_slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
let mut files = db::list_note_files(&state.db, note.id).await?;
for file in &mut files {
let attached = note.content.contains(&file.url);
if attached != file.is_attached {
db::set_note_file_attached(&state.db, file.id, attached).await?;
file.is_attached = attached;
file.detached_at = if attached { None } else { Some(chrono::Utc::now().to_rfc3339()) };
file.detached_at = if attached {
None
} else {
Some(chrono::Utc::now().to_rfc3339())
};
}
file.created_at = db::normalize_timestamp(&file.created_at);
}
@@ -732,16 +959,39 @@ pub async fn delete_note_file(
Path((workspace_slug, note_slug, file_id)): Path<(String, String, i64)>,
Json(payload): Json<PasswordRequest>,
) -> Result<Json<serde_json::Value>, ApiError> {
let (workspace, note) = authorized_note(&state, &workspace_slug, &note_slug, payload.password.as_deref(), payload.access_token.as_deref()).await?;
if workspace.password_hash.is_none() || payload.password.as_deref().unwrap_or_default().is_empty() {
let (workspace, note) = authorized_note(
&state,
&workspace_slug,
&note_slug,
payload.password.as_deref(),
payload.access_token.as_deref(),
)
.await?;
if workspace.password_hash.is_none()
|| payload.password.as_deref().unwrap_or_default().is_empty()
{
return Err(ApiError::unauthorized());
}
let file = db::find_note_file(&state.db, note.id, file_id).await?
let file = db::find_note_file(&state.db, note.id, file_id)
.await?
.ok_or_else(ApiError::not_found_file)?;
let relative = file.url.trim_start_matches('/').split('/').collect::<Vec<_>>();
let relative = file
.url
.trim_start_matches('/')
.split('/')
.collect::<Vec<_>>();
if relative.len() == 3 && relative[0] == "f" {
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"))?;
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})))
@@ -762,7 +1012,8 @@ pub async fn download_legacy_file(
return Err(ApiError::not_found_file());
};
let id: i64 = id_part.parse().map_err(|_| ApiError::not_found_file())?;
let owner = db::find_file_owner(&state.db, token).await?
let owner = db::find_file_owner(&state.db, token)
.await?
.ok_or_else(ApiError::not_found_file)?;
if owner.id != id {
return Err(ApiError::not_found_file());
@@ -770,12 +1021,17 @@ pub async fn download_legacy_file(
serve_token_file(&state, token, &filename).await
}
async fn serve_token_file(state: &SharedState, token: &str, filename: &str) -> Result<Response, ApiError> {
async fn serve_token_file(
state: &SharedState,
token: &str,
filename: &str,
) -> Result<Response, ApiError> {
let safe = sanitize_filename(filename);
if safe != filename {
return Err(ApiError::not_found_file());
}
let owner = db::find_file_owner(&state.db, token).await?
let owner = db::find_file_owner(&state.db, token)
.await?
.ok_or_else(ApiError::not_found_file)?;
let kind = match owner.kind {
db::FileOwnerKind::Pad => "pads",
@@ -783,27 +1039,53 @@ async fn serve_token_file(state: &SharedState, token: &str, filename: &str) -> R
};
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
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(
header::CONTENT_TYPE,
HeaderValue::from_str(mime.as_ref()).unwrap_or_else(|_| HeaderValue::from_static("application/octet-stream")),
HeaderValue::from_str(mime.as_ref())
.unwrap_or_else(|_| HeaderValue::from_static("application/octet-stream")),
);
response.headers_mut().insert(
header::X_CONTENT_TYPE_OPTIONS,
HeaderValue::from_static("nosniff"),
);
response.headers_mut().insert(header::X_CONTENT_TYPE_OPTIONS, HeaderValue::from_static("nosniff"));
response.headers_mut().insert(
header::CACHE_CONTROL,
HeaderValue::from_str(&format!("public, max-age={}", state.file_cache_max_age_seconds))
HeaderValue::from_str(&format!(
"public, max-age={}",
state.file_cache_max_age_seconds
))
.expect("valid file cache-control header"),
);
Ok(response)
}
fn sanitize_filename(value: &str) -> String {
let name = std::path::Path::new(value).file_name().and_then(|v| v.to_str()).unwrap_or("plik");
let clean: String = name.chars().map(|c| if c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_') { c } else { '_' }).collect();
if clean.is_empty() || clean == "." || clean == ".." { "plik".into() } else { clean.chars().take(160).collect() }
let name = std::path::Path::new(value)
.file_name()
.and_then(|v| v.to_str())
.unwrap_or("plik");
let clean: String = name
.chars()
.map(|c| {
if c.is_ascii_alphanumeric() || matches!(c, '.' | '-' | '_') {
c
} else {
'_'
}
})
.collect();
if clean.is_empty() || clean == "." || clean == ".." {
"plik".into()
} else {
clean.chars().take(160).collect()
}
}
#[derive(Debug, Deserialize)]
@@ -827,13 +1109,17 @@ pub async fn create_resource_access_token(
let slug = payload.slug.trim();
match kind {
"workspace" => {
let workspace = db::find_workspace(&state.db, slug).await?.ok_or_else(ApiError::not_found_workspace)?;
let workspace = db::find_workspace(&state.db, slug)
.await?
.ok_or_else(ApiError::not_found_workspace)?;
if !db::verify_workspace_password(&workspace, Some(payload.password.as_str())) {
return Err(ApiError::unauthorized());
}
}
"pad" => {
let pad = db::find_pad(&state.db, slug).await?.ok_or_else(ApiError::not_found_note)?;
let pad = db::find_pad(&state.db, slug)
.await?
.ok_or_else(ApiError::not_found_note)?;
if !db::verify_pad_password(&pad, Some(payload.password.as_str())) {
return Err(ApiError::unauthorized());
}
@@ -844,7 +1130,8 @@ pub async fn create_resource_access_token(
let mut bytes = [0u8; 32];
OsRng.fill_bytes(&mut bytes);
let token = hex::encode(bytes);
let expires_at = (Utc::now() + Duration::days(state.anonymous_access_token_ttl_days)).to_rfc3339();
let expires_at =
(Utc::now() + Duration::days(state.anonymous_access_token_ttl_days)).to_rfc3339();
sqlx::query(queries::get(state.db.kind(), "INSERT INTO resource_access_tokens (token_hash, resource_kind, resource_slug, expires_at) VALUES (?, ?, ?, ?)"))
.bind(hash_access_token(&token))
.bind(kind)
@@ -852,7 +1139,10 @@ pub async fn create_resource_access_token(
.bind(&expires_at)
.execute(state.db.pool())
.await?;
Ok(Json(AccessTokenResponse { access_token: token, expires_at }))
Ok(Json(AccessTokenResponse {
access_token: token,
expires_at,
}))
}
pub async fn verify_resource_access_token(
@@ -864,7 +1154,11 @@ pub async fn verify_resource_access_token(
let Some(token) = token.map(str::trim).filter(|value| !value.is_empty()) else {
return Ok(false);
};
if crate::auth::resource_permission(state, kind, slug, Some(token)).await.map_err(|error| ApiError::forbidden(&error.message))?.is_some() {
if crate::auth::resource_permission(state, kind, slug, Some(token))
.await
.map_err(|error| ApiError::forbidden(&error.message))?
.is_some()
{
return Ok(true);
}
let count: i64 = sqlx::query_scalar(queries::get(state.db.kind(), "SELECT COUNT(*) FROM resource_access_tokens WHERE token_hash = ? AND resource_kind = ? AND resource_slug = ? AND expires_at > ?"))
@@ -901,7 +1195,10 @@ impl ApiError {
}
}
fn not_found_file() -> Self {
Self { status: StatusCode::NOT_FOUND, message: "File not found".into() }
Self {
status: StatusCode::NOT_FOUND,
message: "File not found".into(),
}
}
fn unauthorized() -> Self {
Self {
@@ -910,7 +1207,10 @@ impl ApiError {
}
}
fn forbidden(message: &str) -> Self {
Self { status: StatusCode::FORBIDDEN, message: message.into() }
Self {
status: StatusCode::FORBIDDEN,
message: message.into(),
}
}
fn not_found_workspace() -> Self {
Self {
@@ -947,6 +1247,10 @@ impl From<sqlx::Error> for ApiError {
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
(self.status, Json(serde_json::json!({"error": self.message}))).into_response()
(
self.status,
Json(serde_json::json!({"error": self.message})),
)
.into_response()
}
}
src/app.rs
+114 -40
View File
@@ -1,17 +1,22 @@
use axum::{
Router,
extract::{DefaultBodyLimit, Path, State},
http::{header, HeaderValue, StatusCode},
http::{HeaderValue, StatusCode, header},
response::{Html, IntoResponse, Response},
routing::{get, post},
Router,
};
use tower::{service_fn, ServiceBuilder};
use tower::{ServiceBuilder, service_fn};
use tower_http::{services::ServeDir, set_header::SetResponseHeaderLayer, trace::TraceLayer};
use crate::{api, auth, db, state::SharedState, websocket};
use std::convert::Infallible;
pub fn router(state: SharedState, static_dir: &str, upload_max_size_bytes: usize, asset_cache_max_age_seconds: u64) -> Router {
pub fn router(
state: SharedState,
static_dir: &str,
upload_max_size_bytes: usize,
asset_cache_max_age_seconds: u64,
) -> Router {
let asset_version = state.asset_version.clone();
let asset_not_found = service_fn(move |_request| {
let asset_version = asset_version.clone();
@@ -28,7 +33,8 @@ pub fn router(state: SharedState, static_dir: &str, upload_max_size_bytes: usize
}
});
let asset_cache_control = HeaderValue::from_str(&format!("public, max-age={asset_cache_max_age_seconds}"))
let asset_cache_control =
HeaderValue::from_str(&format!("public, max-age={asset_cache_max_age_seconds}"))
.expect("valid asset cache-control header");
Router::new()
@@ -40,7 +46,10 @@ pub fn router(state: SharedState, static_dir: &str, upload_max_size_bytes: usize
.route("/errors/private-workspace", get(private_workspace_error))
.route("/health", get(health))
.route("/f/{token}/{filename}", get(api::download_file))
.route("/files/{directory}/{filename}", get(api::download_legacy_file))
.route(
"/files/{directory}/{filename}",
get(api::download_legacy_file),
)
.route("/api/auth/identity", post(auth::identity))
.route("/api/access-token", post(api::create_resource_access_token))
.route("/api/auth/register", post(auth::register))
@@ -48,13 +57,37 @@ pub fn router(state: SharedState, static_dir: &str, upload_max_size_bytes: usize
.route("/api/auth/confirm-account", post(auth::confirm_account))
.route("/api/auth/me", get(auth::me))
.route("/api/auth/logout", post(auth::logout))
.route("/api/auth/resources", get(auth::resources).put(auth::update_resource).delete(auth::delete_resource))
.route("/api/auth/resources/privacy", post(auth::set_resource_privacy))
.route("/api/auth/resources/sharing", get(auth::resource_sharing).post(auth::share_resource_users).delete(auth::remove_resource_user))
.route("/api/auth/resources/share-links", post(auth::create_share_link).put(auth::update_share_link).delete(auth::revoke_share_link))
.route("/share-invitations/{token}/accept", get(auth::accept_share_invitation))
.route(
"/api/auth/resources",
get(auth::resources)
.put(auth::update_resource)
.delete(auth::delete_resource),
)
.route(
"/api/auth/resources/privacy",
post(auth::set_resource_privacy),
)
.route(
"/api/auth/resources/sharing",
get(auth::resource_sharing)
.post(auth::share_resource_users)
.delete(auth::remove_resource_user),
)
.route(
"/api/auth/resources/share-links",
post(auth::create_share_link)
.put(auth::update_share_link)
.delete(auth::revoke_share_link),
)
.route(
"/share-invitations/{token}/accept",
get(auth::accept_share_invitation),
)
.route("/api/auth/password-reset", post(auth::request_reset))
.route("/api/auth/password-reset/confirm", post(auth::confirm_reset))
.route(
"/api/auth/password-reset/confirm",
post(auth::confirm_reset),
)
.route("/api/public/{token}", get(api::public_page))
.route("/api/public/{token}/tasks", post(api::update_public_task))
.route("/api/pads", post(api::create_pad))
@@ -62,11 +95,20 @@ pub fn router(state: SharedState, static_dir: &str, upload_max_size_bytes: usize
.route("/api/pads/{slug}/history", post(api::pad_history))
.route("/api/pads/{slug}/publish", post(api::publish_pad_page))
.route("/api/pads/{slug}/restore", post(api::pad_restore))
.route("/api/pads/{slug}/files", post(api::upload_pad_file).put(api::pad_files))
.route(
"/api/pads/{slug}/files",
post(api::upload_pad_file).put(api::pad_files),
)
.route("/api/workspaces", post(api::create_workspace))
.route("/api/workspaces/{workspace_slug}", get(api::workspace_info))
.route("/api/workspaces/{workspace_slug}/open", post(api::open_workspace))
.route("/api/workspaces/{workspace_slug}/notes", post(api::create_note))
.route(
"/api/workspaces/{workspace_slug}/open",
post(api::open_workspace),
)
.route(
"/api/workspaces/{workspace_slug}/notes",
post(api::create_note),
)
.route(
"/api/workspaces/{workspace_slug}/notes/{note_slug}",
get(api::note_info).delete(api::delete_note),
@@ -92,10 +134,7 @@ pub fn router(state: SharedState, static_dir: &str, upload_max_size_bytes: usize
axum::routing::delete(api::delete_note_file),
)
.route("/ws/p/{slug}", get(websocket::upgrade_pad))
.route(
"/ws/{workspace_slug}/{note_slug}",
get(websocket::upgrade),
)
.route("/ws/{workspace_slug}/{note_slug}", get(websocket::upgrade))
.route("/static", get(static_not_found))
.route("/static/{*path}", get(static_not_found))
.nest_service(
@@ -109,12 +148,13 @@ pub fn router(state: SharedState, static_dir: &str, upload_max_size_bytes: usize
)
.fallback(not_found)
.method_not_allowed_fallback(method_not_allowed)
.layer(DefaultBodyLimit::max(upload_max_size_bytes.saturating_add(1024 * 1024)))
.layer(DefaultBodyLimit::max(
upload_max_size_bytes.saturating_add(1024 * 1024),
))
.layer(TraceLayer::new_for_http())
.with_state(state)
}
async fn private_workspace_error(State(state): State<SharedState>) -> Response {
error_response(
StatusCode::FORBIDDEN,
@@ -132,19 +172,26 @@ async fn health() -> &'static str {
}
async fn home(State(state): State<SharedState>) -> Response {
versioned_html(include_str!("../static/home.html"), &state.asset_version, state.registration_enabled, &state.frontend_log_level)
versioned_html(
include_str!("../static/home.html"),
&state.asset_version,
state.registration_enabled,
&state.frontend_log_level,
)
}
async fn pad(
State(state): State<SharedState>,
Path(slug): Path<String>,
) -> Response {
async fn pad(State(state): State<SharedState>, Path(slug): Path<String>) -> Response {
match db::find_pad(&state.db, &slug).await {
Ok(Some(pad)) => {
let html = include_str!("../static/pad.html")
.replace("__PAD_TITLE__", &escape_html(&pad.title));
versioned_html(&html, &state.asset_version, state.registration_enabled, &state.frontend_log_level)
},
versioned_html(
&html,
&state.asset_version,
state.registration_enabled,
&state.frontend_log_level,
)
}
Ok(None) => error_response(
StatusCode::NOT_FOUND,
"404",
@@ -161,12 +208,14 @@ async fn pad(
}
}
async fn public_page(
State(state): State<SharedState>,
Path(token): Path<String>,
) -> Response {
async fn public_page(State(state): State<SharedState>, Path(token): Path<String>) -> Response {
match db::find_published_page(&state.db, &token).await {
Ok(Some(_)) => versioned_html(include_str!("../static/public.html"), &state.asset_version, state.registration_enabled, &state.frontend_log_level),
Ok(Some(_)) => versioned_html(
include_str!("../static/public.html"),
&state.asset_version,
state.registration_enabled,
&state.frontend_log_level,
),
Ok(None) => error_response(
StatusCode::NOT_FOUND,
"404",
@@ -191,8 +240,13 @@ async fn workspace(
Ok(Some(workspace)) => {
let html = include_str!("../static/workspace.html")
.replace("__WORKSPACE_TITLE__", &escape_html(&workspace.title));
versioned_html(&html, &state.asset_version, state.registration_enabled, &state.frontend_log_level)
},
versioned_html(
&html,
&state.asset_version,
state.registration_enabled,
&state.frontend_log_level,
)
}
Ok(None) => error_response(
StatusCode::NOT_FOUND,
"404",
@@ -238,8 +292,13 @@ async fn note(
.replace("__NOTE_TITLE__", &escape_html(&note.title))
.replace("__WORKSPACE_TITLE__", &escape_html(&workspace.title))
.replace("__WORKSPACE_SLUG__", &escape_html(&workspace_slug));
versioned_html(&html, &state.asset_version, state.registration_enabled, &state.frontend_log_level)
},
versioned_html(
&html,
&state.asset_version,
state.registration_enabled,
&state.frontend_log_level,
)
}
Ok(None) => error_response(
StatusCode::NOT_FOUND,
"404",
@@ -326,14 +385,26 @@ fn error_response(
response
}
fn versioned_html(template: &str, asset_version: &str, registration_enabled: bool, frontend_log_level: &str) -> Response {
fn versioned_html(
template: &str,
asset_version: &str,
registration_enabled: bool,
frontend_log_level: &str,
) -> Response {
let frontend_config = format!(
r#"<script>window.__RUSTPAD_CONFIG__=Object.freeze({{frontendLogLevel:"{}"}});</script>"#,
escape_js_string(frontend_log_level),
);
let html = template
.replace("__ASSET_VERSION__", asset_version)
.replace("__REGISTRATION_ENABLED__", if registration_enabled { "true" } else { "false" })
.replace(
"__REGISTRATION_ENABLED__",
if registration_enabled {
"true"
} else {
"false"
},
)
.replace("</head>", &format!("{frontend_config}</head>"));
let mut response = Html(html).into_response();
no_store(&mut response);
@@ -341,7 +412,10 @@ fn versioned_html(template: &str, asset_version: &str, registration_enabled: boo
}
fn escape_js_string(value: &str) -> String {
value.replace('\\', "\\\\").replace('"', "\\\"").replace('<', "\\u003c")
value
.replace('\\', "\\\\")
.replace('"', "\\\"")
.replace('<', "\\u003c")
}
fn no_store(response: &mut Response) {
src/auth.rs
+839 -216
View File
File diff suppressed because it is too large Load Diff
src/config.rs
+26 -16
View File
@@ -26,16 +26,21 @@ impl Config {
pub fn from_env() -> Result<Self, Box<dyn std::error::Error>> {
let host = env_var("APP_HOST", "127.0.0.1").parse()?;
let port = env_var("APP_PORT", "3000").parse()?;
let database_max_connections =
env_var("DATABASE_MAX_CONNECTIONS", "8").parse()?;
let database_max_connections = env_var("DATABASE_MAX_CONNECTIONS", "8").parse()?;
let upload_max_size_mb: usize =
env_var("UPLOAD_MAX_SIZE_MB", "20").parse()?;
let anonymous_access_token_ttl_days = env_positive_i64("ANONYMOUS_ACCESS_TOKEN_TTL_DAYS", 7)?;
let upload_max_size_mb: usize = 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() },
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"),
@@ -51,25 +56,28 @@ impl Config {
return Err("UPLOAD_MAX_SIZE_MB must be greater than 0".into());
}
let smtp_host = std::env::var("SMTP_HOST").ok().filter(|v| !v.trim().is_empty());
let smtp_host = std::env::var("SMTP_HOST")
.ok()
.filter(|v| !v.trim().is_empty());
let smtp = if let Some(host) = smtp_host {
Some(crate::state::SmtpConfig {
host,
port: env_var("SMTP_PORT", "587").parse()?,
username: std::env::var("SMTP_USERNAME").unwrap_or_default(),
password: std::env::var("SMTP_PASSWORD").unwrap_or_default(),
from: std::env::var("SMTP_FROM").map_err(|_| "SMTP_FROM is required when SMTP_HOST is set")?,
public_url: std::env::var("PUBLIC_URL").map_err(|_| "PUBLIC_URL is required when SMTP_HOST is set")?,
from: std::env::var("SMTP_FROM")
.map_err(|_| "SMTP_FROM is required when SMTP_HOST is set")?,
public_url: std::env::var("PUBLIC_URL")
.map_err(|_| "PUBLIC_URL is required when SMTP_HOST is set")?,
})
} else { None };
} else {
None
};
Ok(Self {
host,
port,
database_url: env_var(
"DATABASE_URL",
"sqlite:///data/db/rustpad.db?mode=rwc",
),
database_url: env_var("DATABASE_URL", "sqlite:///data/db/rustpad.db?mode=rwc"),
database_max_connections,
static_dir: env_var("STATIC_DIR", "static"),
files_dir,
@@ -128,6 +136,8 @@ fn env_nonnegative_u64(name: &str, default: u64) -> Result<u64, Box<dyn std::err
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()); }
if value.trim().is_empty() {
return Err(format!("{name} cannot be empty when STORAGE_DRIVER=s3").into());
}
Ok(value)
}
src/database.rs
+21 -8
View File
@@ -1,5 +1,5 @@
use crate::queries;
use sqlx::{any::AnyPoolOptions, AnyPool};
use sqlx::{AnyPool, any::AnyPoolOptions};
use tracing::{debug, info};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -26,9 +26,15 @@ impl Database {
.await?;
if kind == DatabaseKind::Sqlite {
debug!("applying SQLite connection pragmas");
sqlx::query(queries::SQLITE_FOREIGN_KEYS_ON).execute(&pool).await?;
sqlx::query(queries::SQLITE_JOURNAL_WAL).execute(&pool).await?;
sqlx::query(queries::SQLITE_BUSY_TIMEOUT).execute(&pool).await?;
sqlx::query(queries::SQLITE_FOREIGN_KEYS_ON)
.execute(&pool)
.await?;
sqlx::query(queries::SQLITE_JOURNAL_WAL)
.execute(&pool)
.await?;
sqlx::query(queries::SQLITE_BUSY_TIMEOUT)
.execute(&pool)
.await?;
}
info!(?kind, max_connections, "database pool ready");
Ok(Self { pool, kind })
@@ -44,9 +50,16 @@ impl Database {
impl DatabaseKind {
fn from_url(url: &str) -> Result<Self, sqlx::Error> {
if url.starts_with("sqlite:") { Ok(Self::Sqlite) }
else if url.starts_with("postgres:") || url.starts_with("postgresql:") { Ok(Self::Postgres) }
else if url.starts_with("mysql:") { Ok(Self::MySql) }
else { Err(sqlx::Error::Configuration("DATABASE_URL must use sqlite://, postgres:// or mysql://".into())) }
if url.starts_with("sqlite:") {
Ok(Self::Sqlite)
} else if url.starts_with("postgres:") || url.starts_with("postgresql:") {
Ok(Self::Postgres)
} else if url.starts_with("mysql:") {
Ok(Self::MySql)
} else {
Err(sqlx::Error::Configuration(
"DATABASE_URL must use sqlite://, postgres:// or mysql://".into(),
))
}
}
}
src/db.rs
+227 -60
View File
@@ -1,15 +1,19 @@
use argon2::{
password_hash::SaltString, Argon2, PasswordHash, PasswordHasher, PasswordVerifier,
use crate::{
database::{Database, DatabaseKind},
queries,
};
use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier, password_hash::SaltString};
use chrono::{DateTime, NaiveDateTime, Utc};
use rand_core::{OsRng, RngCore};
use serde::Serialize;
use sqlx::FromRow;
use crate::{database::{Database, DatabaseKind}, queries};
use sqlx::{Any, Transaction};
async fn inserted_id(kind: DatabaseKind, tx: &mut Transaction<'_, Any>, table: &str) -> Result<i64, sqlx::Error> {
async fn inserted_id(
kind: DatabaseKind,
tx: &mut Transaction<'_, Any>,
table: &str,
) -> Result<i64, sqlx::Error> {
let query = match kind {
DatabaseKind::Sqlite => queries::SQLITE_LAST_INSERT_ID,
DatabaseKind::MySql => queries::MYSQL_LAST_INSERT_ID,
@@ -103,7 +107,9 @@ pub async fn create_workspace(
title: &str,
password: Option<&str>,
) -> Result<Workspace, sqlx::Error> {
let password_hash = password.filter(|value| !value.is_empty()).map(hash_password);
let password_hash = password
.filter(|value| !value.is_empty())
.map(hash_password);
sqlx::query(queries::get(pool.kind(), queries::Q002))
.bind(slug)
.bind(title)
@@ -118,7 +124,10 @@ pub async fn create_workspace(
}
pub fn verify_workspace_password(workspace: &Workspace, password: Option<&str>) -> bool {
match (&workspace.password_hash, password.filter(|value| !value.is_empty())) {
match (
&workspace.password_hash,
password.filter(|value| !value.is_empty()),
) {
(None, _) => true,
(Some(hash), Some(password)) => PasswordHash::new(hash)
.ok()
@@ -265,7 +274,9 @@ pub fn normalize_timestamp(value: &str) -> String {
let offset_start = postgres.len() - 3;
let offset = &postgres[offset_start..];
if (offset.starts_with('+') || offset.starts_with('-'))
&& offset[1..].chars().all(|character| character.is_ascii_digit())
&& offset[1..]
.chars()
.all(|character| character.is_ascii_digit())
{
postgres.push_str(":00");
}
@@ -316,7 +327,9 @@ pub async fn create_pad(
title: &str,
password: Option<&str>,
) -> Result<Pad, sqlx::Error> {
let password_hash = password.filter(|value| !value.is_empty()).map(hash_password);
let password_hash = password
.filter(|value| !value.is_empty())
.map(hash_password);
sqlx::query(queries::get(pool.kind(), queries::Q012))
.bind(slug)
.bind(title)
@@ -331,7 +344,10 @@ pub async fn create_pad(
}
pub fn verify_pad_password(pad: &Pad, password: Option<&str>) -> bool {
match (&pad.password_hash, password.filter(|value| !value.is_empty())) {
match (
&pad.password_hash,
password.filter(|value| !value.is_empty()),
) {
(None, _) => true,
(Some(hash), Some(password)) => PasswordHash::new(hash)
.ok()
@@ -407,7 +423,6 @@ struct PublishedPageRow {
updated_at: String,
}
#[derive(Debug, Clone, FromRow)]
struct PostgresPublishedPageRow {
token: String,
@@ -421,7 +436,15 @@ struct PostgresPublishedPageRow {
impl From<PostgresPublishedPageRow> for PublishedPage {
fn from(value: PostgresPublishedPageRow) -> Self {
Self { token: value.token, pad_id: value.pad_id, note_id: value.note_id, allow_task_updates: value.allow_task_updates, title: value.title, content: value.content, updated_at: value.updated_at }
Self {
token: value.token,
pad_id: value.pad_id,
note_id: value.note_id,
allow_task_updates: value.allow_task_updates,
title: value.title,
content: value.content,
updated_at: value.updated_at,
}
}
}
impl From<PublishedPageRow> for PublishedPage {
@@ -472,75 +495,148 @@ pub async fn publish_note(pool: &Database, note_id: i64) -> Result<String, sqlx:
Ok(token)
}
pub async fn find_published_page(pool: &Database, token: &str) -> Result<Option<PublishedPage>, sqlx::Error> {
pub async fn find_published_page(
pool: &Database,
token: &str,
) -> Result<Option<PublishedPage>, sqlx::Error> {
if pool.kind() == DatabaseKind::Postgres {
return Ok(sqlx::query_as::<_, PostgresPublishedPageRow>(queries::Q021_POSTGRES)
.bind(token).fetch_optional(pool.pool()).await?.map(Into::into));
return Ok(
sqlx::query_as::<_, PostgresPublishedPageRow>(queries::Q021_POSTGRES)
.bind(token)
.fetch_optional(pool.pool())
.await?
.map(Into::into),
);
}
Ok(sqlx::query_as::<_, PublishedPageRow>(queries::get(pool.kind(), queries::Q021))
.bind(token).fetch_optional(pool.pool()).await?.map(Into::into))
Ok(
sqlx::query_as::<_, PublishedPageRow>(queries::get(pool.kind(), queries::Q021))
.bind(token)
.fetch_optional(pool.pool())
.await?
.map(Into::into),
)
}
pub async fn pad_public_task_updates(pool: &Database, pad_id: i64) -> Result<bool, sqlx::Error> {
if pool.kind() == DatabaseKind::Postgres {
return Ok(sqlx::query_scalar::<_, bool>(queries::Q044_POSTGRES)
.bind(pad_id).fetch_optional(pool.pool()).await?.unwrap_or(false));
.bind(pad_id)
.fetch_optional(pool.pool())
.await?
.unwrap_or(false));
}
let value = sqlx::query_scalar::<_, i64>(queries::get(pool.kind(), queries::Q044))
.bind(pad_id).fetch_optional(pool.pool()).await?.unwrap_or(0);
.bind(pad_id)
.fetch_optional(pool.pool())
.await?
.unwrap_or(0);
Ok(value != 0)
}
pub async fn note_public_task_updates(pool: &Database, note_id: i64) -> Result<bool, sqlx::Error> {
if pool.kind() == DatabaseKind::Postgres {
return Ok(sqlx::query_scalar::<_, bool>(queries::Q045_POSTGRES)
.bind(note_id).fetch_optional(pool.pool()).await?.unwrap_or(false));
.bind(note_id)
.fetch_optional(pool.pool())
.await?
.unwrap_or(false));
}
let value = sqlx::query_scalar::<_, i64>(queries::get(pool.kind(), queries::Q045))
.bind(note_id).fetch_optional(pool.pool()).await?.unwrap_or(0);
.bind(note_id)
.fetch_optional(pool.pool())
.await?
.unwrap_or(0);
Ok(value != 0)
}
pub async fn set_pad_public_task_updates(pool: &Database, pad_id: i64, allow: bool) -> Result<(), sqlx::Error> {
pub async fn set_pad_public_task_updates(
pool: &Database,
pad_id: i64,
allow: bool,
) -> Result<(), sqlx::Error> {
publish_pad(pool, pad_id).await?;
let mut query = sqlx::query(queries::get(pool.kind(), queries::Q040));
query = if pool.kind() == DatabaseKind::Postgres { query.bind(allow) } else { query.bind(if allow { 1i64 } else { 0i64 }) };
query = if pool.kind() == DatabaseKind::Postgres {
query.bind(allow)
} else {
query.bind(if allow { 1i64 } else { 0i64 })
};
query.bind(pad_id).execute(pool.pool()).await?;
Ok(())
}
pub async fn set_note_public_task_updates(pool: &Database, note_id: i64, allow: bool) -> Result<(), sqlx::Error> {
pub async fn set_note_public_task_updates(
pool: &Database,
note_id: i64,
allow: bool,
) -> Result<(), sqlx::Error> {
publish_note(pool, note_id).await?;
let mut query = sqlx::query(queries::get(pool.kind(), queries::Q041));
query = if pool.kind() == DatabaseKind::Postgres { query.bind(allow) } else { query.bind(if allow { 1i64 } else { 0i64 }) };
query = if pool.kind() == DatabaseKind::Postgres {
query.bind(allow)
} else {
query.bind(if allow { 1i64 } else { 0i64 })
};
query.bind(note_id).execute(pool.pool()).await?;
Ok(())
}
pub async fn update_public_task(pool: &Database, token: &str, source_line: usize, checked: bool) -> Result<Option<PublishedPage>, sqlx::Error> {
let Some(mut page) = find_published_page(pool, token).await? else { return Ok(None); };
if !page.allow_task_updates || source_line == 0 { return Ok(Some(page)); }
pub async fn update_public_task(
pool: &Database,
token: &str,
source_line: usize,
checked: bool,
) -> Result<Option<PublishedPage>, sqlx::Error> {
let Some(mut page) = find_published_page(pool, token).await? else {
return Ok(None);
};
if !page.allow_task_updates || source_line == 0 {
return Ok(Some(page));
}
let mut lines: Vec<String> = page.content.split('\n').map(str::to_owned).collect();
let Some(line) = lines.get_mut(source_line - 1) else { return Ok(Some(page)); };
let Some(line) = lines.get_mut(source_line - 1) else {
return Ok(Some(page));
};
let bytes = line.as_bytes();
let mut i = 0usize;
while i < bytes.len() && bytes[i].is_ascii_whitespace() { i += 1; }
if i >= bytes.len() || !matches!(bytes[i], b'-' | b'*' | b'+') { return Ok(Some(page)); }
while i < bytes.len() && bytes[i].is_ascii_whitespace() {
i += 1;
while i < bytes.len() && bytes[i].is_ascii_whitespace() { i += 1; }
if i + 2 >= bytes.len() || bytes[i] != b'[' || !matches!(bytes[i + 1], b' ' | b'x' | b'X') || bytes[i + 2] != b']' { return Ok(Some(page)); }
}
if i >= bytes.len() || !matches!(bytes[i], b'-' | b'*' | b'+') {
return Ok(Some(page));
}
i += 1;
while i < bytes.len() && bytes[i].is_ascii_whitespace() {
i += 1;
}
if i + 2 >= bytes.len()
|| bytes[i] != b'['
|| !matches!(bytes[i + 1], b' ' | b'x' | b'X')
|| bytes[i + 2] != b']'
{
return Ok(Some(page));
}
line.replace_range(i + 1..i + 2, if checked { "x" } else { " " });
page.content = lines.join("\n");
if let Some(id) = page.pad_id {
sqlx::query(queries::get(pool.kind(), queries::Q042)).bind(&page.content).bind(id).execute(pool.pool()).await?;
sqlx::query(queries::get(pool.kind(), queries::Q042))
.bind(&page.content)
.bind(id)
.execute(pool.pool())
.await?;
} else if let Some(id) = page.note_id {
sqlx::query(queries::get(pool.kind(), queries::Q043)).bind(&page.content).bind(id).execute(pool.pool()).await?;
sqlx::query(queries::get(pool.kind(), queries::Q043))
.bind(&page.content)
.bind(id)
.execute(pool.pool())
.await?;
}
find_published_page(pool, token).await
}
pub async fn pad_file_token(pool: &Database, pad_id: i64) -> Result<String, sqlx::Error> {
if let Some(token) = sqlx::query_scalar::<_, Option<String>>(queries::get(pool.kind(), queries::Q022))
if let Some(token) =
sqlx::query_scalar::<_, Option<String>>(queries::get(pool.kind(), queries::Q022))
.bind(pad_id)
.fetch_one(pool.pool())
.await?
@@ -562,7 +658,8 @@ pub async fn pad_file_token(pool: &Database, pad_id: i64) -> Result<String, sqlx
}
pub async fn note_file_token(pool: &Database, note_id: i64) -> Result<String, sqlx::Error> {
if let Some(token) = sqlx::query_scalar::<_, Option<String>>(queries::get(pool.kind(), queries::Q024))
if let Some(token) =
sqlx::query_scalar::<_, Option<String>>(queries::get(pool.kind(), queries::Q024))
.bind(note_id)
.fetch_one(pool.pool())
.await?
@@ -583,7 +680,6 @@ pub async fn note_file_token(pool: &Database, note_id: i64) -> Result<String, sq
.await
}
#[derive(Debug, Clone, Copy)]
pub enum FileOwnerKind {
Pad,
@@ -596,25 +692,33 @@ pub struct FileOwner {
pub id: i64,
}
pub async fn find_file_owner(pool: &Database, token: &str) -> Result<Option<FileOwner>, sqlx::Error> {
pub async fn find_file_owner(
pool: &Database,
token: &str,
) -> Result<Option<FileOwner>, sqlx::Error> {
if let Some(id) = sqlx::query_scalar::<_, i64>(queries::get(pool.kind(), queries::Q026))
.bind(token)
.fetch_optional(pool.pool())
.await?
{
return Ok(Some(FileOwner { kind: FileOwnerKind::Pad, id }));
return Ok(Some(FileOwner {
kind: FileOwnerKind::Pad,
id,
}));
}
if let Some(id) = sqlx::query_scalar::<_, i64>(queries::get(pool.kind(), queries::Q027))
.bind(token)
.fetch_optional(pool.pool())
.await?
{
return Ok(Some(FileOwner { kind: FileOwnerKind::Note, id }));
return Ok(Some(FileOwner {
kind: FileOwnerKind::Note,
id,
}));
}
Ok(None)
}
#[derive(Debug, Clone, Serialize, FromRow)]
pub struct NoteFile {
pub id: i64,
@@ -655,14 +759,29 @@ impl From<SqliteNoteFile> for NoteFile {
}
pub async fn delete_note(pool: &Database, note_id: i64) -> Result<(), sqlx::Error> {
sqlx::query(queries::get(pool.kind(), queries::Q031)).bind(note_id).execute(pool.pool()).await?;
sqlx::query(queries::get(pool.kind(), queries::Q031))
.bind(note_id)
.execute(pool.pool())
.await?;
Ok(())
}
pub async fn register_note_file(pool: &Database, note_id: i64, filename: &str, url: &str, mime_type: &str, size_bytes: i64) -> Result<(), sqlx::Error> {
pub async fn register_note_file(
pool: &Database,
note_id: i64,
filename: &str,
url: &str,
mime_type: &str,
size_bytes: i64,
) -> Result<(), sqlx::Error> {
sqlx::query(queries::get(pool.kind(), queries::Q032))
.bind(note_id).bind(filename).bind(url).bind(mime_type).bind(size_bytes)
.execute(pool.pool()).await?;
.bind(note_id)
.bind(filename)
.bind(url)
.bind(mime_type)
.bind(size_bytes)
.execute(pool.pool())
.await?;
Ok(())
}
@@ -670,7 +789,11 @@ pub async fn list_note_files(pool: &Database, note_id: i64) -> Result<Vec<NoteFi
list_files(pool, queries::Q033, note_id).await
}
async fn list_files(pool: &Database, query: &'static str, owner_id: i64) -> Result<Vec<NoteFile>, sqlx::Error> {
async fn list_files(
pool: &Database,
query: &'static str,
owner_id: i64,
) -> Result<Vec<NoteFile>, sqlx::Error> {
if pool.kind() == DatabaseKind::Sqlite {
return Ok(sqlx::query_as::<_, SqliteNoteFile>(query)
.bind(owner_id)
@@ -686,17 +809,41 @@ async fn list_files(pool: &Database, query: &'static str, owner_id: i64) -> Resu
.await
}
pub async fn set_note_file_attached(pool: &Database, file_id: i64, attached: bool) -> Result<(), sqlx::Error> {
let detached_at: Option<String> = if attached { None } else { Some(chrono::Utc::now().to_rfc3339()) };
sqlx::query(queries::get(pool.kind(), queries::Q034)).bind(attached).bind(detached_at).bind(file_id).execute(pool.pool()).await?;
pub async fn set_note_file_attached(
pool: &Database,
file_id: i64,
attached: bool,
) -> Result<(), sqlx::Error> {
let detached_at: Option<String> = if attached {
None
} else {
Some(chrono::Utc::now().to_rfc3339())
};
sqlx::query(queries::get(pool.kind(), queries::Q034))
.bind(attached)
.bind(detached_at)
.bind(file_id)
.execute(pool.pool())
.await?;
Ok(())
}
pub async fn register_pad_file(pool: &Database, pad_id: i64, filename: &str, url: &str, mime_type: &str, size_bytes: i64) -> Result<(), sqlx::Error> {
pub async fn register_pad_file(
pool: &Database,
pad_id: i64,
filename: &str,
url: &str,
mime_type: &str,
size_bytes: i64,
) -> Result<(), sqlx::Error> {
sqlx::query(queries::get(pool.kind(), queries::Q035))
.bind(pad_id).bind(filename).bind(url).bind(mime_type).bind(size_bytes)
.execute(pool.pool()).await?;
.bind(pad_id)
.bind(filename)
.bind(url)
.bind(mime_type)
.bind(size_bytes)
.execute(pool.pool())
.await?;
Ok(())
}
@@ -704,14 +851,30 @@ pub async fn list_pad_files(pool: &Database, pad_id: i64) -> Result<Vec<NoteFile
list_files(pool, queries::Q036, pad_id).await
}
pub async fn set_pad_file_attached(pool: &Database, file_id: i64, attached: bool) -> Result<(), sqlx::Error> {
let detached_at: Option<String> = if attached { None } else { Some(chrono::Utc::now().to_rfc3339()) };
sqlx::query(queries::get(pool.kind(), queries::Q037)).bind(attached).bind(detached_at).bind(file_id).execute(pool.pool()).await?;
pub async fn set_pad_file_attached(
pool: &Database,
file_id: i64,
attached: bool,
) -> Result<(), sqlx::Error> {
let detached_at: Option<String> = if attached {
None
} else {
Some(chrono::Utc::now().to_rfc3339())
};
sqlx::query(queries::get(pool.kind(), queries::Q037))
.bind(attached)
.bind(detached_at)
.bind(file_id)
.execute(pool.pool())
.await?;
Ok(())
}
pub async fn find_note_file(pool: &Database, note_id: i64, file_id: i64) -> Result<Option<NoteFile>, sqlx::Error> {
pub async fn find_note_file(
pool: &Database,
note_id: i64,
file_id: i64,
) -> Result<Option<NoteFile>, sqlx::Error> {
if pool.kind() == DatabaseKind::Sqlite {
return Ok(sqlx::query_as::<_, SqliteNoteFile>(queries::Q038)
.bind(file_id)
@@ -727,7 +890,11 @@ pub async fn find_note_file(pool: &Database, note_id: i64, file_id: i64) -> Resu
.await
}
pub async fn delete_note_file(pool: &Database, note_id: i64, file_id: i64) -> Result<(), sqlx::Error> {
pub async fn delete_note_file(
pool: &Database,
note_id: i64,
file_id: i64,
) -> Result<(), sqlx::Error> {
sqlx::query(queries::get(pool.kind(), queries::Q039))
.bind(file_id)
.bind(note_id)
src/main.rs
+26 -9
View File
@@ -1,6 +1,6 @@
mod api;
mod auth;
mod app;
mod auth;
mod config;
mod database;
mod db;
@@ -45,8 +45,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
asset_version = %config.asset_version,
"configuration loaded"
);
if let Some(path) = config.database_url.strip_prefix("sqlite://").and_then(|v| v.split('?').next()) {
if let Some(parent) = std::path::Path::new(path).parent() { std::fs::create_dir_all(parent)?; }
if let Some(path) = config
.database_url
.strip_prefix("sqlite://")
.and_then(|v| v.split('?').next())
{
if let Some(parent) = std::path::Path::new(path).parent() {
std::fs::create_dir_all(parent)?;
}
}
info!("connecting to database");
let db = Database::connect(&config.database_url, config.database_max_connections).await?;
@@ -55,7 +61,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
info!(database_kind = ?db.kind(), "database migrations completed");
let storage = storage::Storage::from_config(config.storage.clone()).await?;
info!(storage_driver = storage.backend_name(), "file storage ready");
info!(
storage_driver = storage.backend_name(),
"file storage ready"
);
let state = Arc::new(AppState::new(
db,
config.asset_version.clone(),
@@ -124,12 +133,20 @@ async fn run_migrations(db: &Database) -> Result<(), sqlx::migrate::MigrateError
DatabaseKind::Postgres => std::path::Path::new("migrations/postgres"),
DatabaseKind::MySql => std::path::Path::new("migrations/mysql"),
};
sqlx::migrate::Migrator::new(path).await?.run(db.pool()).await
sqlx::migrate::Migrator::new(path)
.await?
.run(db.pool())
.await
}
fn database_kind_label(url: &str) -> &'static str {
if url.starts_with("sqlite:") { "sqlite" }
else if url.starts_with("postgres:") || url.starts_with("postgresql:") { "postgres" }
else if url.starts_with("mysql:") { "mysql" }
else { "unknown" }
if url.starts_with("sqlite:") {
"sqlite"
} else if url.starts_with("postgres:") || url.starts_with("postgresql:") {
"postgres"
} else if url.starts_with("mysql:") {
"mysql"
} else {
"unknown"
}
}
src/queries.rs
+64 -32
View File
@@ -1,6 +1,8 @@
use std::{collections::HashMap, sync::{Mutex, OnceLock}};
use crate::database::DatabaseKind;
use std::{
collections::HashMap,
sync::{Mutex, OnceLock},
};
// Database bootstrap and identity helpers.
pub const SQLITE_FOREIGN_KEYS_ON: &str = "PRAGMA foreign_keys = ON";
@@ -8,55 +10,77 @@ pub const SQLITE_JOURNAL_WAL: &str = "PRAGMA journal_mode = WAL";
pub const SQLITE_BUSY_TIMEOUT: &str = "PRAGMA busy_timeout = 5000";
pub const SQLITE_LAST_INSERT_ID: &str = "SELECT last_insert_rowid()";
pub const MYSQL_LAST_INSERT_ID: &str = "SELECT LAST_INSERT_ID()";
pub const POSTGRES_NOTE_REVISION_LAST_INSERT_ID: &str = "SELECT currval(pg_get_serial_sequence('note_revisions', 'id'))";
pub const POSTGRES_PAD_REVISION_LAST_INSERT_ID: &str = "SELECT currval(pg_get_serial_sequence('revisions', 'id'))";
pub const POSTGRES_NOTE_REVISION_LAST_INSERT_ID: &str =
"SELECT currval(pg_get_serial_sequence('note_revisions', 'id'))";
pub const POSTGRES_PAD_REVISION_LAST_INSERT_ID: &str =
"SELECT currval(pg_get_serial_sequence('revisions', 'id'))";
// Authentication queries.
pub const AUTH_INSERT_USER: &str = "INSERT INTO users (nickname, nickname_key, email, email_key, password_hash, confirmed_at) VALUES (?, ?, ?, ?, ?, ?)";
pub const AUTH_DELETE_USER: &str = "DELETE FROM users WHERE id = ?";
pub const AUTH_SESSION_EXPIRES_AT: &str = "SELECT expires_at FROM user_sessions WHERE token = ?";
pub const AUTH_DELETE_SESSION_BY_TOKEN: &str = "DELETE FROM user_sessions WHERE token = ?";
pub const AUTH_DELETE_CONFIRMATION_TOKENS_BY_USER: &str = "DELETE FROM account_confirmation_tokens WHERE user_id = ?";
pub const AUTH_INSERT_CONFIRMATION_TOKEN: &str = "INSERT INTO account_confirmation_tokens (token, user_id, expires_at) VALUES (?, ?, ?)";
pub const AUTH_FIND_CONFIRMATION_TOKEN: &str = "SELECT user_id, expires_at, used_at FROM account_confirmation_tokens WHERE token = ?";
pub const AUTH_CONFIRM_USER: &str = "UPDATE users SET confirmed_at = ?, updated_at = ? WHERE id = ?";
pub const AUTH_MARK_CONFIRMATION_TOKEN_USED: &str = "UPDATE account_confirmation_tokens SET used_at = ? WHERE token = ?";
pub const AUTH_DELETE_RESET_TOKENS_BY_USER: &str = "DELETE FROM password_reset_tokens WHERE user_id = ?";
pub const AUTH_INSERT_RESET_TOKEN: &str = "INSERT INTO password_reset_tokens (token, user_id, expires_at) VALUES (?, ?, ?)";
pub const AUTH_FIND_RESET_TOKEN: &str = "SELECT user_id, expires_at, used_at FROM password_reset_tokens WHERE token = ?";
pub const AUTH_UPDATE_PASSWORD: &str = "UPDATE users SET password_hash = ?, updated_at = ? WHERE id = ?";
pub const AUTH_MARK_RESET_TOKEN_USED: &str = "UPDATE password_reset_tokens SET used_at = ? WHERE token = ?";
pub const AUTH_DELETE_CONFIRMATION_TOKENS_BY_USER: &str =
"DELETE FROM account_confirmation_tokens WHERE user_id = ?";
pub const AUTH_INSERT_CONFIRMATION_TOKEN: &str =
"INSERT INTO account_confirmation_tokens (token, user_id, expires_at) VALUES (?, ?, ?)";
pub const AUTH_FIND_CONFIRMATION_TOKEN: &str =
"SELECT user_id, expires_at, used_at FROM account_confirmation_tokens WHERE token = ?";
pub const AUTH_CONFIRM_USER: &str =
"UPDATE users SET confirmed_at = ?, updated_at = ? WHERE id = ?";
pub const AUTH_MARK_CONFIRMATION_TOKEN_USED: &str =
"UPDATE account_confirmation_tokens SET used_at = ? WHERE token = ?";
pub const AUTH_DELETE_RESET_TOKENS_BY_USER: &str =
"DELETE FROM password_reset_tokens WHERE user_id = ?";
pub const AUTH_INSERT_RESET_TOKEN: &str =
"INSERT INTO password_reset_tokens (token, user_id, expires_at) VALUES (?, ?, ?)";
pub const AUTH_FIND_RESET_TOKEN: &str =
"SELECT user_id, expires_at, used_at FROM password_reset_tokens WHERE token = ?";
pub const AUTH_UPDATE_PASSWORD: &str =
"UPDATE users SET password_hash = ?, updated_at = ? WHERE id = ?";
pub const AUTH_MARK_RESET_TOKEN_USED: &str =
"UPDATE password_reset_tokens SET used_at = ? WHERE token = ?";
pub const AUTH_DELETE_SESSIONS_BY_USER: &str = "DELETE FROM user_sessions WHERE user_id = ?";
pub const AUTH_USER_BY_SESSION: &str = "SELECT u.id, u.nickname, u.email, u.password_hash, u.confirmed_at FROM user_sessions s JOIN users u ON u.id = s.user_id WHERE s.token = ? AND s.expires_at > ?";
pub const AUTH_INSERT_SESSION: &str = "INSERT INTO user_sessions (token, user_id, expires_at) VALUES (?, ?, ?)";
pub const AUTH_USER_BY_NICKNAME: &str = "SELECT id, nickname, email, password_hash, confirmed_at FROM users WHERE nickname_key = ?";
pub const AUTH_USER_BY_EMAIL: &str = "SELECT id, nickname, email, password_hash, confirmed_at FROM users WHERE email_key = ?";
pub const AUTH_INSERT_SESSION: &str =
"INSERT INTO user_sessions (token, user_id, expires_at) VALUES (?, ?, ?)";
pub const AUTH_USER_BY_NICKNAME: &str =
"SELECT id, nickname, email, password_hash, confirmed_at FROM users WHERE nickname_key = ?";
pub const AUTH_USER_BY_EMAIL: &str =
"SELECT id, nickname, email, password_hash, confirmed_at FROM users WHERE email_key = ?";
pub const USER_ATTACH_WORKSPACE: &str = "INSERT INTO user_workspaces (user_id, workspace_id) SELECT ?, id FROM workspaces WHERE slug = ?";
pub const USER_ATTACH_PAD: &str = "INSERT INTO user_pads (user_id, pad_id) SELECT ?, id FROM pads WHERE slug = ?";
pub const USER_ATTACH_PAD: &str =
"INSERT INTO user_pads (user_id, pad_id) SELECT ?, id FROM pads WHERE slug = ?";
pub const USER_LIST_WORKSPACES: &str = "SELECT w.slug, w.title, CASE WHEN w.password_hash IS NULL THEN 0 ELSE 1 END AS protected, w.updated_at, CASE WHEN w.is_private THEN 1 ELSE 0 END AS private, 1 AS owned, 'rw' AS permission, '' AS shared_by FROM user_workspaces uw JOIN workspaces w ON w.id = uw.workspace_id WHERE uw.user_id = ? UNION SELECT w.slug, w.title, CASE WHEN w.password_hash IS NULL THEN 0 ELSE 1 END, w.updated_at, CASE WHEN w.is_private THEN 1 ELSE 0 END, 0, rp.permission, COALESCE((SELECT u.nickname FROM user_workspaces owner_uw JOIN users u ON u.id = owner_uw.user_id WHERE owner_uw.workspace_id = w.id LIMIT 1), 'Unknown user') AS shared_by FROM resource_permissions rp JOIN workspaces w ON w.slug = rp.resource_slug WHERE rp.resource_kind = 'workspace' AND rp.user_id = ? ORDER BY updated_at DESC";
pub const USER_LIST_PADS: &str = "SELECT p.slug, p.title, CASE WHEN p.password_hash IS NULL THEN 0 ELSE 1 END AS protected, p.updated_at, CASE WHEN p.is_private THEN 1 ELSE 0 END AS private, 1 AS owned, 'rw' AS permission, '' AS shared_by FROM user_pads up JOIN pads p ON p.id = up.pad_id WHERE up.user_id = ? UNION SELECT p.slug, p.title, CASE WHEN p.password_hash IS NULL THEN 0 ELSE 1 END, p.updated_at, CASE WHEN p.is_private THEN 1 ELSE 0 END, 0, rp.permission, COALESCE((SELECT u.nickname FROM user_pads owner_up JOIN users u ON u.id = owner_up.user_id WHERE owner_up.pad_id = p.id LIMIT 1), 'Unknown user') AS shared_by FROM resource_permissions rp JOIN pads p ON p.slug = rp.resource_slug WHERE rp.resource_kind = 'pad' AND rp.user_id = ? ORDER BY updated_at DESC";
pub const USER_OWNS_WORKSPACE: &str = "SELECT COUNT(*) FROM user_workspaces uw JOIN workspaces w ON w.id = uw.workspace_id WHERE uw.user_id = ? AND w.slug = ?";
pub const USER_OWNS_PAD: &str = "SELECT COUNT(*) FROM user_pads up JOIN pads p ON p.id = up.pad_id WHERE up.user_id = ? AND p.slug = ?";
pub const USER_DELETE_WORKSPACE: &str = "DELETE FROM workspaces WHERE slug = ?";
pub const USER_DELETE_PAD: &str = "DELETE FROM pads WHERE slug = ?";
pub const USER_SET_WORKSPACE_PASSWORD: &str = "UPDATE workspaces SET password_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE slug = ?";
pub const USER_SET_PAD_PASSWORD: &str = "UPDATE pads SET password_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE slug = ?";
pub const USER_SET_WORKSPACE_PASSWORD: &str =
"UPDATE workspaces SET password_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE slug = ?";
pub const USER_SET_PAD_PASSWORD: &str =
"UPDATE pads SET password_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE slug = ?";
pub const Q001: &str = "SELECT id, slug, title, password_hash, created_at, updated_at, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS BIGINT) AS is_private FROM workspaces WHERE slug = ?";
pub const Q002: &str = "INSERT INTO workspaces (slug, title, password_hash) VALUES (?, ?, ?)";
pub const Q003: &str = "SELECT id, workspace_id, slug, title, content, created_at, updated_at, owner_map, protected, created_by FROM notes WHERE workspace_id = ? ORDER BY updated_at DESC, id DESC";
pub const Q004: &str = "SELECT id, workspace_id, slug, title, content, created_at, updated_at, owner_map, protected, created_by FROM notes WHERE workspace_id = ? AND slug = ?";
pub const Q005: &str = "INSERT INTO notes (workspace_id, slug, title, protected, created_by) VALUES (?, ?, ?, ?, ?)";
pub const Q006: &str = "UPDATE notes SET content = ?, owner_map = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?";
pub const Q005: &str =
"INSERT INTO notes (workspace_id, slug, title, protected, created_by) VALUES (?, ?, ?, ?, ?)";
pub const Q006: &str =
"UPDATE notes SET content = ?, owner_map = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?";
pub const Q007: &str = "UPDATE workspaces SET updated_at = CURRENT_TIMESTAMP WHERE id = ?";
pub const Q008: &str = "INSERT INTO note_revisions (note_id, content, author, owner_map) VALUES (?, ?, ?, ?)";
pub const Q008: &str =
"INSERT INTO note_revisions (note_id, content, author, owner_map) VALUES (?, ?, ?, ?)";
pub const Q009: &str = "SELECT updated_at FROM notes WHERE id = ?";
pub const Q010: &str = "SELECT id, content, created_at, author, owner_map FROM note_revisions WHERE note_id = ? ORDER BY id DESC LIMIT 100";
pub const Q011: &str = "SELECT id, slug, title, content, password_hash, created_at, updated_at, owner_map, CAST(CASE WHEN is_private THEN 1 ELSE 0 END AS BIGINT) AS is_private FROM pads WHERE slug = ?";
pub const Q012: &str = "INSERT INTO pads (slug, title, password_hash) VALUES (?, ?, ?)";
pub const Q013: &str = "UPDATE pads SET content = ?, owner_map = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?";
pub const Q014: &str = "INSERT INTO revisions (pad_id, content, author, owner_map) VALUES (?, ?, ?, ?)";
pub const Q013: &str =
"UPDATE pads SET content = ?, owner_map = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?";
pub const Q014: &str =
"INSERT INTO revisions (pad_id, content, author, owner_map) VALUES (?, ?, ?, ?)";
pub const Q015: &str = "SELECT updated_at FROM pads WHERE id = ?";
pub const Q016: &str = "SELECT id, content, created_at, author, owner_map FROM revisions WHERE pad_id = ? ORDER BY id DESC LIMIT 100";
pub const Q017: &str = "SELECT token FROM published_pages WHERE pad_id = ?";
@@ -74,10 +98,12 @@ pub const Q028: &str = "SELECT content FROM note_revisions WHERE id = ? AND note
pub const Q029: &str = "SELECT content FROM revisions WHERE id = ? AND pad_id = ?";
pub const Q030: &str = "SELECT owner_map FROM revisions WHERE id = ? AND pad_id = ?";
pub const Q031: &str = "DELETE FROM notes WHERE id = ?";
pub const Q032: &str = "INSERT INTO note_files (note_id, filename, url, mime_type, size_bytes) VALUES (?, ?, ?, ?, ?)";
pub const Q032: &str =
"INSERT INTO note_files (note_id, filename, url, mime_type, size_bytes) VALUES (?, ?, ?, ?, ?)";
pub const Q033: &str = "SELECT id, filename, url, mime_type, size_bytes, created_at, is_attached, detached_at FROM note_files WHERE note_id = ? ORDER BY id DESC";
pub const Q034: &str = "UPDATE note_files SET is_attached = ?, detached_at = ? WHERE id = ?";
pub const Q035: &str = "INSERT INTO pad_files (pad_id, filename, url, mime_type, size_bytes) VALUES (?, ?, ?, ?, ?)";
pub const Q035: &str =
"INSERT INTO pad_files (pad_id, filename, url, mime_type, size_bytes) VALUES (?, ?, ?, ?, ?)";
pub const Q036: &str = "SELECT id, filename, url, mime_type, size_bytes, created_at, is_attached, detached_at FROM pad_files WHERE pad_id = ? ORDER BY id DESC";
pub const Q037: &str = "UPDATE pad_files SET is_attached = ?, detached_at = ? WHERE id = ?";
pub const Q038: &str = "SELECT id, filename, url, mime_type, size_bytes, created_at, is_attached, detached_at FROM note_files WHERE id = ? AND note_id = ?";
@@ -86,10 +112,14 @@ pub const Q039: &str = "DELETE FROM note_files WHERE id = ? AND note_id = ?";
static POSTGRES_QUERIES: OnceLock<Mutex<HashMap<&'static str, &'static str>>> = OnceLock::new();
pub fn get(kind: DatabaseKind, query: &'static str) -> &'static str {
if kind != DatabaseKind::Postgres { return query; }
if kind != DatabaseKind::Postgres {
return query;
}
let cache = POSTGRES_QUERIES.get_or_init(|| Mutex::new(HashMap::new()));
let mut cache = cache.lock().expect("query cache lock poisoned");
if let Some(value) = cache.get(query) { return value; }
if let Some(value) = cache.get(query) {
return value;
}
let cache_key = query;
let query = query.replace("CURRENT_TIMESTAMP", "(CURRENT_TIMESTAMP::text)");
let mut index = 0;
@@ -113,8 +143,10 @@ pub const Q041: &str = "UPDATE published_pages SET allow_task_updates = ? WHERE
pub const Q042: &str = "UPDATE pads SET content = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?";
pub const Q043: &str = "UPDATE notes SET content = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?";
pub const Q044: &str = "SELECT CASE WHEN allow_task_updates THEN 1 ELSE 0 END FROM published_pages WHERE pad_id = ?";
pub const Q045: &str = "SELECT CASE WHEN allow_task_updates THEN 1 ELSE 0 END FROM published_pages WHERE note_id = ?";
pub const Q044: &str =
"SELECT CASE WHEN allow_task_updates THEN 1 ELSE 0 END FROM published_pages WHERE pad_id = ?";
pub const Q045: &str =
"SELECT CASE WHEN allow_task_updates THEN 1 ELSE 0 END FROM published_pages WHERE note_id = ?";
pub const Q021_POSTGRES: &str = "SELECT pp.token, pp.pad_id, pp.note_id, pp.allow_task_updates, COALESCE(p.title, n.title) AS title, COALESCE(p.content, n.content) AS content, COALESCE(p.updated_at, n.updated_at) AS updated_at FROM published_pages pp LEFT JOIN pads p ON p.id = pp.pad_id LEFT JOIN notes n ON n.id = pp.note_id WHERE pp.token = $1";
pub const Q044_POSTGRES: &str = "SELECT allow_task_updates FROM published_pages WHERE pad_id = $1";
src/state.rs
+101 -19
View File
@@ -1,13 +1,24 @@
use std::{collections::HashMap, sync::{Arc, atomic::{AtomicU64, Ordering}}};
use crate::database::Database;
use tokio::sync::{broadcast, RwLock};
use serde::Serialize;
use std::{
collections::HashMap,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
};
use tokio::sync::{RwLock, broadcast};
const CHANNEL_CAPACITY: usize = 256;
#[derive(Debug, Clone)]
pub struct SmtpConfig {
pub host: String, pub port: u16, pub username: String, pub password: String, pub from: String, pub public_url: String
pub host: String,
pub port: u16,
pub username: String,
pub password: String,
pub from: String,
pub public_url: String,
}
#[derive(Debug, Clone)]
@@ -52,31 +63,98 @@ pub struct AppState {
}
impl AppState {
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, share_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, share_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,
share_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,
share_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(); }
let mut channels = self.channels.write().await;
channels.entry(key).or_insert_with(|| broadcast::channel(CHANNEL_CAPACITY).0).clone()
if let Some(sender) = self.channels.read().await.get(&key) {
return sender.clone();
}
pub fn note_room_key(workspace_slug: &str, note_slug: &str) -> String { format!("workspace:{workspace_slug}/{note_slug}") }
pub fn pad_room_key(slug: &str) -> String { format!("pad:{slug}") }
pub async fn note_channel(&self, workspace_slug: &str, note_slug: &str) -> broadcast::Sender<RoomEvent> { self.channel_for_key(Self::note_room_key(workspace_slug, note_slug)).await }
pub async fn pad_channel(&self, slug: &str) -> broadcast::Sender<RoomEvent> { self.channel_for_key(Self::pad_room_key(slug)).await }
pub async fn join_room(&self, key: &str, nickname: String, color: Option<String>) -> (u64, Vec<PresenceUser>) {
let mut channels = self.channels.write().await;
channels
.entry(key)
.or_insert_with(|| broadcast::channel(CHANNEL_CAPACITY).0)
.clone()
}
pub fn note_room_key(workspace_slug: &str, note_slug: &str) -> String {
format!("workspace:{workspace_slug}/{note_slug}")
}
pub fn pad_room_key(slug: &str) -> String {
format!("pad:{slug}")
}
pub async fn note_channel(
&self,
workspace_slug: &str,
note_slug: &str,
) -> broadcast::Sender<RoomEvent> {
self.channel_for_key(Self::note_room_key(workspace_slug, note_slug))
.await
}
pub async fn pad_channel(&self, slug: &str) -> broadcast::Sender<RoomEvent> {
self.channel_for_key(Self::pad_room_key(slug)).await
}
pub async fn join_room(
&self,
key: &str,
nickname: String,
color: Option<String>,
) -> (u64, Vec<PresenceUser>) {
let id = self.next_connection_id.fetch_add(1, Ordering::Relaxed);
let mut presence = self.presence.write().await;
let room = presence.entry(key.to_owned()).or_default();
room.insert(id, PresenceUser { name: nickname, color });
room.insert(
id,
PresenceUser {
name: nickname,
color,
},
);
(id, sorted_users(room))
}
pub async fn update_room_color(&self, key: &str, id: u64, color: Option<String>) -> Vec<PresenceUser> {
pub async fn update_room_color(
&self,
key: &str,
id: u64,
color: Option<String>,
) -> Vec<PresenceUser> {
let mut presence = self.presence.write().await;
if let Some(room) = presence.get_mut(key) {
if let Some(user) = room.get_mut(&id) { user.color = color; }
if let Some(user) = room.get_mut(&id) {
user.color = color;
}
sorted_users(room)
} else { Vec::new() }
} else {
Vec::new()
}
}
pub async fn leave_room(&self, key: &str, id: u64) -> Vec<PresenceUser> {
let mut presence = self.presence.write().await;
@@ -84,9 +162,13 @@ impl AppState {
room.remove(&id);
let users = sorted_users(room);
let empty = room.is_empty();
if empty { presence.remove(key); }
if empty {
presence.remove(key);
}
users
} else { Vec::new() }
} else {
Vec::new()
}
}
}
src/storage.rs
+85 -19
View File
@@ -2,12 +2,14 @@ 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 aws_sdk_s3::{Client, config::Builder as S3ConfigBuilder, primitives::ByteStream};
use bytes::Bytes;
#[derive(Debug, Clone)]
pub enum StorageConfig {
Local { root: PathBuf },
Local {
root: PathBuf,
},
S3 {
endpoint: Option<String>,
region: String,
@@ -40,8 +42,16 @@ impl Storage {
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");
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)
@@ -53,42 +63,70 @@ impl Storage {
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) })
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" }
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 {
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)
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> {
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?; }
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()
client
.put_object()
.bucket(bucket.as_ref())
.key(key)
.content_type(content_type)
.cache_control(cache_control)
.body(ByteStream::from(bytes))
.send().await
.send()
.await
.map_err(|error| StorageError::Backend(error.to_string()))?;
Ok(())
}
@@ -99,9 +137,17 @@ impl Storage {
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
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
let bytes = output
.body
.collect()
.await
.map_err(|error| StorageError::Backend(error.to_string()))?
.into_bytes();
Ok(bytes)
@@ -109,11 +155,19 @@ impl Storage {
}
}
pub async fn get_local_with_legacy(&self, key: &str, legacy_key: &str) -> Result<Bytes, StorageError> {
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) };
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,
@@ -128,7 +182,12 @@ impl Storage {
Err(error) => Err(error.into()),
},
Self::S3 { client, bucket } => {
client.delete_object().bucket(bucket.as_ref()).key(key).send().await
client
.delete_object()
.bucket(bucket.as_ref())
.key(key)
.send()
.await
.map_err(|error| StorageError::Backend(error.to_string()))?;
Ok(())
}
@@ -144,11 +203,18 @@ pub enum StorageError {
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) }
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) } }
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}")
src/websocket.rs
+339 -62
View File
@@ -1,62 +1,188 @@
use axum::{extract::{ws::{Message, WebSocket}, Path, State, WebSocketUpgrade}, response::Response};
use crate::{
auth, db,
state::{AppState, NoteUpdate, PresenceUser, RoomEvent, SharedState},
};
use axum::{
extract::{
Path, State, WebSocketUpgrade,
ws::{Message, WebSocket},
},
response::Response,
};
use futures_util::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use tracing::{debug, info, warn};
use crate::{auth, db, state::{AppState, NoteUpdate, PresenceUser, RoomEvent, SharedState}};
use std::time::{Duration, Instant};
use tracing::{debug, info, warn};
#[derive(Debug, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum ClientMessage {
Authenticate { password: Option<String>, access_token: Option<String>, nickname: Option<String>, session_token: Option<String>, color: Option<String> },
Update { content: String, owner_map: Option<String> },
Ping { nonce: u64 },
Chat { text: String },
SetColor { color: Option<String> },
Authenticate {
password: Option<String>,
access_token: Option<String>,
nickname: Option<String>,
session_token: Option<String>,
color: Option<String>,
},
Update {
content: String,
owner_map: Option<String>,
},
Ping {
nonce: u64,
},
Chat {
text: String,
},
SetColor {
color: Option<String>,
},
}
#[derive(Debug, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum ServerMessage {
Authenticated { workspace_title: String, note_title: String, content: String, owner_map: String },
Document { content: String, revision_id: i64, updated_at: String, author: Option<String>, owner_map: String },
Presence { users: Vec<PresenceUser> },
Chat { sender: String, text: String },
Pong { nonce: u64 },
Error { message: String },
Authenticated {
workspace_title: String,
note_title: String,
content: String,
owner_map: String,
},
Document {
content: String,
revision_id: i64,
updated_at: String,
author: Option<String>,
owner_map: String,
},
Presence {
users: Vec<PresenceUser>,
},
Chat {
sender: String,
text: String,
},
Pong {
nonce: u64,
},
Error {
message: String,
},
}
pub async fn upgrade(ws: WebSocketUpgrade, Path((workspace_slug, note_slug)): Path<(String, String)>, State(state): State<SharedState>) -> Response {
pub async fn upgrade(
ws: WebSocketUpgrade,
Path((workspace_slug, note_slug)): Path<(String, String)>,
State(state): State<SharedState>,
) -> Response {
ws.on_upgrade(move |socket| handle_socket(socket, state, workspace_slug, note_slug))
}
async fn handle_socket(mut socket: WebSocket, state: SharedState, workspace_slug: String, note_slug: String) {
async fn handle_socket(
mut socket: WebSocket,
state: SharedState,
workspace_slug: String,
note_slug: String,
) {
info!(%workspace_slug, %note_slug, "note websocket connected");
let Some(workspace) = db::find_workspace(&state.db, &workspace_slug).await.ok().flatten() else { warn!(%workspace_slug, %note_slug, "note websocket rejected: workspace not found"); let _=send_error(&mut socket,"Workspace not found").await; return; };
let Some(note) = db::find_note(&state.db, workspace.id, &note_slug).await.ok().flatten() else { warn!(%workspace_slug, %note_slug, "note websocket rejected: note not found"); let _=send_error(&mut socket,"Note not found").await; return; };
let Some(workspace) = db::find_workspace(&state.db, &workspace_slug)
.await
.ok()
.flatten()
else {
warn!(%workspace_slug, %note_slug, "note websocket rejected: workspace not found");
let _ = send_error(&mut socket, "Workspace not found").await;
return;
};
let Some(note) = db::find_note(&state.db, workspace.id, &note_slug)
.await
.ok()
.flatten()
else {
warn!(%workspace_slug, %note_slug, "note websocket rejected: note not found");
let _ = send_error(&mut socket, "Note not found").await;
return;
};
let (password, access_token, nickname, session_token, color) = match socket.recv().await {
Some(Ok(Message::Text(text))) => match serde_json::from_str::<ClientMessage>(&text) {
Ok(ClientMessage::Authenticate { password, access_token, nickname, session_token, color }) => (password, access_token, clean_nickname(nickname), session_token, clean_color(color)),
_ => { let _=send_error(&mut socket,"Wymagane uwierzytelnienie").await; return; }
}, _ => return
Ok(ClientMessage::Authenticate {
password,
access_token,
nickname,
session_token,
color,
}) => (
password,
access_token,
clean_nickname(nickname),
session_token,
clean_color(color),
),
_ => {
let _ = send_error(&mut socket, "Wymagane uwierzytelnienie").await;
return;
}
},
_ => return,
};
let nickname = match auth::authorize_nickname(&state, nickname, session_token.clone()).await { Ok(value) => value, Err(message) => { let _=send_error(&mut socket,&message).await; return; } };
let permission = auth::resource_permission(&state, "workspace", &workspace_slug, session_token.as_deref().or(access_token.as_deref())).await.ok().flatten();
let nickname = match auth::authorize_nickname(&state, nickname, session_token.clone()).await {
Ok(value) => value,
Err(message) => {
let _ = send_error(&mut socket, &message).await;
return;
}
};
let permission = auth::resource_permission(
&state,
"workspace",
&workspace_slug,
session_token.as_deref().or(access_token.as_deref()),
)
.await
.ok()
.flatten();
let password_ok = db::verify_workspace_password(&workspace, password.as_deref());
if workspace.is_private != 0 && permission.is_none() { let _=send_error(&mut socket,"This workspace is private").await; return; }
if workspace.password_hash.is_some() && !password_ok && permission.is_none() { warn!(workspace_id = workspace.id, note_id = note.id, "note websocket rejected: invalid workspace password"); let _=send_error(&mut socket,"Invalid password").await; return; }
if workspace.is_private != 0 && permission.is_none() {
let _ = send_error(&mut socket, "This workspace is private").await;
return;
}
if workspace.password_hash.is_some() && !password_ok && permission.is_none() {
warn!(
workspace_id = workspace.id,
note_id = note.id,
"note websocket rejected: invalid workspace password"
);
let _ = send_error(&mut socket, "Invalid password").await;
return;
}
let write_allowed = password_ok || permission.as_deref() != Some("ro");
info!(workspace_id = workspace.id, note_id = note.id, nickname = ?nickname, "note websocket authenticated");
if send(&mut socket,&ServerMessage::Authenticated { workspace_title:workspace.title.clone(), note_title:note.title.clone(), content:note.content.clone(), owner_map:note.owner_map.clone() }).await.is_err(){return;}
if send(
&mut socket,
&ServerMessage::Authenticated {
workspace_title: workspace.title.clone(),
note_title: note.title.clone(),
content: note.content.clone(),
owner_map: note.owner_map.clone(),
},
)
.await
.is_err()
{
return;
}
let room_key = AppState::note_room_key(&workspace_slug, &note_slug);
let channel = state.note_channel(&workspace_slug, &note_slug).await;
let mut updates = channel.subscribe();
let display_name = nickname.clone().unwrap_or_else(|| "Guest".into());
let (connection_id, users) = state.join_room(&room_key, display_name.clone(), color).await;
let (connection_id, users) = state
.join_room(&room_key, display_name.clone(), color)
.await;
let _ = channel.send(RoomEvent::Presence(users));
let mut last_chat = Instant::now() - Duration::from_secs(1);
let (mut sender, mut receiver) = socket.split();
loop { tokio::select! {
loop {
tokio::select! {
incoming=receiver.next()=>match incoming {
Some(Ok(Message::Text(text)))=>match serde_json::from_str::<ClientMessage>(&text) {
Ok(ClientMessage::Update{content,owner_map})=>{
@@ -85,72 +211,211 @@ async fn handle_socket(mut socket: WebSocket, state: SharedState, workspace_slug
Err(tokio::sync::broadcast::error::RecvError::Lagged(_))=>if let Ok(Some(current))=db::find_note(&state.db,workspace.id,&note_slug).await { if send_split(&mut sender,&ServerMessage::Document{content:current.content,revision_id:0,updated_at:current.updated_at,author:None,owner_map:current.owner_map}).await.is_err(){break;} },
Err(tokio::sync::broadcast::error::RecvError::Closed)=>break,
}
}}
}
}
let users = state.leave_room(&room_key, connection_id).await;
let _ = channel.send(RoomEvent::Presence(users));
info!(workspace_id = workspace.id, note_id = note.id, "note websocket disconnected");
info!(
workspace_id = workspace.id,
note_id = note.id,
"note websocket disconnected"
);
}
fn clean_nickname(value: Option<String>) -> Option<String> {
value.map(|v|v.trim().chars().take(40).collect::<String>()).filter(|v|!v.is_empty())
value
.map(|v| v.trim().chars().take(40).collect::<String>())
.filter(|v| !v.is_empty())
}
fn clean_color(value: Option<String>) -> Option<String> {
value.map(|v| v.trim().to_ascii_lowercase()).filter(|v| v.len() == 7 && v.starts_with('#') && v[1..].chars().all(|c| c.is_ascii_hexdigit()))
value.map(|v| v.trim().to_ascii_lowercase()).filter(|v| {
v.len() == 7 && v.starts_with('#') && v[1..].chars().all(|c| c.is_ascii_hexdigit())
})
}
fn clean_chat(value: String) -> String {
value.chars().map(|c| if matches!(c, '\r' | '\n' | '\0') { ' ' } else { c }).collect::<String>().trim().chars().take(1000).collect()
value
.chars()
.map(|c| {
if matches!(c, '\r' | '\n' | '\0') {
' '
} else {
c
}
})
.collect::<String>()
.trim()
.chars()
.take(1000)
.collect()
}
async fn send_error(socket: &mut WebSocket, message: &str) -> Result<(), axum::Error> {
send(socket,&ServerMessage::Error {
message:message.into()
}
).await
send(
socket,
&ServerMessage::Error {
message: message.into(),
},
)
.await
}
async fn send(socket: &mut WebSocket, message: &ServerMessage) -> Result<(), axum::Error> {
socket.send(Message::Text(serde_json::to_string(message).unwrap().into())).await
socket
.send(Message::Text(
serde_json::to_string(message).unwrap().into(),
))
.await
}
async fn send_split(sender:&mut futures_util::stream::SplitSink<WebSocket,Message>,message:&ServerMessage)->Result<(),axum::Error> {
sender.send(Message::Text(serde_json::to_string(message).unwrap().into())).await
async fn send_split(
sender: &mut futures_util::stream::SplitSink<WebSocket, Message>,
message: &ServerMessage,
) -> Result<(), axum::Error> {
sender
.send(Message::Text(
serde_json::to_string(message).unwrap().into(),
))
.await
}
#[derive(Debug, Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum PadServerMessage {
Authenticated { title: String, content: String, owner_map: String },
Document { content: String, revision_id: i64, updated_at: String, author: Option<String>, owner_map: String },
Presence { users: Vec<PresenceUser> },
Chat { sender: String, text: String },
Pong { nonce: u64 },
Error { message: String },
Authenticated {
title: String,
content: String,
owner_map: String,
},
Document {
content: String,
revision_id: i64,
updated_at: String,
author: Option<String>,
owner_map: String,
},
Presence {
users: Vec<PresenceUser>,
},
Chat {
sender: String,
text: String,
},
Pong {
nonce: u64,
},
Error {
message: String,
},
}
pub async fn upgrade_pad(ws:WebSocketUpgrade,Path(slug):Path<String>,State(state):State<SharedState>)->Response{
pub async fn upgrade_pad(
ws: WebSocketUpgrade,
Path(slug): Path<String>,
State(state): State<SharedState>,
) -> Response {
ws.on_upgrade(move |socket| handle_pad_socket(socket, state, slug))
}
async fn handle_pad_socket(mut socket: WebSocket, state: SharedState, slug: String) {
info!(%slug, "pad websocket connected");
let Some(pad)=db::find_pad(&state.db,&slug).await.ok().flatten() else {warn!(%slug, "pad websocket rejected: pad not found");let _=send_pad(&mut socket,&PadServerMessage::Error{message:"Note not found".into()}).await;return;};
let Some(pad) = db::find_pad(&state.db, &slug).await.ok().flatten() else {
warn!(%slug, "pad websocket rejected: pad not found");
let _ = send_pad(
&mut socket,
&PadServerMessage::Error {
message: "Note not found".into(),
},
)
.await;
return;
};
let (password, access_token, nickname, session_token, color) = match socket.recv().await {
Some(Ok(Message::Text(text))) => match serde_json::from_str::<ClientMessage>(&text) {
Ok(ClientMessage::Authenticate{password,access_token,nickname,session_token,color})=>(password,access_token,clean_nickname(nickname),session_token,clean_color(color)),
_=>{let _=send_pad(&mut socket,&PadServerMessage::Error{message:"Wymagane uwierzytelnienie".into()}).await;return;}
},_=>return
Ok(ClientMessage::Authenticate {
password,
access_token,
nickname,
session_token,
color,
}) => (
password,
access_token,
clean_nickname(nickname),
session_token,
clean_color(color),
),
_ => {
let _ = send_pad(
&mut socket,
&PadServerMessage::Error {
message: "Wymagane uwierzytelnienie".into(),
},
)
.await;
return;
}
},
_ => return,
};
let nickname=match auth::authorize_nickname(&state,nickname,session_token.clone()).await{Ok(value)=>value,Err(message)=>{let _=send_pad(&mut socket,&PadServerMessage::Error{message}).await;return;}};
let permission=auth::resource_permission(&state,"pad",&slug,session_token.as_deref().or(access_token.as_deref())).await.ok().flatten();
let nickname = match auth::authorize_nickname(&state, nickname, session_token.clone()).await {
Ok(value) => value,
Err(message) => {
let _ = send_pad(&mut socket, &PadServerMessage::Error { message }).await;
return;
}
};
let permission = auth::resource_permission(
&state,
"pad",
&slug,
session_token.as_deref().or(access_token.as_deref()),
)
.await
.ok()
.flatten();
let password_ok = db::verify_pad_password(&pad, password.as_deref());
if pad.is_private != 0 && permission.is_none(){let _=send_pad(&mut socket,&PadServerMessage::Error{message:"This note is private".into()}).await;return;}
if pad.password_hash.is_some() && !password_ok && permission.is_none(){warn!(pad_id = pad.id, "pad websocket rejected: invalid password");let _=send_pad(&mut socket,&PadServerMessage::Error{message:"Invalid password".into()}).await;return;}
if pad.is_private != 0 && permission.is_none() {
let _ = send_pad(
&mut socket,
&PadServerMessage::Error {
message: "This note is private".into(),
},
)
.await;
return;
}
if pad.password_hash.is_some() && !password_ok && permission.is_none() {
warn!(pad_id = pad.id, "pad websocket rejected: invalid password");
let _ = send_pad(
&mut socket,
&PadServerMessage::Error {
message: "Invalid password".into(),
},
)
.await;
return;
}
let write_allowed = password_ok || permission.as_deref() != Some("ro");
info!(pad_id = pad.id, nickname = ?nickname, "pad websocket authenticated");
if send_pad(&mut socket,&PadServerMessage::Authenticated{title:pad.title.clone(),content:pad.content.clone(),owner_map:pad.owner_map.clone()}).await.is_err(){return;}
if send_pad(
&mut socket,
&PadServerMessage::Authenticated {
title: pad.title.clone(),
content: pad.content.clone(),
owner_map: pad.owner_map.clone(),
},
)
.await
.is_err()
{
return;
}
let room_key = AppState::pad_room_key(&slug);
let channel = state.pad_channel(&slug).await;
let mut updates = channel.subscribe();
let display_name = nickname.clone().unwrap_or_else(|| "Guest".into());
let (connection_id, users) = state.join_room(&room_key, display_name.clone(), color).await;
let (connection_id, users) = state
.join_room(&room_key, display_name.clone(), color)
.await;
let _ = channel.send(RoomEvent::Presence(users));
let mut last_chat = Instant::now() - Duration::from_secs(1);
let (mut sender, mut receiver) = socket.split();
loop{tokio::select!{
loop {
tokio::select! {
incoming=receiver.next()=>match incoming{
Some(Ok(Message::Text(text)))=>match serde_json::from_str::<ClientMessage>(&text){
Ok(ClientMessage::Update{content,owner_map})=>{if !write_allowed{let _=send_pad_split(&mut sender,&PadServerMessage::Error{message:"Read-only access".into()}).await;continue;}
@@ -180,14 +445,26 @@ async fn handle_pad_socket(mut socket:WebSocket,state:SharedState,slug:String){
Err(tokio::sync::broadcast::error::RecvError::Lagged(_))=>if let Ok(Some(current))=db::find_pad(&state.db,&slug).await { if send_pad_split(&mut sender,&PadServerMessage::Document{content:current.content,revision_id:0,updated_at:current.updated_at,author:None,owner_map:current.owner_map}).await.is_err(){break;} },
Err(tokio::sync::broadcast::error::RecvError::Closed)=>break,
}
}}
}
}
let users = state.leave_room(&room_key, connection_id).await;
let _ = channel.send(RoomEvent::Presence(users));
info!(pad_id = pad.id, "pad websocket disconnected");
}
async fn send_pad(socket: &mut WebSocket, message: &PadServerMessage) -> Result<(), axum::Error> {
socket.send(Message::Text(serde_json::to_string(message).unwrap().into())).await
socket
.send(Message::Text(
serde_json::to_string(message).unwrap().into(),
))
.await
}
async fn send_pad_split(sender:&mut futures_util::stream::SplitSink<WebSocket,Message>,message:&PadServerMessage)->Result<(),axum::Error> {
sender.send(Message::Text(serde_json::to_string(message).unwrap().into())).await
async fn send_pad_split(
sender: &mut futures_util::stream::SplitSink<WebSocket, Message>,
message: &PadServerMessage,
) -> Result<(), axum::Error> {
sender
.send(Message::Text(
serde_json::to_string(message).unwrap().into(),
))
.await
}