From b5dea58ed444fc9f22c5f55eab6665c5f22ab51f Mon Sep 17 00:00:00 2001 From: Ashwin Kumar Sivakumar Date: Tue, 21 Jul 2026 01:39:25 +0530 Subject: [PATCH] fix: customer document submission and profile verification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit customer_profiles already had custom_data but never got a `status` column, so — same root cause already fixed for job_seeker_profiles — the generic profile save/get/submit handlers failed outright for the CUSTOMER role, and the admin final-approval path couldn't write a verified status either. - Migration: add customer_profiles.status (default 'DRAFT'). - Special-case CUSTOMER in the generic profile.rs handlers (get_profile, save_profile, fetch_saved_profile, set_profile_status), mirroring the existing JOB_SEEKER special-case, storing basic-tab fields under custom_data.basic_info. - Re-enable the CUSTOMER branch in activate_profile_after_final_approval now that the status column exists. - Add the missing POST /api/customers/profile/documents upload endpoint — the frontend's document upload (required: Aadhar/Government ID) targets this exact path for CUSTOMER and previously 404'd since no such route was ever registered. Mirrors the B2-upload-only pattern used by the profession apps' shared upload_document handler. Verified separately: requirement posting (POST /api/customers/requirements) and requirement submission-for-verification (POST /api/customers/requirements/:id/submit) already work correctly — both use the `leads` table (an active migration, despite the model's "Requirement" naming) and properly create a verification record (case_type REQUIREMENT_APPROVAL) that lands in admin Verification Management via the existing approve_requirement/reject_requirement handlers. No changes needed there. --- Cargo.lock | 2 + apps/customers/Cargo.toml | 4 +- apps/customers/src/handlers.rs | 69 ++++++++++++++++++- apps/customers/src/main.rs | 4 +- apps/users/src/handlers/approvals.rs | 7 +- apps/users/src/handlers/profile.rs | 43 +++++++----- ...21020000_customer_profiles_status.down.sql | 1 + ...0721020000_customer_profiles_status.up.sql | 7 ++ 8 files changed, 109 insertions(+), 28 deletions(-) create mode 100644 crates/db/migrations/20260721020000_customer_profiles_status.down.sql create mode 100644 crates/db/migrations/20260721020000_customer_profiles_status.up.sql diff --git a/Cargo.lock b/Cargo.lock index fdc26ab..d9dbc69 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1143,6 +1143,7 @@ version = "0.1.0" dependencies = [ "auth", "axum", + "bytes", "chrono", "contracts", "db", @@ -1150,6 +1151,7 @@ dependencies = [ "serde", "serde_json", "sqlx", + "storage", "tokio", "tracing", "tracing-subscriber", diff --git a/apps/customers/Cargo.toml b/apps/customers/Cargo.toml index 578fdcf..c129315 100644 --- a/apps/customers/Cargo.toml +++ b/apps/customers/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -axum = { workspace = true } +axum = { workspace = true, features = ["multipart"] } tokio = { workspace = true } serde = { workspace = true } sqlx = { workspace = true } @@ -12,9 +12,11 @@ tracing = { workspace = true } tracing-subscriber = { workspace = true } uuid = { workspace = true } chrono = { workspace = true } +bytes = { workspace = true } db = { path = "../../crates/db" } auth = { path = "../../crates/auth" } contracts = { path = "../../crates/contracts" } +storage = { path = "../../crates/storage" } serde_json = { workspace = true } email = { path = "../../crates/email" } diff --git a/apps/customers/src/handlers.rs b/apps/customers/src/handlers.rs index c02f7ac..4ed7806 100644 --- a/apps/customers/src/handlers.rs +++ b/apps/customers/src/handlers.rs @@ -1,10 +1,11 @@ use axum::{ - extract::{Path, Query, State}, + extract::{Multipart, Path, Query, State}, http::StatusCode, response::IntoResponse, routing::{get, post}, Json, Router, }; +use bytes::BufMut; use serde::Deserialize; use uuid::Uuid; use db::models::customer::{CustomerRepository, UpsertCustomerProfilePayload}; @@ -20,6 +21,7 @@ pub fn router() -> Router { Router::new() .route("/profile/me", get(get_profile).patch(update_profile)) .route("/profile/submit", post(submit_for_verification)) + .route("/profile/documents", post(upload_document)) .route("/requirements", get(list_requirements).post(create_requirement)) .route("/requirements/{id}", get(get_requirement).patch(update_requirement)) .route("/requirements/{id}/submit", post(submit_requirement)) @@ -74,6 +76,71 @@ async fn update_profile( } } +/// POST /api/customers/profile/documents — uploads a KYC document (e.g. Aadhar) +/// to B2 storage and returns its URL. The caller merges the URL into +/// profile_data and persists it via PATCH /api/profile, same as every other +/// role's document flow (see extract_documents in apps/users/src/handlers/profile.rs). +async fn upload_document( + State(state): State, + _auth: AuthUser, + mut multipart: Multipart, +) -> impl IntoResponse { + let mut file_bytes = bytes::BytesMut::new(); + let mut content_type = "application/octet-stream".to_string(); + let mut ext = "bin".to_string(); + let mut found = false; + + while let Ok(Some(field)) = multipart.next_field().await { + let name = field.name().unwrap_or("").to_string(); + if name == "document" || name == "file" || !found { + if let Some(ct) = field.content_type() { + content_type = ct.to_string(); + ext = match ct { + "image/jpeg" => "jpg", + "image/png" => "png", + "image/webp" => "webp", + "application/pdf" => "pdf", + _ => "bin", + } + .to_string(); + } else if let Some(fname) = field.file_name() { + if let Some(e) = fname.rsplit('.').next() { + ext = e.to_lowercase(); + } + } + + let data = match field.bytes().await { + Ok(b) => b, + Err(e) => return (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": format!("Failed to read file: {}", e) }))).into_response(), + }; + + if data.is_empty() { + return (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": "Empty file" }))).into_response(); + } + + if data.len() > 10 * 1024 * 1024 { + return (StatusCode::PAYLOAD_TOO_LARGE, Json(serde_json::json!({ "error": "File too large. Maximum 10 MB." }))).into_response(); + } + + file_bytes.put(data); + found = true; + break; + } + } + + if !found || file_bytes.is_empty() { + return (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": "No document file provided. Send a multipart field named 'document'." }))).into_response(); + } + + match state.storage.upload("documents", &ext, file_bytes.freeze(), &content_type).await { + Ok(url) => (StatusCode::OK, Json(serde_json::json!({ "url": url }))).into_response(), + Err(e) => { + tracing::error!("B2 upload failed: {:?}", e); + (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "File upload failed" }))).into_response() + } + } +} + async fn submit_for_verification( State(state): State, auth: AuthUser, diff --git a/apps/customers/src/main.rs b/apps/customers/src/main.rs index ec6ccaa..b34eaa1 100644 --- a/apps/customers/src/main.rs +++ b/apps/customers/src/main.rs @@ -13,6 +13,7 @@ use sqlx::PgPool; pub struct AppState { pub pool: PgPool, pub mail: Arc, + pub storage: Arc, } #[tokio::main] @@ -34,7 +35,8 @@ async fn main() { tracing::info!("Customers service — connected to database"); let mailer = Arc::new(email::Mailer::new()); - let state = AppState { pool, mail: mailer }; + let storage = Arc::new(storage::StorageClient::from_env().await); + let state = AppState { pool, mail: mailer, storage }; let app = Router::new() .nest("/api/customers", handlers::router()) diff --git a/apps/users/src/handlers/approvals.rs b/apps/users/src/handlers/approvals.rs index 1c1c3c2..1b78d23 100644 --- a/apps/users/src/handlers/approvals.rs +++ b/apps/users/src/handlers/approvals.rs @@ -189,12 +189,7 @@ async fn activate_profile_after_final_approval( // by its own id with a separate user_role_profile_id FK column — // matching those against `id` (as this previously did) never hit any // row, silently no-opping the approval. - // NOTE: customer_profiles has the same user_id-keyed shape but is left - // out here — its status column doesn't exist in the DB yet (blocked on - // the still-disabled requirements/leads migration), a separate, - // pre-existing issue in the customer/requirements vertical, out of scope - // for this fix. - if role_key == "COMPANY" || role_key == "JOB_SEEKER" { + if role_key == "COMPANY" || role_key == "JOB_SEEKER" || role_key == "CUSTOMER" { let query = format!( "UPDATE {} SET status = 'APPROVED', updated_at = NOW() WHERE user_id = $1", table diff --git a/apps/users/src/handlers/profile.rs b/apps/users/src/handlers/profile.rs index 3bdb710..5ede67a 100644 --- a/apps/users/src/handlers/profile.rs +++ b/apps/users/src/handlers/profile.rs @@ -144,10 +144,10 @@ async fn get_profile( }; } - if role_key == "JOB_SEEKER" { - return match sqlx::query( - r#"SELECT custom_data, status FROM job_seeker_profiles WHERE user_id = $1"#, - ) + if role_key == "JOB_SEEKER" || role_key == "CUSTOMER" { + let table = if role_key == "JOB_SEEKER" { "job_seeker_profiles" } else { "customer_profiles" }; + let query = format!(r#"SELECT custom_data, status FROM {} WHERE user_id = $1"#, table); + return match sqlx::query(&query) .bind(auth.user_id) .fetch_optional(&state.pool) .await @@ -360,13 +360,15 @@ async fn save_profile( }; } - if role_key == "JOB_SEEKER" { - // job_seeker_profiles also stores the job_seeker_portfolio blob (written - // by the dedicated /api/jobseeker/profile/me endpoint) inside custom_data, - // so basic-tab fields are merged in under a nested "basic_info" key - // instead of overwriting custom_data wholesale. + if role_key == "JOB_SEEKER" || role_key == "CUSTOMER" { + let table = if role_key == "JOB_SEEKER" { "job_seeker_profiles" } else { "customer_profiles" }; + + // Both tables also store other blobs inside custom_data (job_seeker_portfolio + // for JOB_SEEKER, written by the dedicated /api/jobseeker/profile/me endpoint), + // so basic-tab fields are merged in under a nested "basic_info" key instead of + // overwriting custom_data wholesale. let existing_custom_data: serde_json::Value = sqlx::query_scalar( - r#"SELECT custom_data FROM job_seeker_profiles WHERE user_id = $1"#, + &format!(r#"SELECT custom_data FROM {} WHERE user_id = $1"#, table), ) .bind(auth.user_id) .fetch_optional(&state.pool) @@ -382,15 +384,16 @@ async fn save_profile( merged.insert("basic_info".to_string(), input.profile_data.clone()); let merged_custom_data = serde_json::Value::Object(merged); - return match sqlx::query( + let query = format!( r#" - INSERT INTO job_seeker_profiles (user_id, custom_data, status, updated_at) + INSERT INTO {table} (user_id, custom_data, status, updated_at) VALUES ($1, $2, 'DRAFT', NOW()) ON CONFLICT (user_id) DO UPDATE SET custom_data = EXCLUDED.custom_data, updated_at = NOW() - "#, - ) + "# + ); + return match sqlx::query(&query) .bind(auth.user_id) .bind(&merged_custom_data) .execute(&state.pool) @@ -402,7 +405,7 @@ async fn save_profile( ) .into_response(), Err(e) => { - tracing::error!("save_profile(JOB_SEEKER) failed for user {}: {}", auth.user_id, e); + tracing::error!("save_profile({}) failed for user {}: {}", role_key, auth.user_id, e); (StatusCode::INTERNAL_SERVER_ERROR, format!("Database error: {}", e)).into_response() } }; @@ -679,9 +682,10 @@ async fn fetch_saved_profile( }; } - if role_key == "JOB_SEEKER" { + if role_key == "JOB_SEEKER" || role_key == "CUSTOMER" { + let table = if role_key == "JOB_SEEKER" { "job_seeker_profiles" } else { "customer_profiles" }; let custom_data: serde_json::Value = sqlx::query_scalar( - r#"SELECT custom_data FROM job_seeker_profiles WHERE user_id = $1"#, + &format!(r#"SELECT custom_data FROM {} WHERE user_id = $1"#, table), ) .bind(user_id) .fetch_optional(&state.pool) @@ -716,9 +720,10 @@ async fn set_profile_status(state: &AppState, user_id: Uuid, role_key: &str, sta return; } - if role_key == "JOB_SEEKER" { + if role_key == "JOB_SEEKER" || role_key == "CUSTOMER" { + let table = if role_key == "JOB_SEEKER" { "job_seeker_profiles" } else { "customer_profiles" }; sqlx::query( - r#"UPDATE job_seeker_profiles SET status = $1, updated_at = NOW() WHERE user_id = $2"#, + &format!(r#"UPDATE {} SET status = $1, updated_at = NOW() WHERE user_id = $2"#, table), ) .bind(status) .bind(user_id) diff --git a/crates/db/migrations/20260721020000_customer_profiles_status.down.sql b/crates/db/migrations/20260721020000_customer_profiles_status.down.sql new file mode 100644 index 0000000..8a7833b --- /dev/null +++ b/crates/db/migrations/20260721020000_customer_profiles_status.down.sql @@ -0,0 +1 @@ +ALTER TABLE customer_profiles DROP COLUMN IF EXISTS status; diff --git a/crates/db/migrations/20260721020000_customer_profiles_status.up.sql b/crates/db/migrations/20260721020000_customer_profiles_status.up.sql new file mode 100644 index 0000000..35f108a --- /dev/null +++ b/crates/db/migrations/20260721020000_customer_profiles_status.up.sql @@ -0,0 +1,7 @@ +-- customer_profiles has always had custom_data (used for generic profile +-- save via apps/users/src/handlers/profile.rs) but never got a `status` +-- column, so verification-status tracking for the CUSTOMER role fails +-- outright with "column does not exist" — same root cause already fixed +-- for job_seeker_profiles. +ALTER TABLE customer_profiles + ADD COLUMN IF NOT EXISTS status VARCHAR(50) NOT NULL DEFAULT 'DRAFT';