backend signup

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,

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<dyn JobHandler>> {
         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<Vec<u8>, 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) => {

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<AppState>,
+        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<Uuid>,
+}
+
+impl ProvisionPayload {
+    fn from_job(job: &crate::models::Job) -> Option<Vec<Uuid>> {
+        serde_json::from_value(job.payload.clone())
+            .map(|payload: ProvisionPayload| payload.members)
+            .ok()
+    }
+}

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,
 };