diff --git a/apps/companies/src/handlers/mod.rs b/apps/companies/src/handlers/mod.rs index 8f58449..99e3830 100644 --- a/apps/companies/src/handlers/mod.rs +++ b/apps/companies/src/handlers/mod.rs @@ -23,6 +23,7 @@ pub fn router() -> Router { .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, + 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 = None; + let mut uploaded_files: Vec<(String, i64, String)> = Vec::new(); // (url, data_len, content_type) + let mut documents_meta: Vec = 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 = 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() +}