nils/papercrate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rls

Browse Source
This commit is contained in:
nils
2025-10-31 19:13:11 +01:00
parent 2d8a4432cc
commit 175387f0d3
19 changed files with 463 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/db.rs
+30 -1
View File
@@ -1,12 +1,40 @@
use std::time::Duration;
use diesel::pg::PgConnection;
use diesel::r2d2::{ConnectionManager, Pool};
use diesel::r2d2::{ConnectionManager, CustomizeConnection, Pool};
use diesel::RunQueryDsl;
pub type PgPool = Pool<ConnectionManager<PgConnection>>;
pub const DEFAULT_MAX_POOL_SIZE: u32 = 2;
#[derive(Debug)]
struct SchemaCustomizer;
impl CustomizeConnection<PgConnection, diesel::r2d2::Error> for SchemaCustomizer {
fn on_acquire(&self, conn: &mut PgConnection) -> Result<(), diesel::r2d2::Error> {
diesel::sql_query(
"SELECT set_config('search_path', (
SELECT string_agg(schema_name, ', ')
FROM (
SELECT 'tenant' AS schema_name WHERE EXISTS (
SELECT 1 FROM pg_namespace WHERE nspname = 'tenant'
)
UNION ALL
SELECT 'shared' AS schema_name WHERE EXISTS (
SELECT 1 FROM pg_namespace WHERE nspname = 'shared'
)
UNION ALL
SELECT 'public' AS schema_name
) AS schemas
), false)",
)
.execute(conn)
.map(|_| ())
.map_err(diesel::r2d2::Error::QueryError)
}
}
pub fn init_pool(database_url: &str) -> anyhow::Result<PgPool> {
init_pool_with_size(database_url, DEFAULT_MAX_POOL_SIZE)
}
@@ -17,6 +45,7 @@ pub fn init_pool_with_size(database_url: &str, max_size: u32) -> anyhow::Result<
let pool = Pool::builder()
.max_size(pool_size)
.connection_timeout(Duration::from_secs(10))
.connection_customizer(Box::new(SchemaCustomizer))
.build(manager)?;
Ok(pool)
}
backend/src/routes/auth.rs
+58 -29
View File
@@ -26,14 +26,17 @@ use crate::{
},
error::{AppError, AppResult},
models::{
MagicToken, MagicTokenKind, NewRefreshToken, NewUser, RefreshToken, Tenant, TenantStatus,
User, UserMembership,
MagicToken, MagicTokenKind, NewRefreshToken, NewUser, RefreshToken, TenantStatus, User,
},
schema::{
magic_tokens::dsl as magic_dsl, refresh_tokens, tenants::dsl as tenant_dsl,
user_memberships::dsl as memberships_dsl, user_passkeys::dsl as passkey_dsl, users::dsl,
},
state::AppState,
tenants::{
apply_refresh_token_hash, apply_tenant_guc, apply_user_guc, clear_refresh_token_hash,
clear_user_guc,
},
};
use crate::schema::refresh_tokens::dsl as refresh_dsl;
@@ -337,6 +340,7 @@ pub async fn refresh(
let now = Utc::now();
let now_naive = now.naive_utc();
apply_refresh_token_hash(&mut conn, &hashed)?;
let token = match refresh_dsl::refresh_tokens
.filter(refresh_dsl::token_hash.eq(&hashed))
.filter(refresh_dsl::revoked_at.is_null())
@@ -348,6 +352,10 @@ pub async fn refresh(
Err(err) => return Err(AppError::from(err)),
};
clear_refresh_token_hash(&mut conn)?;
apply_tenant_guc(&mut conn, token.tenant_id)?;
clear_user_guc(&mut conn)?;
diesel::update(refresh_dsl::refresh_tokens.filter(refresh_dsl::id.eq(token.id)))
.set((
refresh_dsl::revoked_at.eq(now_naive),
@@ -398,15 +406,17 @@ pub async fn select_tenant(
};
let mut conn = state.db_unscoped()?;
apply_user_guc(&mut conn, user_id)?;
let membership_exists = memberships_dsl::user_memberships
.filter(memberships_dsl::user_id.eq(user_id))
.filter(memberships_dsl::tenant_id.eq(payload.tenant_id))
.inner_join(tenant_dsl::tenants)
.select(memberships_dsl::id)
.select(memberships_dsl::tenant_id)
.first::<Uuid>(&mut conn)
.optional()?;
clear_user_guc(&mut conn)?;
if membership_exists.is_none() {
return Err(AppError::unauthorized());
}
@@ -430,7 +440,7 @@ pub async fn logout(
user: AuthenticatedUser,
jar: Option<TypedHeader<Cookie>>,
) -> AppResult<(HeaderMap, StatusCode)> {
let mut conn = state.db_unscoped()?;
let mut conn = state.db_for_tenant(user.tenant_id)?;
let now = Utc::now().naive_utc();
let mut rows_affected = 0;
@@ -505,15 +515,26 @@ pub async fn list_tenants(
};
let mut conn = state.db_unscoped()?;
apply_user_guc(&mut conn, user_id)?;
let tenants = memberships_dsl::user_memberships
.inner_join(tenant_dsl::tenants)
let tenant_ids: Vec<Uuid> = memberships_dsl::user_memberships
.filter(memberships_dsl::user_id.eq(user_id))
.select((tenant_dsl::id, tenant_dsl::name))
.load::<(Uuid, String)>(&mut conn)?
.into_iter()
.map(|(id, name)| TenantSnippet { id, name })
.collect();
.select(memberships_dsl::tenant_id)
.load(&mut conn)?;
clear_user_guc(&mut conn)?;
drop(conn);
let mut tenants = Vec::with_capacity(tenant_ids.len());
for tenant_id in tenant_ids {
let mut tenant_conn = state.db_for_tenant(tenant_id)?;
let name: String = tenant_dsl::tenants
.find(tenant_id)
.select(tenant_dsl::name)
.first(&mut tenant_conn)
.map_err(AppError::from)?;
tenants.push(TenantSnippet { id: tenant_id, name });
}
Ok(Json(TenantListResponse { tenants }))
}
@@ -642,26 +663,27 @@ fn complete_login(
user: &User,
preferred_tenant_id: Option<Uuid>,
) -> AppResult<Response> {
let memberships: Vec<(UserMembership, Tenant)> = memberships_dsl::user_memberships
.inner_join(tenant_dsl::tenants)
apply_user_guc(conn, user.id)?;
let tenant_ids: Vec<Uuid> = memberships_dsl::user_memberships
.filter(memberships_dsl::user_id.eq(user.id))
.select(memberships_dsl::tenant_id)
.load(conn)?;
clear_user_guc(conn)?;
if memberships.is_empty() {
tracing::debug!(user_id = %user.id, tenants = tenant_ids.len(), "passkey login memberships");
if tenant_ids.is_empty() {
return Err(AppError::unauthorized());
}
if let Some(preferred_id) = preferred_tenant_id {
if let Some((_, tenant)) = memberships
.iter()
.find(|(_, tenant)| tenant.id == preferred_id)
{
return issue_session(state, conn, user, tenant.id);
if tenant_ids.iter().any(|id| *id == preferred_id) {
return issue_session(state, conn, user, preferred_id);
}
}
if memberships.len() == 1 {
return issue_session(state, conn, user, memberships[0].1.id);
if tenant_ids.len() == 1 {
return issue_session(state, conn, user, tenant_ids[0]);
}
let selection_token = state
@@ -669,13 +691,16 @@ fn complete_login(
.generate_tenant_selector_token(user.id)
.map_err(AppError::from)?;
let tenants = memberships
.into_iter()
.map(|(_, tenant)| TenantSnippet {
id: tenant.id,
name: tenant.name,
})
.collect();
let mut tenants = Vec::with_capacity(tenant_ids.len());
for tenant_id in tenant_ids {
let mut tenant_conn = state.db_for_tenant(tenant_id)?;
let name: String = tenant_dsl::tenants
.find(tenant_id)
.select(tenant_dsl::name)
.first(&mut tenant_conn)
.map_err(AppError::from)?;
tenants.push(TenantSnippet { id: tenant_id, name });
}
Ok(Json(TenantSelectionResponse {
access_token: selection_token,
@@ -747,6 +772,10 @@ fn issue_session(
user: &User,
tenant_id: Uuid,
) -> AppResult<Response> {
apply_tenant_guc(conn, tenant_id)?;
clear_user_guc(conn)?;
clear_refresh_token_hash(conn)?;
let now = Utc::now();
let access_token = state
.jwt
backend/src/routes/webdav/mod.rs
+13 -8
View File
@@ -21,10 +21,11 @@ use crate::error::{AppError, AppResult};
use crate::models::{Document, DocumentVersion, Folder, User};
use crate::schema::{
document_versions::dsl as document_versions_dsl, documents::dsl as documents_dsl,
folders::dsl as folders_dsl, tenants::dsl as tenant_dsl,
user_memberships::dsl as memberships_dsl, users::dsl as users_dsl,
folders::dsl as folders_dsl, user_memberships::dsl as memberships_dsl,
users::dsl as users_dsl,
};
use crate::state::AppState;
use crate::tenants::{apply_tenant_guc, apply_user_guc, clear_user_guc};
use crate::utils::{error::StorageResultExt, http::inline_content_disposition, time::to_http_date};
const REALM: &str = "Papercrate WebDAV";
@@ -453,16 +454,19 @@ fn authenticate(state: &AppState, headers: &HeaderMap) -> Result<Option<WebDavCo
}
};
let tenant_row = memberships_dsl::user_memberships
.inner_join(tenant_dsl::tenants)
apply_user_guc(&mut conn, user.id)?;
let membership_exists = memberships_dsl::user_memberships
.filter(memberships_dsl::user_id.eq(user.id))
.filter(memberships_dsl::tenant_id.eq(token.tenant_id))
.select((tenant_dsl::id, tenant_dsl::name))
.first::<(Uuid, String)>(&mut conn)
.select(memberships_dsl::tenant_id)
.first::<Uuid>(&mut conn)
.optional()?;
let (tenant_id, _name) = match tenant_row {
Some(row) => row,
clear_user_guc(&mut conn)?;
let tenant_id = match membership_exists {
Some(id) => id,
None => {
tracing::warn!(
%username,
@@ -473,6 +477,7 @@ fn authenticate(state: &AppState, headers: &HeaderMap) -> Result<Option<WebDavCo
}
};
apply_tenant_guc(&mut conn, tenant_id)?;
touch_webdav_token(&mut conn, token.id)?;
tracing::debug!(
backend/src/state.rs
+6 -3
View File
@@ -12,7 +12,7 @@ use crate::{
db::PgPool,
error::{AppError, AppResult},
storage::{ObjectStorage, TenantStorage},
tenants::{apply_tenant_guc, TenantService},
tenants::{apply_tenant_guc, clear_tenant_context, clear_user_guc, TenantService},
};
pub type PgPooledConnection = PooledConnection<ConnectionManager<PgConnection>>;
@@ -75,14 +75,17 @@ impl AppState {
debug_assert!(!tenant_id.is_nil(), "nil tenant_id passed to db_for_tenant");
let mut conn = self.db_unscoped()?;
apply_tenant_guc(&mut conn, tenant_id)?;
clear_user_guc(&mut conn)?;
Ok(conn)
}
pub(crate) fn db_unscoped(&self) -> AppResult<PgPooledConnection> {
self.pool.get().map_err(|err| {
let mut conn = self.pool.get().map_err(|err| {
tracing::error!(error = ?err, "database pool error");
AppError::internal("database pool error")
})
})?;
clear_tenant_context(&mut conn)?;
Ok(conn)
}
pub fn storage_for_tenant(&self, tenant_id: Uuid) -> AppResult<TenantStorage> {
backend/src/tenants.rs
+43 -1
View File
@@ -128,13 +128,55 @@ impl TenantService {
}
pub fn apply_tenant_guc(conn: &mut PgConnection, tenant_id: Uuid) -> AppResult<()> {
diesel::sql_query("SELECT set_config('papercrate.tenant_id', $1, true)")
diesel::sql_query("SELECT set_config('papercrate.tenant_id', $1, false)")
.bind::<Text, _>(tenant_id.to_string())
.execute(conn)
.map(|_| ())
.map_err(AppError::from)
}
pub fn apply_user_guc(conn: &mut PgConnection, user_id: Uuid) -> AppResult<()> {
diesel::sql_query("SELECT set_config('papercrate.user_id', $1, false)")
.bind::<Text, _>(user_id.to_string())
.execute(conn)
.map(|_| ())
.map_err(AppError::from)
}
pub fn clear_tenant_context(conn: &mut PgConnection) -> AppResult<()> {
diesel::sql_query(
"SELECT \
set_config('papercrate.tenant_id', '', false), \
set_config('papercrate.user_id', '', false), \
set_config('papercrate.refresh_token_hash', '', false)"
)
.execute(conn)
.map(|_| ())
.map_err(AppError::from)
}
pub fn clear_user_guc(conn: &mut PgConnection) -> AppResult<()> {
diesel::sql_query("SELECT set_config('papercrate.user_id', '', false)")
.execute(conn)
.map(|_| ())
.map_err(AppError::from)
}
pub fn apply_refresh_token_hash(conn: &mut PgConnection, hash: &str) -> AppResult<()> {
diesel::sql_query("SELECT set_config('papercrate.refresh_token_hash', $1, false)")
.bind::<Text, _>(hash)
.execute(conn)
.map(|_| ())
.map_err(AppError::from)
}
pub fn clear_refresh_token_hash(conn: &mut PgConnection) -> AppResult<()> {
diesel::sql_query("SELECT set_config('papercrate.refresh_token_hash', '', false)")
.execute(conn)
.map(|_| ())
.map_err(AppError::from)
}
fn normalize_storage_root(raw: Option<&str>, tenant_id: Uuid) -> String {
match raw.map(str::trim) {
Some(root) if !root.is_empty() => {
backend/src/workers/analyze.rs
+12 -2
View File
@@ -91,7 +91,12 @@ fn analyze_document(
let version: DocumentVersion = document_versions::table
.find(payload.document_version_id)
.first(&mut conn)
.map_err(|err| format!("{err:?}"))?;
.map_err(|err| {
format!(
"failed to load document_version {} for tenant {}: {err:?}",
payload.document_version_id, tenant_id
)
})?;
if version.document_id != payload.document_id {
return Err("document/version mismatch".into());
@@ -100,7 +105,12 @@ fn analyze_document(
let document: Document = documents::table
.find(payload.document_id)
.first(&mut conn)
.map_err(|err| format!("{err:?}"))?;
.map_err(|err| {
format!(
"failed to load document {} for tenant {}: {err:?}",
payload.document_id, tenant_id
)
})?;
let tenant_id = document.tenant_id;
backend/src/workers/common.rs
+4 -3
View File
@@ -15,10 +15,13 @@ pub(crate) struct DocumentVersionContext {
pub(crate) fn load_document_version(
state: &AppState,
tenant_id: Uuid,
document_id: Uuid,
version_id: Uuid,
) -> Result<DocumentVersionContext, String> {
let mut conn = state.db_unscoped().map_err(|err| format!("{err:?}"))?;
let mut conn = state
.db_for_tenant(tenant_id)
.map_err(|err| format!("{err:?}"))?;
let version: DocumentVersion = document_versions::table
.find(version_id)
@@ -34,8 +37,6 @@ pub(crate) fn load_document_version(
.first(&mut conn)
.map_err(|err| format!("{err:?}"))?;
let tenant_id = document.tenant_id;
Ok(DocumentVersionContext {
document,
version,
backend/src/workers/index.rs
+12 -13
View File
@@ -89,8 +89,8 @@ impl JobHandler for IndexDocumentTextJob {
let client = Client::new();
let state_clone = state.clone();
let context = match task::spawn_blocking(move || load_context(state_clone, &payload)).await
{
let tenant_id = job.tenant_id;
let context = match task::spawn_blocking(move || load_context(state_clone, tenant_id, payload)).await {
Ok(Ok(ctx)) => ctx,
Ok(Err(err)) => {
warn!(job_id = %job.id, error = %err, "index job will retry");
@@ -166,12 +166,18 @@ struct IndexContext {
text_s3_key: Option<String>,
}
fn load_context(state: Arc<AppState>, payload: &IndexPayload) -> Result<IndexContext, String> {
let mut base_conn = state.db_unscoped().map_err(|err| format!("{err:?}"))?;
fn load_context(
state: Arc<AppState>,
tenant_id: Uuid,
payload: IndexPayload,
) -> Result<IndexContext, String> {
let mut conn = state
.db_for_tenant(tenant_id)
.map_err(|err| format!("{err:?}"))?;
let version: DocumentVersion = document_versions::table
.find(payload.document_version_id)
.first(&mut base_conn)
.first(&mut conn)
.map_err(|err| format!("{err:?}"))?;
if version.document_id != payload.document_id {
@@ -180,14 +186,7 @@ fn load_context(state: Arc<AppState>, payload: &IndexPayload) -> Result<IndexCon
let document: Document = documents::table
.find(payload.document_id)
.first(&mut base_conn)
.map_err(|err| format!("{err:?}"))?;
let tenant_id = document.tenant_id;
drop(base_conn);
let mut conn = state
.db_for_tenant(tenant_id)
.first(&mut conn)
.map_err(|err| format!("{err:?}"))?;
let text_s3_key: Option<String> = document_asset_objects::table
backend/src/workers/ocr.rs
+15 -15
View File
@@ -24,7 +24,7 @@ use crate::{
Document, DocumentAsset, DocumentAssetObject, DocumentVersion, NewDocumentAsset,
NewDocumentAssetObject,
},
schema::{document_asset_objects, document_assets, documents},
schema::{document_asset_objects, document_assets},
state::AppState,
storage::TenantStorage,
utils::storage_paths::document_asset_object_prefix,
@@ -77,9 +77,12 @@ impl JobHandler for GenerateOcrTextJob {
let state_clone = state.clone();
let payload_clone = payload.clone();
let context =
match task::spawn_blocking(move || load_ocr_context(state_clone, &payload_clone)).await
{
let tenant_id = job.tenant_id;
let context = match task::spawn_blocking(move || {
load_ocr_context(state_clone, tenant_id, payload_clone)
})
.await
{
Ok(Ok(ctx)) => ctx,
Ok(Err(err)) => {
warn!(job_id = %job.id, error = %err, "ocr job will retry");
@@ -195,7 +198,7 @@ impl JobHandler for GenerateOcrTextJob {
.await
{
Ok(Ok(())) => {
if let Err(err) = enqueue_index_job(&state, &payload) {
if let Err(err) = enqueue_index_job(&state, job.tenant_id, &payload) {
warn!(job_id = %job.id, error = %err, "failed to enqueue index job");
}
JobExecution::Success
@@ -236,9 +239,14 @@ struct OcrGeneration {
source: &'static str,
}
fn load_ocr_context(state: Arc<AppState>, payload: &OcrPayload) -> Result<OcrContext, String> {
fn load_ocr_context(
state: Arc<AppState>,
tenant_id: Uuid,
payload: OcrPayload,
) -> Result<OcrContext, String> {
let base = load_document_version(
state.as_ref(),
tenant_id,
payload.document_id,
payload.document_version_id,
)?;
@@ -481,15 +489,7 @@ fn persist_ocr_metadata(
Ok(())
}
fn enqueue_index_job(state: &AppState, payload: &OcrPayload) -> Result<(), String> {
let mut base_conn = state.db_unscoped().map_err(|err| format!("{err:?}"))?;
let tenant_id: Uuid = documents::table
.find(payload.document_id)
.select(documents::tenant_id)
.first(&mut base_conn)
.map_err(|err| format!("{err:?}"))?;
drop(base_conn);
fn enqueue_index_job(state: &AppState, tenant_id: Uuid, payload: &OcrPayload) -> Result<(), String> {
let mut conn = state
.db_for_tenant(tenant_id)
.map_err(|err| format!("{err:?}"))?;
backend/src/workers/thumbnails.rs
+19 -12
View File
@@ -18,7 +18,7 @@ use crate::{
Document, DocumentAsset, DocumentAssetObject, DocumentVersion, NewDocumentAsset,
NewDocumentAssetObject,
},
schema::{document_asset_objects, document_assets, document_versions, documents},
schema::{document_asset_objects, document_assets, document_versions},
state::AppState,
storage::TenantStorage,
utils::storage_paths::document_asset_object_key,
@@ -37,7 +37,7 @@ const PREVIEW_HEIGHT: u32 = THUMBNAIL_HEIGHT * 4;
const THUMBNAIL_ASSET_TYPE: &str = "thumbnail";
const PREVIEW_ASSET_TYPE: &str = "preview";
#[derive(Debug, Deserialize)]
#[derive(Debug, Deserialize, Clone)]
struct ThumbnailPayload {
document_id: Uuid,
document_version_id: Uuid,
@@ -75,9 +75,13 @@ impl JobHandler for GenerateThumbnailsJob {
};
let state_clone = state.clone();
let initial =
match task::spawn_blocking(move || load_thumbnail_context(state_clone, &payload)).await
{
let tenant_id = job.tenant_id;
let payload_for_context = payload.clone();
let initial = match task::spawn_blocking(move || {
load_thumbnail_context(state_clone, tenant_id, payload_for_context)
})
.await
{
Ok(Ok(ctx)) => ctx,
Ok(Err(err)) => {
warn!(job_id = %job.id, error = %err, "thumbnail job will retry");
@@ -123,8 +127,9 @@ impl JobHandler for GenerateThumbnailsJob {
let state_clone = state.clone();
let document_id = initial.document.id;
let version_id = initial.version.id;
let tenant_id = initial.tenant_id;
match task::spawn_blocking(move || {
persist_document_page_count(state_clone, document_id, version_id, page_count)
persist_document_page_count(state_clone, tenant_id, document_id, version_id, page_count)
})
.await
{
@@ -372,6 +377,7 @@ struct ThumbnailContext {
existing_preview: Option<DocumentAsset>,
existing_preview_objects: Vec<DocumentAssetObject>,
skip: bool,
tenant_id: Uuid,
}
struct GeneratedImage {
@@ -405,10 +411,12 @@ struct AssetPersistence {
fn load_thumbnail_context(
state: Arc<AppState>,
payload: &ThumbnailPayload,
tenant_id: Uuid,
payload: ThumbnailPayload,
) -> Result<ThumbnailContext, String> {
let base = load_document_version(
state.as_ref(),
tenant_id,
payload.document_id,
payload.document_version_id,
)?;
@@ -467,6 +475,7 @@ fn load_thumbnail_context(
existing_preview,
existing_preview_objects,
skip,
tenant_id: base.tenant_id,
})
}
@@ -708,15 +717,13 @@ fn persist_assets_metadata(
fn persist_document_page_count(
state: Arc<AppState>,
tenant_id: Uuid,
document_id: Uuid,
document_version_id: Uuid,
page_count: u32,
) -> Result<(), String> {
let mut conn = state.db_unscoped().map_err(|err| format!("{err:?}"))?;
let tenant_id: Uuid = documents::table
.find(document_id)
.select(documents::tenant_id)
.first(&mut conn)
let mut conn = state
.db_for_tenant(tenant_id)
.map_err(|err| format!("{err:?}"))?;
let existing_metadata: Value = document_versions::table