feat(companies): add /profile/submit-with-documents endpoint with local-disk fallback for B2 outages

apps/companies/src/handlers/mod.rs

@@ -23,6 +23,7 @@ pub fn router() -> Router<AppState> {
         .route("/profile/me", get(get_profile).patch(update_profile))
         .route("/profile/documents", post(upload_documents))
         .route("/profile/submit", post(submit_for_verification))
+        .route("/profile/submit-with-documents", post(submit_with_documents))
         .route("/jobs", get(list_jobs).post(create_job))
         .route("/jobs/{id}", get(get_job).patch(update_job))
         .route("/jobs/{id}/submit", post(submit_job))
@@ -746,3 +747,401 @@ async fn view_contact(
         }
     }
 }
+
+/// Local-disk fallback directory for company documents when B2 is down.
+const COMPANY_DOCS_LOCAL_DIR: &str = "/var/lib/nxtgauge-uploads/company_documents";
+
+/// POST /api/companies/profile/submit-with-documents
+///
+/// Accepts multipart/form-data with:
+///   * `profile`   — JSON string with company fields
+///   * `documents` — one or more file parts
+///
+/// In one request this handler:
+///   1. Saves the profile into `company_profiles` (DRAFT upsert, same SQL as users service).
+///   2. Uploads each document, trying B2 first then falling back to local disk
+///      (`/var/lib/nxtgauge-uploads/company_documents/{uuid}.{ext}`) and inserting
+///      a `company_documents` row regardless of source.
+///   3. Creates a `verifications` row (case_type = "PROFILE_VERIFICATION", priority = "MEDIUM").
+///   4. Updates `company_profiles.status` to 'PENDING'.
+///   5. Returns `{ verification_id, status, documents_uploaded, documents_failed, storage_backend }`.
+async fn submit_with_documents(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    mut multipart: Multipart,
+) -> impl IntoResponse {
+    // ---- 1. Parse multipart: pull `profile` JSON first, then read each `documents` file
+    //         in-place (we can't hold a Field across loop iterations because Multipart is mutably
+    //         borrowed by next_field). We delay the company_documents DB insert until step 2b
+    //         because the company_profiles.id FK isn't available until the profile upsert runs. ----
+    let mut profile_json_str: Option<String> = None;
+    let mut uploaded_files: Vec<(String, i64, String)> = Vec::new(); // (url, data_len, content_type)
+    let mut documents_meta: Vec<serde_json::Value> = Vec::new();
+    let mut documents_uploaded: usize = 0;
+    let mut documents_failed: usize = 0;
+    let mut storage_backend: &str = "b2";
+    let mut failure_details: Vec<serde_json::Value> = Vec::new();
+
+    while let Ok(Some(field)) = multipart.next_field().await {
+        let name = field.name().unwrap_or("").to_string();
+        if name == "profile" {
+            match field.text().await {
+                Ok(t) => profile_json_str = Some(t),
+                Err(e) => {
+                    return (
+                        StatusCode::BAD_REQUEST,
+                        Json(serde_json::json!({ "error": format!("Failed to read profile field: {}", e) })),
+                    )
+                        .into_response();
+                }
+            }
+        } else if name == "documents" || name == "files" || name == "file" {
+            // ---- 2a. Upload to storage (B2 with local-disk fallback). ----
+            let original_filename = field.file_name().unwrap_or("").to_string();
+            let content_type = field
+                .content_type()
+                .unwrap_or("application/octet-stream")
+                .to_string();
+
+            let ext = if !original_filename.is_empty() {
+                original_filename
+                    .rsplit('.')
+                    .next()
+                    .unwrap_or("bin")
+                    .to_lowercase()
+            } else {
+                match content_type.as_str() {
+                    "application/pdf" => "pdf".to_string(),
+                    "image/jpeg" => "jpg".to_string(),
+                    "image/png" => "png".to_string(),
+                    _ => "bin".to_string(),
+                }
+            };
+
+            let data = match field.bytes().await {
+                Ok(b) => b,
+                Err(e) => {
+                    documents_failed += 1;
+                    failure_details.push(serde_json::json!({
+                        "filename": original_filename,
+                        "error": format!("Failed to read file bytes: {}", e),
+                    }));
+                    continue;
+                }
+            };
+
+            if data.is_empty() {
+                // silently skip empty files (matches existing upload_documents behaviour)
+                continue;
+            }
+
+            if data.len() > 10 * 1024 * 1024 {
+                documents_failed += 1;
+                failure_details.push(serde_json::json!({
+                    "filename": original_filename,
+                    "error": "File too large. Maximum 10 MB per file.",
+                }));
+                continue;
+            }
+
+            let data_len = data.len() as i64;
+
+            // Try B2 first.
+            let (url, source) = match state
+                .storage
+                .upload("company_documents", &ext, data.clone(), &content_type)
+                .await
+            {
+                Ok(u) => (u, "b2"),
+                Err(b2_err) => {
+                    tracing::warn!(
+                        "B2 upload failed for company user {} ({}) — falling back to local disk: {}",
+                        auth.user_id,
+                        original_filename,
+                        b2_err
+                    );
+                    // Fallback: write to local disk.
+                    let dir = std::path::Path::new(COMPANY_DOCS_LOCAL_DIR);
+                    if let Err(mkdir_err) = tokio::fs::create_dir_all(dir).await {
+                        tracing::error!(
+                            "Local-disk fallback unavailable: failed to create {}: {}",
+                            COMPANY_DOCS_LOCAL_DIR,
+                            mkdir_err
+                        );
+                        documents_failed += 1;
+                        failure_details.push(serde_json::json!({
+                            "filename": original_filename,
+                            "error": format!("B2 failed and local fallback mkdir failed: {}", mkdir_err),
+                            "b2_error": b2_err.to_string(),
+                        }));
+                        continue;
+                    }
+                    let file_id = Uuid::new_v4();
+                    let file_path = dir.join(format!("{}.{}", file_id, ext));
+                    if let Err(write_err) = tokio::fs::write(&file_path, &data).await {
+                        tracing::error!(
+                            "Local-disk fallback write failed for {}: {}",
+                            file_path.display(),
+                            write_err
+                        );
+                        documents_failed += 1;
+                        failure_details.push(serde_json::json!({
+                            "filename": original_filename,
+                            "error": format!("B2 failed and local fallback write failed: {}", write_err),
+                            "b2_error": b2_err.to_string(),
+                        }));
+                        continue;
+                    }
+                    (format!("file://{}", file_path.display()), "local")
+                }
+            };
+
+            // Track storage backend for the response: once any file goes local, the request is "local".
+            if source == "local" {
+                storage_backend = "local";
+            }
+
+            // Derive a stable document_type from the original filename (stem) so the admin UI can group.
+            let document_type = if !original_filename.is_empty() {
+                std::path::Path::new(&original_filename)
+                    .file_stem()
+                    .and_then(|s| s.to_str())
+                    .unwrap_or("document")
+                    .to_string()
+            } else {
+                "document".to_string()
+            };
+
+            documents_meta.push(serde_json::json!({
+                "url": url,
+                "document_type": document_type,
+                "file_size": data_len,
+                "mime_type": content_type,
+            }));
+            uploaded_files.push((url, data_len, content_type));
+            documents_uploaded += 1;
+        }
+        // ignore other fields silently
+    }
+
+    let profile_str = match profile_json_str {
+        Some(s) => s,
+        None => {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(serde_json::json!({ "error": "Missing 'profile' multipart field (JSON string)." })),
+            )
+                .into_response();
+        }
+    };
+
+    let profile_obj: serde_json::Value = match serde_json::from_str(&profile_str) {
+        Ok(v) => v,
+        Err(e) => {
+            return (
+                StatusCode::BAD_REQUEST,
+                Json(serde_json::json!({ "error": format!("Invalid profile JSON: {}", e) })),
+            )
+                .into_response();
+        }
+    };
+
+    // ---- 2. Save profile to `company_profiles` (DRAFT upsert). ----
+    // SQL mapping copied verbatim from apps/users/src/handlers/profile.rs:236-335 (COMPANY branch).
+    let name = profile_obj
+        .get("company_name")
+        .and_then(|v| v.as_str())
+        .unwrap_or_default()
+        .to_string();
+    let email = profile_obj
+        .get("company_email")
+        .and_then(|v| v.as_str())
+        .unwrap_or_default()
+        .to_string();
+    let phone = profile_obj
+        .get("company_phone")
+        .and_then(|v| v.as_str())
+        .unwrap_or_default()
+        .to_string();
+    let website = profile_obj
+        .get("website")
+        .and_then(|v| v.as_str())
+        .unwrap_or_default()
+        .to_string();
+    let city = profile_obj
+        .get("location")
+        .and_then(|v| v.as_str())
+        .unwrap_or_default()
+        .to_string();
+    let state_val = profile_obj
+        .get("state")
+        .and_then(|v| v.as_str())
+        .unwrap_or_default()
+        .to_string();
+    let postal = profile_obj
+        .get("pin_code")
+        .and_then(|v| v.as_str())
+        .unwrap_or_default()
+        .to_string();
+    let address = profile_obj
+        .get("address")
+        .and_then(|v| v.as_str())
+        .unwrap_or_default()
+        .to_string();
+    let gst = profile_obj
+        .get("gst_number")
+        .and_then(|v| v.as_str())
+        .unwrap_or_default()
+        .to_string();
+
+    if let Err(e) = sqlx::query(
+        r#"
+        INSERT INTO company_profiles (
+            user_id, company_name, contact_email, contact_phone, website_url,
+            address_line1, city, state, postal_code, gst_number, status, updated_at
+        )
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, 'DRAFT', NOW())
+        ON CONFLICT (user_id) DO UPDATE SET
+            company_name = EXCLUDED.company_name,
+            contact_email = EXCLUDED.contact_email,
+            contact_phone = EXCLUDED.contact_phone,
+            website_url = EXCLUDED.website_url,
+            address_line1 = EXCLUDED.address_line1,
+            city = EXCLUDED.city,
+            state = EXCLUDED.state,
+            postal_code = EXCLUDED.postal_code,
+            gst_number = EXCLUDED.gst_number,
+            updated_at = NOW()
+        "#,
+    )
+    .bind(auth.user_id)
+    .bind(&name)
+    .bind(&email)
+    .bind(&phone)
+    .bind(&website)
+    .bind(&address)
+    .bind(&city)
+    .bind(&state_val)
+    .bind(&postal)
+    .bind(&gst)
+    .execute(&state.pool)
+    .await
+    {
+        tracing::error!("submit_with_documents: profile upsert failed for user {}: {}", auth.user_id, e);
+        return (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(serde_json::json!({ "error": format!("Failed to save profile: {}", e) })),
+        )
+            .into_response();
+    }
+
+    // Look up the company_profiles.id for the company_documents FK.
+    let company_id: Uuid = match sqlx::query_scalar::<_, Uuid>(
+        "SELECT id FROM company_profiles WHERE user_id = $1",
+    )
+    .bind(auth.user_id)
+    .fetch_one(&state.pool)
+    .await
+    {
+        Ok(id) => id,
+        Err(e) => {
+            tracing::error!("submit_with_documents: cannot resolve company_id: {}", e);
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({ "error": "Profile saved but company record lookup failed" })),
+            )
+                .into_response();
+        }
+    };
+
+    // ---- 2b. Insert company_documents rows (deferred until we have company_id). ----
+    // Same SQL as the existing upload_documents handler (apps/companies/src/handlers/mod.rs:612-624).
+    // Errors are logged but not propagated — matches existing behaviour.
+    for (url, data_len, content_type) in &uploaded_files {
+        if let Err(e) = sqlx::query(
+            r#"
+            INSERT INTO company_documents (company_id, document_name, document_url, file_size, mime_type)
+            VALUES ($1, $2, $3, $4, $5)
+            "#,
+        )
+        .bind(company_id)
+        .bind(format!("document_{}", Uuid::new_v4()))
+        .bind(url)
+        .bind(*data_len)
+        .bind(content_type)
+        .execute(&state.pool)
+        .await
+        {
+            tracing::error!(
+                "Failed to save company_documents row for company {}: {}",
+                company_id,
+                e
+            );
+        }
+    }
+
+    // ---- 4. Create the verification record. ----
+    // Signature: VerificationRepository::create(pool, user_id, role_key, case_type, priority, payload, documents)
+    // (see crates/db/src/models/verification.rs:38). The DB column is `case_type`; the user service
+    // populates it with "PROFILE_VERIFICATION" — we do the same.
+    let verification = match VerificationRepository::create(
+        &state.pool,
+        auth.user_id,
+        "COMPANY",
+        "PROFILE_VERIFICATION",
+        "MEDIUM",
+        profile_obj.clone(),
+        serde_json::Value::Array(documents_meta.clone()),
+    )
+    .await
+    {
+        Ok(v) => v,
+        Err(e) => {
+            tracing::error!(
+                "submit_with_documents: failed to create verification for user {}: {}",
+                auth.user_id,
+                e
+            );
+            return (
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({ "error": format!("Failed to create verification record: {}", e) })),
+            )
+                .into_response();
+        }
+    };
+
+    // ---- 5. Mark the company_profiles row as PENDING. ----
+    // Mirrors `set_profile_status` (COMPANY branch) in apps/users/src/handlers/profile.rs:606-614.
+    if let Err(e) = sqlx::query(
+        "UPDATE company_profiles SET status = 'PENDING', updated_at = NOW() WHERE user_id = $1",
+    )
+    .bind(auth.user_id)
+    .execute(&state.pool)
+    .await
+    {
+        tracing::error!(
+            "submit_with_documents: failed to update company_profiles.status for user {}: {}",
+            auth.user_id,
+            e
+        );
+        // Don't fail the whole request — the verification record is the source of truth for the admin queue.
+    }
+
+    // Invalidate cached profile so the next read reflects the new status.
+    let cache_key = format!("profile:company:{}", auth.user_id);
+    let mut redis = state.redis.clone();
+    let _ = redis.del::<_, ()>(&cache_key).await;
+
+    let mut response = serde_json::json!({
+        "verification_id": verification.id,
+        "status": "PENDING",
+        "documents_uploaded": documents_uploaded,
+        "documents_failed": documents_failed,
+        "storage_backend": storage_backend,
+    });
+    if !failure_details.is_empty() {
+        response["document_errors"] = serde_json::Value::Array(failure_details);
+    }
+
+    (StatusCode::OK, Json(response)).into_response()
+}