diff --git a/backend/migrations/202512010120_add_tenant_status_column/down.sql b/backend/migrations/202512010120_add_tenant_status_column/down.sql new file mode 100644 index 0000000..3e2ff3d --- /dev/null +++ b/backend/migrations/202512010120_add_tenant_status_column/down.sql @@ -0,0 +1,5 @@ +ALTER TABLE tenants + DROP COLUMN IF EXISTS status, + DROP COLUMN IF EXISTS created_by; + +DROP TYPE IF EXISTS tenant_status; diff --git a/backend/migrations/202512010120_add_tenant_status_column/up.sql b/backend/migrations/202512010120_add_tenant_status_column/up.sql new file mode 100644 index 0000000..f102138 --- /dev/null +++ b/backend/migrations/202512010120_add_tenant_status_column/up.sql @@ -0,0 +1,19 @@ +ALTER TABLE tenants + DROP COLUMN IF EXISTS status; + +ALTER TABLE tenants + DROP COLUMN IF EXISTS created_by; + +DROP TYPE IF EXISTS tenant_status; + +CREATE TYPE tenant_status AS ENUM ('creating', 'active', 'suspended', 'deleting', 'error'); + +ALTER TABLE tenants + ADD COLUMN status tenant_status, + ADD COLUMN created_by UUID; + +UPDATE tenants +SET status = 'active'; + +ALTER TABLE tenants + ALTER COLUMN status SET NOT NULL; diff --git a/backend/src/auth/password.rs b/backend/src/auth/password.rs index 2b4f2be..ff7489f 100644 --- a/backend/src/auth/password.rs +++ b/backend/src/auth/password.rs @@ -1,8 +1,9 @@ use anyhow::{anyhow, Result}; use argon2::{ - password_hash::{PasswordHash, PasswordVerifier}, + password_hash::{PasswordHash, PasswordHasher, PasswordVerifier, SaltString}, Argon2, }; +use rand::rngs::OsRng; pub fn verify_password(password: &str, password_hash: &str) -> Result { let parsed_hash = PasswordHash::new(password_hash).map_err(|err| anyhow!(err))?; @@ -10,3 +11,11 @@ pub fn verify_password(password: &str, password_hash: &str) -> Result { .verify_password(password.as_bytes(), &parsed_hash) .is_ok()) } + +pub fn hash_password(password: &str) -> Result { + let salt = SaltString::generate(&mut OsRng); + let hash = Argon2::default() + .hash_password(password.as_bytes(), &salt) + .map_err(|err| anyhow!(err))?; + Ok(hash.to_string()) +} diff --git a/backend/src/auth/webdav_tokens.rs b/backend/src/auth/webdav_tokens.rs index 8e919ba..389a631 100644 --- a/backend/src/auth/webdav_tokens.rs +++ b/backend/src/auth/webdav_tokens.rs @@ -140,21 +140,18 @@ pub fn touch_webdav_token(conn: &mut PgPooledConnection, token_id: Uuid) -> Resu } pub fn verify_token_secret(secret: &str, token_hash: &str) -> Result { - crate::auth::password::verify_password(secret, token_hash) - .map_err(|err| { - tracing::error!(error = ?err, "failed to verify token"); - AppError::internal("failed to verify token") - }) + crate::auth::password::verify_password(secret, token_hash).map_err(|err| { + tracing::error!(error = ?err, "failed to verify token"); + AppError::internal("failed to verify token") + }) } fn generate_secret() -> Result { let mut buffer = [0u8; TOKEN_SECRET_LENGTH]; - OsRng - .try_fill_bytes(&mut buffer) - .map_err(|err| { - tracing::error!(error = ?err, "failed to generate token"); - AppError::internal("failed to generate token") - })?; + OsRng.try_fill_bytes(&mut buffer).map_err(|err| { + tracing::error!(error = ?err, "failed to generate token"); + AppError::internal("failed to generate token") + })?; Ok(hex::encode(buffer)) } diff --git a/backend/src/bin/admin.rs b/backend/src/bin/admin.rs index 63b2062..19deca1 100644 --- a/backend/src/bin/admin.rs +++ b/backend/src/bin/admin.rs @@ -2,59 +2,28 @@ use std::env; use std::sync::Arc; use anyhow::{anyhow, bail, Context, Result}; -use argon2::{ - password_hash::{PasswordHasher, SaltString}, - Argon2, -}; use diesel::{dsl::exists, prelude::*, select}; -use once_cell::sync::Lazy; use reqwest::{Client, Method, StatusCode}; -use serde_json::json; use uuid::Uuid; use backend::{ + auth::password, config::AppConfig, db::{self, PgPool}, + documents::search::ensure_quickwit_index, jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT}, - models::{DocumentAsset, DocumentAssetObject, NewUser, NewUserMembership, Tenant, User}, + models::{ + DocumentAsset, DocumentAssetObject, NewUser, NewUserMembership, Tenant, TenantStatus, User, + }, s3, schema::{ document_asset_objects, document_assets, documents, tenants, user_memberships, users, }, storage::{ObjectStorage, S3Storage, TenantStorage}, + tenants::TenantService, utils::tracing::init_tracing, }; -use rand::rngs::OsRng; - -static QUICKWIT_INDEX_TEMPLATE: Lazy = Lazy::new(|| { - json!({ - "version": "0.8", - "index_id": "documents", - "doc_mapping": { - "tokenizers": [ - { - "name": "substring", - "type": "ngram", - "min_gram": 2, - "max_gram": 20, - "prefix_only": false - } - ], - "field_mappings": [ - { "name": "tenant_id", "type": "text", "stored": true }, - { "name": "document_id", "type": "text", "stored": true }, - { "name": "version_id", "type": "text", "stored": true }, - { "name": "title", "type": "text", "tokenizer": "substring", "stored": true }, - { "name": "text", "type": "text", "tokenizer": "substring", "record": "position" } - ] - }, - "search_settings": { - "default_search_fields": ["title", "text"] - } - }) -}); - #[derive(Debug)] enum Command { CreateUser { @@ -218,7 +187,7 @@ fn create_user(pool: &PgPool, username: &str, password: &str) -> Result<()> { bail!("user '{}' already exists", username); } - let password_hash = hash_password(password)?; + let password_hash = password::hash_password(password).map_err(|err| anyhow!(err))?; let new_user = NewUser { id: Uuid::new_v4(), username: username.to_string(), @@ -239,7 +208,7 @@ fn set_password(pool: &PgPool, username: &str, password: &str) -> Result<()> { } let mut conn = pool.get().context("failed to get database connection")?; - let password_hash = hash_password(password)?; + let password_hash = password::hash_password(password).map_err(|err| anyhow!(err))?; let updated = diesel::update(users::table.filter(users::username.eq(username))) .set(users::password_hash.eq(password_hash)) @@ -253,14 +222,6 @@ fn set_password(pool: &PgPool, username: &str, password: &str) -> Result<()> { Ok(()) } -fn hash_password(password: &str) -> Result { - let salt = SaltString::generate(&mut OsRng); - let hash = Argon2::default() - .hash_password(password.as_bytes(), &salt) - .map_err(|err| anyhow!(err))?; - Ok(hash.to_string()) -} - fn list_users(pool: &PgPool) -> Result<()> { let mut conn = pool.get().context("failed to get database connection")?; @@ -315,46 +276,28 @@ fn create_tenant( storage_root_arg: Option, quickwit_index_arg: Option, ) -> Result<()> { - if slug.trim().is_empty() { - bail!("tenant slug must not be empty"); - } + let service = TenantService::new(pool.clone()); + let tenant = service + .create_tenant( + slug, + storage_root_arg.as_deref(), + quickwit_index_arg.as_deref(), + TenantStatus::Creating, + &[], + None, + ) + .map_err(|err| anyhow!(format!("{err:?}")))?; - let mut conn = pool.get().context("failed to get database connection")?; - let exists: bool = - select(exists(tenants::table.filter(tenants::slug.eq(slug)))).get_result(&mut conn)?; - if exists { - bail!("tenant '{}' already exists", slug); - } - - let id = Uuid::new_v4(); - let storage_root = storage_root_arg - .map(|mut s| { - if s.is_empty() { - format!("tenants/{}/", id) - } else { - if !s.ends_with('/') { - s.push('/'); - } - s - } - }) - .unwrap_or_else(|| format!("tenants/{}/", id)); - let quickwit_index = quickwit_index_arg.unwrap_or_else(|| format!("documents-{}", id)); - - diesel::insert_into(tenants::table) - .values(( - tenants::id.eq(id), - tenants::slug.eq(slug), - tenants::storage_root.eq(Some(storage_root.clone())), - tenants::quickwit_index.eq(Some(quickwit_index.clone())), - tenants::status.eq("active"), - tenants::config.eq(serde_json::json!({})), - )) - .execute(&mut conn)?; + let storage_root = tenant.storage_root.as_deref().unwrap_or(""); + let quickwit_index = tenant.quickwit_index.as_deref().unwrap_or(""); println!( - "created tenant '{}' with id {}, storage_root '{}', quickwit_index '{}'", - slug, id, storage_root, quickwit_index + "created tenant '{}' with id {}, storage_root '{}', quickwit_index '{}', status '{}'", + tenant.slug, + tenant.id, + storage_root, + quickwit_index, + tenant.status.as_str() ); Ok(()) } @@ -605,63 +548,19 @@ async fn quickwit_index( match method { Method::POST => { - let payload = render_index_template(&index_id); - let response = client - .post(format!("{}/api/v1/indexes", base_endpoint)) - .header("content-type", "application/json") - .body(payload) - .send() + ensure_quickwit_index(&client, base_endpoint, &index_id) .await - .context("failed to send create index request")?; + .context("failed to ensure quickwit index")?; - match response.status() { - status if status.is_success() => { - diesel::update(tenants::table.filter(tenants::id.eq(tenant.id))) - .set(tenants::quickwit_index.eq(Some(index_id.clone()))) - .execute(&mut conn) - .context("failed to update tenant quickwit_index")?; + diesel::update(tenants::table.filter(tenants::id.eq(tenant.id))) + .set(tenants::quickwit_index.eq(Some(index_id.clone()))) + .execute(&mut conn) + .context("failed to update tenant quickwit_index")?; - println!( - "Tenant '{}' quickwit index set to '{}'.", - tenant.slug, index_id - ); - } - StatusCode::CONFLICT => { - let lookup = client - .get(format!("{}/api/v1/indexes/{}", base_endpoint, index_id)) - .send() - .await - .context("failed to verify existing quickwit index")?; - - let lookup_status = lookup.status(); - if !lookup_status.is_success() { - let body = lookup.text().await.unwrap_or_default(); - bail!( - "quickwit reported conflict but index lookup failed with status {}: {}", - lookup_status, - body - ); - } - - diesel::update(tenants::table.filter(tenants::id.eq(tenant.id))) - .set(tenants::quickwit_index.eq(Some(index_id.clone()))) - .execute(&mut conn) - .context("failed to update tenant quickwit_index")?; - - println!( - "Tenant '{}' quickwit index set to '{}'.", - tenant.slug, index_id - ); - } - status => { - let body = response.text().await.unwrap_or_default(); - bail!( - "quickwit create index failed with status {}: {}", - status, - body - ); - } - } + println!( + "Tenant '{}' quickwit index set to '{}'.", + tenant.slug, index_id + ); } Method::DELETE => { let response = client @@ -694,14 +593,3 @@ async fn quickwit_index( Ok(()) } - -fn render_index_template(index_id: &str) -> String { - let mut template = QUICKWIT_INDEX_TEMPLATE.clone(); - if let Some(obj) = template.as_object_mut() { - obj.insert( - "index_id".to_string(), - serde_json::Value::String(index_id.to_string()), - ); - } - template.to_string() -} diff --git a/backend/src/documents/search.rs b/backend/src/documents/search.rs index 976a9d4..d1e41c9 100644 --- a/backend/src/documents/search.rs +++ b/backend/src/documents/search.rs @@ -1,7 +1,7 @@ use std::collections::HashSet; -use anyhow::{anyhow, Result}; -use reqwest::Client; +use anyhow::{anyhow, bail, Result}; +use reqwest::{Client, StatusCode}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; use tracing::{debug, error}; @@ -109,6 +109,60 @@ pub async fn quickwit_search( Ok(doc_ids) } +pub fn quickwit_index_template(index_id: &str) -> Value { + json!({ + "version": "0.8", + "index_id": index_id, + "doc_mapping": { + "tokenizers": [ + { + "name": "substring", + "type": "ngram", + "min_gram": 2, + "max_gram": 20, + "prefix_only": false + } + ], + "field_mappings": [ + { "name": "tenant_id", "type": "text", "stored": true }, + { "name": "document_id", "type": "text", "stored": true }, + { "name": "version_id", "type": "text", "stored": true }, + { "name": "title", "type": "text", "tokenizer": "substring", "stored": true }, + { "name": "text", "type": "text", "tokenizer": "substring", "record": "position" } + ] + }, + "search_settings": { + "default_search_fields": ["title", "text"] + } + }) +} + +pub async fn ensure_quickwit_index(client: &Client, endpoint: &str, index_id: &str) -> Result<()> { + let base = endpoint.trim_end_matches('/'); + let create_url = format!("{}/api/v1/indexes", base); + let payload = quickwit_index_template(index_id); + + let response = client.post(&create_url).json(&payload).send().await?; + match response.status() { + status if status.is_success() => Ok(()), + StatusCode::CONFLICT => { + let lookup_url = format!("{}/api/v1/indexes/{}", base, index_id); + let lookup = client.get(&lookup_url).send().await?; + if lookup.status().is_success() { + Ok(()) + } else { + let status = lookup.status(); + let body = lookup.text().await.unwrap_or_default(); + bail!("quickwit index lookup failed with status {status}: {body}"); + } + } + status => { + let body = response.text().await.unwrap_or_default(); + bail!("quickwit create index failed with status {status}: {body}"); + } + } +} + pub fn extract_document_id(hit: &Value) -> Option { for key in ["_source", "source", "fields", "stored_fields"] { if let Some(value) = hit.get(key) { @@ -226,7 +280,9 @@ pub async fn quickwit_ingest( let status = response.status(); let body = response.text().await.unwrap_or_default(); error!(%status, %body, "quickwit ingest request failed"); - return Err(anyhow!("quickwit ingest failed with status {status}: {body}")); + return Err(anyhow!( + "quickwit ingest failed with status {status}: {body}" + )); } debug!("quickwit ingest request succeeded"); diff --git a/backend/src/jobs.rs b/backend/src/jobs.rs index d569cec..14c3a4d 100644 --- a/backend/src/jobs.rs +++ b/backend/src/jobs.rs @@ -19,6 +19,7 @@ pub const JOB_ANALYZE_DOCUMENT: &str = "analyze-document"; pub const JOB_GENERATE_THUMBNAILS: &str = "generate-thumbnails"; pub const JOB_GENERATE_OCR_TEXT: &str = "generate-ocr-text"; pub const JOB_INDEX_DOCUMENT_TEXT: &str = "index-document-text"; +pub const JOB_PROVISION_TENANT: &str = "provision-tenant"; #[derive(Debug, Error)] pub enum JobQueueError { diff --git a/backend/src/models.rs b/backend/src/models.rs index 851452d..2c01896 100644 --- a/backend/src/models.rs +++ b/backend/src/models.rs @@ -1,7 +1,16 @@ use chrono::NaiveDateTime; +use diesel::deserialize::FromSql; +use diesel::pg::{Pg, PgValue}; use diesel::prelude::*; +use diesel::serialize::{IsNull, Output, ToSql}; +use diesel::{deserialize, serialize, AsExpression, FromSqlRow}; +use serde_json::Value; +use std::fmt; +use std::io::Write; +use std::str; use uuid::Uuid; +use crate::schema::sql_types::TenantStatus as TenantStatusSql; use crate::schema::*; #[derive(Debug, Clone, Queryable, Identifiable, Associations)] @@ -26,6 +35,65 @@ pub struct NewUserMembership { pub role: String, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, AsExpression, FromSqlRow)] +#[diesel(sql_type = TenantStatusSql)] +pub enum TenantStatus { + Creating, + Active, + Suspended, + Deleting, + Error, +} + +impl TenantStatus { + pub fn as_str(&self) -> &'static str { + match self { + TenantStatus::Creating => "creating", + TenantStatus::Active => "active", + TenantStatus::Suspended => "suspended", + TenantStatus::Deleting => "deleting", + TenantStatus::Error => "error", + } + } + + pub fn from_str(value: &str) -> Option { + match value { + "creating" => Some(TenantStatus::Creating), + "active" => Some(TenantStatus::Active), + "suspended" => Some(TenantStatus::Suspended), + "deleting" => Some(TenantStatus::Deleting), + "error" => Some(TenantStatus::Error), + _ => None, + } + } +} + +impl fmt::Display for TenantStatus { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +impl ToSql for TenantStatus { + fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> serialize::Result { + out.write_all(self.as_str().as_bytes())?; + Ok(IsNull::No) + } +} + +impl FromSql for TenantStatus { + fn from_sql(bytes: PgValue<'_>) -> deserialize::Result { + let value = str::from_utf8(bytes.as_bytes()) + .map_err(|err| Box::::from(err))?; + TenantStatus::from_str(value).ok_or_else(|| { + Box::::from(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("invalid tenant status '{value}'"), + )) + }) + } +} + #[derive(Debug, Clone, Queryable, Identifiable)] #[diesel(table_name = tenants)] #[diesel(primary_key(id))] @@ -34,10 +102,11 @@ pub struct Tenant { pub slug: String, pub storage_root: Option, pub quickwit_index: Option, - pub status: String, - pub config: serde_json::Value, + pub config: Value, pub created_at: NaiveDateTime, pub updated_at: NaiveDateTime, + pub status: TenantStatus, + pub created_by: Option, } #[derive(Debug, Clone, Queryable, Identifiable)] diff --git a/backend/src/openapi.rs b/backend/src/openapi.rs index 7988549..ca4ea0d 100644 --- a/backend/src/openapi.rs +++ b/backend/src/openapi.rs @@ -8,6 +8,7 @@ use uuid::Uuid; paths( doc::health_check, doc::login, + doc::signup, doc::refresh, doc::logout, doc::me, @@ -53,6 +54,7 @@ use uuid::Uuid; components( schemas( schemas::LoginRequest, + schemas::SignupRequest, schemas::AccessTokenResponse, schemas::TenantSnippet, schemas::TenantSelectionResponse, @@ -151,6 +153,19 @@ mod doc { )] pub(super) fn login() {} + #[utoipa::path( + post, + path = "/api/auth/signup", + request_body = SignupRequest, + responses( + (status = 200, description = "Signup succeeded", body = LoginResponseVariants), + (status = 400, description = "Invalid signup request"), + (status = 409, description = "Username already exists") + ), + tag = "Auth" + )] + pub(super) fn signup() {} + #[utoipa::path( post, path = "/api/auth/refresh", @@ -597,6 +612,12 @@ pub mod schemas { pub preferred_tenant_slug: Option, } + #[derive(Serialize, Deserialize, ToSchema)] + pub struct SignupRequest { + pub username: String, + pub password: String, + } + #[derive(Serialize, Deserialize, ToSchema)] pub struct AccessTokenResponse { pub access_token: String, diff --git a/backend/src/routes/auth.rs b/backend/src/routes/auth.rs index 5334fd9..9ec9952 100644 --- a/backend/src/routes/auth.rs +++ b/backend/src/routes/auth.rs @@ -18,7 +18,7 @@ use uuid::Uuid; use crate::{ auth::{password, AuthenticatedUser}, error::{AppError, AppResult}, - models::{NewRefreshToken, RefreshToken, Tenant, User, UserMembership}, + models::{NewRefreshToken, NewUser, RefreshToken, Tenant, TenantStatus, User, UserMembership}, schema::{ refresh_tokens, tenants::dsl as tenant_dsl, user_memberships::dsl as memberships_dsl, users::dsl, @@ -38,6 +38,12 @@ pub struct LoginRequest { pub preferred_tenant_slug: Option, } +#[derive(Deserialize)] +pub struct SignupRequest { + pub username: String, + pub password: String, +} + #[derive(Serialize)] pub struct LoginResponse { pub access_token: String, @@ -68,6 +74,50 @@ pub struct TenantSelectionRequest { pub tenant_id: Uuid, } +pub async fn signup( + State(state): State, + Json(payload): Json, +) -> AppResult { + let username = payload.username.trim(); + let password = payload.password.trim(); + + if username.is_empty() || password.is_empty() { + return Err(AppError::bad_request( + "username and password must not be empty", + )); + } + + let mut conn = state.db_unscoped()?; + + let exists: bool = dsl::users + .filter(dsl::username.eq(username)) + .first::(&mut conn) + .optional()? + .is_some(); + if exists { + return Err(AppError::conflict("username already exists")); + } + + let user_id = Uuid::new_v4(); + let password_hash = password::hash_password(password)?; + + insert_user(&mut conn, user_id, username, &password_hash)?; + + let tenant_slug = username.to_lowercase(); + let tenant = state.tenants.create_tenant( + &tenant_slug, + None, + None, + TenantStatus::Creating, + &[user_id], + Some(user_id), + )?; + + let user: User = dsl::users.find(user_id).first(&mut conn)?; + + issue_session(&state, &mut conn, &user, tenant.id) +} + pub async fn login( State(state): State, Json(payload): Json, @@ -181,6 +231,25 @@ pub async fn refresh( issue_session(&state, &mut conn, &user, token.tenant_id) } +fn insert_user( + conn: &mut PgConnection, + id: Uuid, + username: &str, + password_hash: &str, +) -> AppResult<()> { + let new_user = NewUser { + id, + username: username.to_string(), + password_hash: password_hash.to_string(), + }; + + diesel::insert_into(dsl::users) + .values(&new_user) + .execute(conn) + .map(|_| ()) + .map_err(AppError::from) +} + pub async fn select_tenant( State(state): State, TypedHeader(Authorization(bearer)): TypedHeader>, diff --git a/backend/src/routes/mod.rs b/backend/src/routes/mod.rs index 2d4a0ca..7f68ccc 100644 --- a/backend/src/routes/mod.rs +++ b/backend/src/routes/mod.rs @@ -54,6 +54,7 @@ pub fn create_router(state: AppState) -> Router<()> { }; let auth_routes = Router::new() + .route("/signup", post(auth::signup)) .route("/login", post(auth::login)) .route("/refresh", post(auth::refresh)) .route("/logout", post(auth::logout)) diff --git a/backend/src/schema.rs b/backend/src/schema.rs index 8efb5d6..444577b 100644 --- a/backend/src/schema.rs +++ b/backend/src/schema.rs @@ -1,5 +1,11 @@ // @generated automatically by Diesel CLI. +pub mod sql_types { + #[derive(diesel::query_builder::QueryId, Clone, diesel::sql_types::SqlType)] + #[diesel(postgres_type(name = "tenant_status"))] + pub struct TenantStatus; +} + diesel::table! { correspondents (id) { id -> Uuid, @@ -148,15 +154,19 @@ diesel::table! { } diesel::table! { + use diesel::sql_types::*; + use super::sql_types::TenantStatus; + tenants (id) { id -> Uuid, slug -> Text, storage_root -> Nullable, quickwit_index -> Nullable, - status -> Text, config -> Jsonb, created_at -> Timestamptz, updated_at -> Timestamptz, + status -> TenantStatus, + created_by -> Nullable, } } diff --git a/backend/src/state.rs b/backend/src/state.rs index 6a29e98..986e96e 100644 --- a/backend/src/state.rs +++ b/backend/src/state.rs @@ -69,12 +69,10 @@ impl AppState { } pub(crate) fn db_unscoped(&self) -> AppResult { - self.pool - .get() - .map_err(|err| { - tracing::error!(error = ?err, "database pool error"); - AppError::internal("database pool error") - }) + self.pool.get().map_err(|err| { + tracing::error!(error = ?err, "database pool error"); + AppError::internal("database pool error") + }) } pub fn storage_for_tenant(&self, tenant_id: Uuid) -> AppResult { diff --git a/backend/src/tenants.rs b/backend/src/tenants.rs index 15b58ed..9882729 100644 --- a/backend/src/tenants.rs +++ b/backend/src/tenants.rs @@ -1,11 +1,18 @@ use axum::{async_trait, extract::FromRequestParts, http::request::Parts}; -use diesel::{pg::PgConnection, prelude::*, sql_types::Text}; +use diesel::{ + dsl::{exists, select}, + pg::PgConnection, + prelude::*, + sql_types::Text, +}; +use serde_json::json; use uuid::Uuid; use crate::{ db::PgPool, error::{AppError, AppResult}, - models::Tenant, + jobs::{enqueue_job, JOB_PROVISION_TENANT}, + models::{Tenant, TenantStatus}, schema::tenants::dsl, state::AppState, }; @@ -50,17 +57,72 @@ impl TenantService { Ok(self.get_by_slug(slug)?.id) } + pub fn create_tenant( + &self, + slug: &str, + storage_root: Option<&str>, + quickwit_index: Option<&str>, + status: TenantStatus, + initial_members: &[Uuid], + created_by: Option, + ) -> AppResult { + let slug = slug.trim(); + if slug.is_empty() { + return Err(AppError::bad_request("tenant slug must not be empty")); + } + + let mut conn = self.pool.get().map_err(|err| { + tracing::error!(error = ?err, "database pool error"); + AppError::internal("database pool error") + })?; + + let exists: bool = + select(exists(dsl::tenants.filter(dsl::slug.eq(slug)))).get_result(&mut conn)?; + if exists { + return Err(AppError::conflict(format!( + "tenant '{}' already exists", + slug + ))); + } + + let id = Uuid::new_v4(); + let storage_root = normalize_storage_root(storage_root, id); + let quickwit_index = normalize_quickwit_index(quickwit_index, id); + + diesel::insert_into(dsl::tenants) + .values(( + dsl::id.eq(id), + dsl::slug.eq(slug), + dsl::storage_root.eq(Some(storage_root.clone())), + dsl::quickwit_index.eq(Some(quickwit_index.clone())), + dsl::config.eq(json!({})), + dsl::status.eq(status), + dsl::created_by.eq(created_by), + )) + .execute(&mut conn)?; + + if status == TenantStatus::Creating { + let payload = json!({ + "members": initial_members, + }); + + enqueue_job(&mut conn, id, JOB_PROVISION_TENANT, payload, None).map_err(|err| { + tracing::error!(error = ?err, tenant_id = %id, "failed to enqueue tenant provisioning job"); + AppError::internal("failed to enqueue tenant provisioning job") + })?; + } + + TenantRepository::get_by_id(&mut conn, id) + } + fn load(&self, loader: F) -> AppResult where F: FnOnce(&mut PgConnection) -> AppResult, { - let mut conn = self - .pool - .get() - .map_err(|err| { - tracing::error!(error = ?err, "database pool error"); - AppError::internal("database pool error") - })?; + let mut conn = self.pool.get().map_err(|err| { + tracing::error!(error = ?err, "database pool error"); + AppError::internal("database pool error") + })?; let tenant = loader(&mut conn)?; Ok(tenant) } @@ -92,3 +154,23 @@ impl FromRequestParts for TenantContext { Ok(Self { tenant }) } } + +fn normalize_storage_root(raw: Option<&str>, tenant_id: Uuid) -> String { + match raw.map(str::trim) { + Some(root) if !root.is_empty() => { + let mut owned = root.to_owned(); + if !owned.ends_with('/') { + owned.push('/'); + } + owned + } + _ => format!("tenants/{tenant_id}/"), + } +} + +fn normalize_quickwit_index(raw: Option<&str>, tenant_id: Uuid) -> String { + match raw.map(str::trim) { + Some(value) if !value.is_empty() => value.to_owned(), + _ => format!("documents-{tenant_id}"), + } +} diff --git a/backend/src/workers/index.rs b/backend/src/workers/index.rs index 53c94fb..0277eae 100644 --- a/backend/src/workers/index.rs +++ b/backend/src/workers/index.rs @@ -19,11 +19,7 @@ use crate::{ }; use super::{ - fetch_version_object, - handle_fetch_error, - ocr::OCR_TEXT_ASSET_TYPE, - JobExecution, - JobHandler, + fetch_version_object, handle_fetch_error, ocr::OCR_TEXT_ASSET_TYPE, JobExecution, JobHandler, }; #[derive(Debug, Deserialize)] @@ -148,12 +144,8 @@ impl JobHandler for IndexDocumentTextJob { }; } - let record = build_quickwit_ingest_record( - &context.document, - &context.version, - job.tenant_id, - &text, - ); + let record = + build_quickwit_ingest_record(&context.document, &context.version, job.tenant_id, &text); match quickwit_ingest(&client, &quickwit_endpoint, &quickwit_index, &[record]).await { Ok(()) => JobExecution::Success, diff --git a/backend/src/workers/mod.rs b/backend/src/workers/mod.rs index a6efcb0..fe3e391 100644 --- a/backend/src/workers/mod.rs +++ b/backend/src/workers/mod.rs @@ -15,8 +15,11 @@ use crate::{ pub mod analyze; pub mod index; pub mod ocr; +pub mod tenants; pub mod thumbnails; +use tenants::ProvisionTenantJob; + #[derive(Debug)] pub enum JobExecution { Success, @@ -146,6 +149,7 @@ pub fn default_handlers() -> Vec> { Arc::new(thumbnails::GenerateThumbnailsJob::new()), Arc::new(ocr::GenerateOcrTextJob::new()), Arc::new(index::IndexDocumentTextJob::new()), + Arc::new(ProvisionTenantJob::new()), ] } @@ -172,12 +176,8 @@ pub(crate) async fn fetch_version_object( s3_key: &str, limit_bytes: u64, ) -> Result, FetchVersionError> { - check_worker_document_limit(version.size_bytes, limit_bytes).map_err(|(size, limit)| { - FetchVersionError::TooLarge { - size, - limit, - } - })?; + check_worker_document_limit(version.size_bytes, limit_bytes) + .map_err(|(size, limit)| FetchVersionError::TooLarge { size, limit })?; storage .get_object(s3_key) @@ -199,9 +199,7 @@ pub(crate) fn handle_fetch_error( "document exceeds worker size limit" ); JobExecution::Failed { - error: format!( - "document size {size} bytes exceeds worker limit of {limit} bytes" - ), + error: format!("document size {size} bytes exceeds worker limit of {limit} bytes"), } } FetchVersionError::Storage(err) => { diff --git a/backend/src/workers/tenants.rs b/backend/src/workers/tenants.rs new file mode 100644 index 0000000..988fa16 --- /dev/null +++ b/backend/src/workers/tenants.rs @@ -0,0 +1,175 @@ +use std::sync::Arc; + +use async_trait::async_trait; +use chrono::Utc; +use diesel::prelude::*; +use reqwest::Client; +use serde::Deserialize; +use tracing::warn; +use uuid::Uuid; + +use crate::documents::search::ensure_quickwit_index; +use crate::jobs::JOB_PROVISION_TENANT; +use crate::models::{NewUserMembership, TenantStatus}; +use crate::schema::{tenants, user_memberships}; +use crate::state::AppState; +use crate::tenants::TenantRepository; +use crate::workers::{JobExecution, JobHandler}; + +pub struct ProvisionTenantJob; + +impl ProvisionTenantJob { + pub fn new() -> Self { + Self + } +} + +#[async_trait] +impl JobHandler for ProvisionTenantJob { + fn job_type(&self) -> &'static str { + JOB_PROVISION_TENANT + } + + async fn handle( + &self, + state: Arc, + job: crate::models::Job, + _storage: crate::storage::TenantStorage, + ) -> JobExecution { + let mut conn = match state.db_unscoped() { + Ok(conn) => conn, + Err(err) => { + warn!(job_id = %job.id, error = ?err, "failed to get connection for tenant provisioning"); + return JobExecution::Retry { + delay: std::time::Duration::from_secs(30), + error: "database connection unavailable".into(), + }; + } + }; + + let tenant = match TenantRepository::get_by_id(&mut conn, job.tenant_id) { + Ok(tenant) => tenant, + Err(err) => { + warn!(job_id = %job.id, error = ?err, "tenant not found for provisioning"); + return JobExecution::Failed { + error: "tenant not found".into(), + }; + } + }; + + if tenant.status == TenantStatus::Active { + warn!( + job_id = %job.id, + tenant_id = %tenant.id, + "tenant already active; skipping provisioning" + ); + return JobExecution::Success; + } + + if tenant.status != TenantStatus::Creating { + warn!( + job_id = %job.id, + tenant_id = %tenant.id, + status = %tenant.status.as_str(), + "tenant not in creating state; skipping provisioning" + ); + return JobExecution::Failed { + error: format!( + "tenant status '{}' not eligible for provisioning", + tenant.status.as_str() + ), + }; + } + let endpoint = match &state.config.quickwit_endpoint { + Some(endpoint) => endpoint.trim_end_matches('/').to_owned(), + None => { + warn!( + job_id = %job.id, + tenant_id = %tenant.id, + "quickwit endpoint not configured; retrying" + ); + return JobExecution::Retry { + delay: std::time::Duration::from_secs(30), + error: "quickwit endpoint not configured".into(), + }; + } + }; + + let index_id = tenant + .quickwit_index + .as_deref() + .map(str::to_owned) + .unwrap_or_else(|| format!("documents-{}", tenant.id)); + + let client = Client::new(); + if let Err(err) = ensure_quickwit_index(&client, &endpoint, &index_id).await { + warn!( + job_id = %job.id, + tenant_id = %tenant.id, + error = %err, + "failed to ensure quickwit index" + ); + return JobExecution::Retry { + delay: std::time::Duration::from_secs(30), + error: err.to_string(), + }; + } + + if let Some(members) = ProvisionPayload::from_job(&job) { + for member in members { + let new_membership = NewUserMembership { + id: Uuid::new_v4(), + user_id: member, + tenant_id: tenant.id, + role: "admin".to_string(), + }; + + if let Err(err) = diesel::insert_into(user_memberships::table) + .values(&new_membership) + .on_conflict((user_memberships::user_id, user_memberships::tenant_id)) + .do_nothing() + .execute(&mut conn) + { + warn!( + job_id = %job.id, + tenant_id = %tenant.id, + user_id = %member, + error = %err, + "failed to assign initial membership" + ); + } + } + } + + if let Err(err) = diesel::update(tenants::table.find(tenant.id)) + .set(( + tenants::status.eq(TenantStatus::Active), + tenants::quickwit_index.eq(Some(index_id)), + tenants::updated_at.eq(Utc::now().naive_utc()), + )) + .execute(&mut conn) + { + warn!(job_id = %job.id, error = %err, "failed to activate tenant"); + return JobExecution::Retry { + delay: std::time::Duration::from_secs(30), + error: format!("failed to update tenant status: {err}"), + }; + } + + JobExecution::Success + } +} + +#[derive(Deserialize, Default)] +struct ProvisionPayload { + #[serde(default)] + members: Vec, +} + +impl ProvisionPayload { + fn from_job(job: &crate::models::Job) -> Option> { + serde_json::from_value(job.payload.clone()) + .map(|payload: ProvisionPayload| payload.members) + .ok() + } +} diff --git a/backend/src/workers/thumbnails.rs b/backend/src/workers/thumbnails.rs index e1b91c5..8ffab70 100644 --- a/backend/src/workers/thumbnails.rs +++ b/backend/src/workers/thumbnails.rs @@ -25,10 +25,7 @@ use crate::{ }; use super::{ - analyze::determine_thumbnail_support, - fetch_version_object, - handle_fetch_error, - JobExecution, + analyze::determine_thumbnail_support, fetch_version_object, handle_fetch_error, JobExecution, JobHandler, }; diff --git a/backend/tests/auth_flow.rs b/backend/tests/auth_flow.rs index 5f58c8b..f87cac5 100644 --- a/backend/tests/auth_flow.rs +++ b/backend/tests/auth_flow.rs @@ -3,8 +3,9 @@ mod common; use anyhow::{anyhow, Context, Result}; use axum::body::Body; use axum::http::{header::SET_COOKIE, StatusCode}; -use backend::models::NewUserMembership; -use backend::schema::{tenants, user_memberships}; +use backend::jobs::JOB_PROVISION_TENANT; +use backend::models::{Job, NewUserMembership, TenantStatus}; +use backend::schema::{jobs, tenants, user_memberships, users}; use common::{acquire_db_lock, body_to_vec, TestApp}; use diesel::prelude::*; use serde::Deserialize; @@ -81,6 +82,77 @@ async fn login_rejects_unknown_user() -> Result<()> { Ok(()) } +#[tokio::test] +async fn signup_creates_user_tenant_and_membership() -> Result<()> { + let _lock = acquire_db_lock().await; + let app = TestApp::new().await?; + + let payload = json!({ + "username": "signup-user", + "password": "super-secret", + }); + + let response = app.post_json("/api/auth/signup", &payload, None).await?; + assert_eq!(response.status(), StatusCode::OK); + let body = body_to_vec(response.into_body()).await?; + let login: LoginResponse = serde_json::from_slice(&body)?; + + // session works immediately + let me = app.get("/api/auth/me", Some(&login.access_token)).await?; + assert_eq!(me.status(), StatusCode::OK); + + app.with_conn(|conn| { + let user: backend::models::User = users::table + .filter(users::username.eq("signup-user")) + .first(conn)?; + + let tenant: backend::models::Tenant = tenants::table + .filter(tenants::slug.eq("signup-user")) + .first(conn)?; + + assert_eq!(tenant.status, TenantStatus::Creating); + assert_eq!(tenant.created_by, Some(user.id)); + + let membership_exists: bool = diesel::select(diesel::dsl::exists( + user_memberships::table + .filter(user_memberships::user_id.eq(user.id)) + .filter(user_memberships::tenant_id.eq(tenant.id)), + )) + .get_result(conn)?; + assert!( + !membership_exists, + "membership should be enqueued, not created synchronously" + ); + + let job: Job = jobs::table + .filter(jobs::tenant_id.eq(tenant.id)) + .filter(jobs::job_type.eq(JOB_PROVISION_TENANT)) + .first(conn) + .context("provision job missing")?; + + let members = job + .payload + .get("members") + .and_then(|value| value.as_array()) + .cloned() + .unwrap_or_default(); + + assert!(members.iter().any(|value| { + value + .as_str() + .and_then(|id| Uuid::parse_str(id).ok()) + .map(|parsed| parsed == user.id) + .unwrap_or(false) + })); + + Ok(()) + }) + .await?; + + app.cleanup().await?; + Ok(()) +} + #[tokio::test] async fn login_rejects_invalid_password() -> Result<()> { let _lock = acquire_db_lock().await; @@ -198,6 +270,7 @@ async fn login_returns_tenant_selection_when_multiple_memberships() -> Result<() .values(( tenants::id.eq(secondary_id), tenants::slug.eq(&slug_for_insert), + tenants::status.eq(TenantStatus::Active), )) .execute(conn)?; diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 68c4da7..77fb72a 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -11,7 +11,7 @@ use axum::Router; use backend::auth::jwt::JwtService; use backend::config::AppConfig; use backend::db::{self, PgPool}; -use backend::models::{Job, NewUser, NewUserMembership, Tenant}; +use backend::models::{Job, NewUser, NewUserMembership, Tenant, TenantStatus}; use backend::routes; use backend::state::AppState; use backend::storage::ObjectStorage; @@ -278,6 +278,7 @@ impl TestApp { tenants_dsl::slug.eq(&slug_value), tenants_dsl::storage_root.eq(Some(root)), tenants_dsl::quickwit_index.eq(quickwit_value), + tenants_dsl::status.eq(TenantStatus::Active), )) .execute(conn) .context("failed to insert default tenant")?; diff --git a/backend/tests/tags_flow.rs b/backend/tests/tags_flow.rs index 49992bd..e1ba0be 100644 --- a/backend/tests/tags_flow.rs +++ b/backend/tests/tags_flow.rs @@ -2,7 +2,7 @@ mod common; use anyhow::Result; use axum::http::StatusCode; -use backend::models::{NewUser, NewUserMembership, Tag}; +use backend::models::{NewUser, NewUserMembership, Tag, TenantStatus}; use backend::schema::{ tags::dsl as tags_dsl, tenants::dsl as tenants_dsl, user_memberships::dsl as memberships_dsl, users::dsl as users_dsl, @@ -213,6 +213,7 @@ async fn tags_are_isolated_between_tenants() -> Result<()> { tenants_dsl::id.eq(tenant_b_id), tenants_dsl::slug.eq("tenant-b"), tenants_dsl::storage_root.eq(Some(storage_root)), + tenants_dsl::status.eq(TenantStatus::Active), )) .execute(conn)?;