fix: customer document submission and profile verification

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

@@ -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",

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" }
 

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<AppState> {
     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<AppState>,
+    _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<AppState>,
     auth: AuthUser,

apps/customers/src/main.rs

@@ -13,6 +13,7 @@ use sqlx::PgPool;
 pub struct AppState {
     pub pool: PgPool,
     pub mail: Arc<email::Mailer>,
+    pub storage: Arc<storage::StorageClient>,
 }
 
 #[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())

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
+    if role_key == "COMPANY" || role_key == "JOB_SEEKER" || role_key == "CUSTOMER" {
-    // 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" {
         let query = format!(
             "UPDATE {} SET status = 'APPROVED', updated_at = NOW() WHERE user_id = $1",
             table

apps/users/src/handlers/profile.rs

@@ -144,10 +144,10 @@ async fn get_profile(
         };
     }
 
-    if role_key == "JOB_SEEKER" {
+    if role_key == "JOB_SEEKER" || role_key == "CUSTOMER" {
-        return match sqlx::query(
+        let table = if role_key == "JOB_SEEKER" { "job_seeker_profiles" } else { "customer_profiles" };
-            r#"SELECT custom_data, status FROM job_seeker_profiles WHERE user_id = $1"#,
+        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" {
+    if role_key == "JOB_SEEKER" || role_key == "CUSTOMER" {
-        // job_seeker_profiles also stores the job_seeker_portfolio blob (written
+        let table = if role_key == "JOB_SEEKER" { "job_seeker_profiles" } else { "customer_profiles" };
-        // by the dedicated /api/jobseeker/profile/me endpoint) inside custom_data,
+
-        // so basic-tab fields are merged in under a nested "basic_info" key
+        // Both tables also store other blobs inside custom_data (job_seeker_portfolio
-        // instead of overwriting custom_data wholesale.
+        // 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)

crates/db/migrations/20260721020000_customer_profiles_status.down.sql

@@ -0,0 +1 @@
+ALTER TABLE customer_profiles DROP COLUMN IF EXISTS status;

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';