Update backend services: catering_services, companies, developers, gateway, job_seekers, photographers, social_media_managers, tutors, ugc_content_creators, users; update cache (otp, token), contracts (profession_shared, profession_state), db (job_seeker, verification), email; add revision-requested email template; update init-db.sql and start-services.sh

Cargo.lock

@@ -764,6 +764,7 @@ dependencies = [
  "db",
  "serde",
  "sqlx",
+ "storage",
  "tokio",
  "tracing",
  "tracing-subscriber",
@@ -847,13 +848,17 @@ version = "0.1.0"
 dependencies = [
  "auth",
  "axum",
+ "bytes",
+ "cache",
  "chrono",
  "contracts",
  "db",
  "email",
+ "redis",
  "serde",
  "serde_json",
  "sqlx",
+ "storage",
  "tokio",
  "tracing",
  "tracing-subscriber",
@@ -882,6 +887,7 @@ dependencies = [
  "anyhow",
  "async-trait",
  "axum",
+ "bytes",
  "cache",
  "chrono",
  "db",
@@ -889,6 +895,7 @@ dependencies = [
  "serde",
  "serde_json",
  "sqlx",
+ "storage",
  "tracing",
  "uuid",
 ]
@@ -1114,6 +1121,7 @@ dependencies = [
  "db",
  "serde",
  "sqlx",
+ "storage",
  "tokio",
  "tracing",
  "tracing-subscriber",
@@ -1562,6 +1570,7 @@ dependencies = [
  "db",
  "serde",
  "sqlx",
+ "storage",
  "tokio",
  "tracing",
  "tracing-subscriber",
@@ -2062,10 +2071,12 @@ dependencies = [
  "auth",
  "axum",
  "bytes",
+ "cache",
  "chrono",
  "contracts",
  "db",
  "email",
+ "redis",
  "serde",
  "serde_json",
  "sqlx",
@@ -2278,6 +2289,7 @@ dependencies = [
  "db",
  "serde",
  "sqlx",
+ "storage",
  "tokio",
  "tracing",
  "tracing-subscriber",
@@ -2627,6 +2639,7 @@ dependencies = [
  "db",
  "serde",
  "sqlx",
+ "storage",
  "tokio",
  "tracing",
  "tracing-subscriber",
@@ -3412,6 +3425,7 @@ dependencies = [
  "db",
  "serde",
  "sqlx",
+ "storage",
  "tokio",
  "tracing",
  "tracing-subscriber",
@@ -4077,6 +4091,7 @@ dependencies = [
  "db",
  "serde",
  "sqlx",
+ "storage",
  "tokio",
  "tracing",
  "tracing-subscriber",
@@ -4241,6 +4256,7 @@ dependencies = [
  "db",
  "serde",
  "sqlx",
+ "storage",
  "tokio",
  "tracing",
  "tracing-subscriber",

apps/catering_services/Cargo.toml

@@ -16,4 +16,5 @@ db = { path = "../../crates/db" }
 auth = { path = "../../crates/auth" }
 contracts = { path = "../../crates/contracts" }
 cache     = { path = "../../crates/cache" }
+storage = { path = "../../crates/storage" }
 

apps/catering_services/src/main.rs

@@ -3,6 +3,7 @@ mod admin;
 
 use axum::{routing::get, Router};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use contracts::ProfessionState;
 
@@ -30,7 +31,8 @@ async fn main() {
 
     tracing::info!("Catering Services service — connected to DB and Redis");
 
-    let state = ProfessionState { pool, redis };
+    let storage = Arc::new(storage::StorageClient::from_env().await);
+    let state = ProfessionState { pool, redis, storage };
 
     let app = Router::new()
         .nest("/api/catering-services", handlers::router())

apps/companies/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 }
@@ -17,4 +17,8 @@ auth = { path = "../../crates/auth" }
 contracts = { path = "../../crates/contracts" }
 serde_json = { workspace = true }
 email = { path = "../../crates/email" }
+storage = { path = "../../crates/storage" }
+bytes = { workspace = true }
+cache = { path = "../../crates/cache" }
+redis = { workspace = true }
 

apps/companies/src/handlers/mod.rs

@@ -1,11 +1,14 @@
 pub mod admin;
 use axum::{
-    extract::{Path, Query, State},
+    extract::{Multipart, Path, Query, State},
     http::StatusCode,
     response::IntoResponse,
     routing::{get, patch, post},
     Json, Router,
 };
+use bytes::BufMut;
+use cache::jobs as cache_jobs;
+use redis::AsyncCommands;
 use serde::Deserialize;
 use uuid::Uuid;
 use db::models::company::{CompanyRepository, UpsertCompanyProfilePayload};
@@ -19,6 +22,7 @@ use crate::AppState;
 pub fn router() -> Router<AppState> {
     Router::new()
         .route("/profile/me", get(get_profile).patch(update_profile))
+        .route("/profile/documents", post(upload_documents))
         .route("/profile/submit", post(submit_for_verification))
         .route("/jobs", get(list_jobs).post(create_job))
         .route("/jobs/{id}", get(get_job).patch(update_job))
@@ -58,8 +62,23 @@ async fn get_profile(
     State(state): State<AppState>,
     auth: AuthUser,
 ) -> impl IntoResponse {
+    let cache_key = format!("profile:company:{}", auth.user_id);
+    let mut redis = state.redis.clone();
+
+    // Try cache first
+    if let Ok(cached) = redis.get::<_, String>(&cache_key).await {
+        tracing::debug!("Cache hit for company profile: {}", auth.user_id);
+        if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&cached) {
+            return (StatusCode::OK, Json(parsed)).into_response();
+        }
+    }
+
     match CompanyRepository::get_by_user_id(&state.pool, auth.user_id).await {
-        Ok(Some(profile)) => (StatusCode::OK, Json(profile)).into_response(),
+        Ok(Some(profile)) => {
+            // Cache for 5 minutes
+            let _: Result<(), _> = redis.set_ex(&cache_key, &serde_json::to_string(&profile).unwrap_or_default(), 300).await;
+            (StatusCode::OK, Json(profile)).into_response()
+        }
         Ok(None) => (StatusCode::NOT_FOUND, "Company profile not found").into_response(),
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
@@ -71,7 +90,13 @@ async fn update_profile(
     Json(payload): Json<UpsertCompanyProfilePayload>,
 ) -> impl IntoResponse {
     match CompanyRepository::upsert(&state.pool, auth.user_id, payload).await {
-        Ok(profile) => (StatusCode::OK, Json(profile)).into_response(),
+        Ok(profile) => {
+            // Invalidate profile cache
+            let cache_key = format!("profile:company:{}", auth.user_id);
+            let mut redis = state.redis.clone();
+            let _ = redis.del::<_, ()>(&cache_key).await;
+            (StatusCode::OK, Json(profile)).into_response()
+        }
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
 }
@@ -99,10 +124,16 @@ async fn submit_for_verification(
     }
 
     match CompanyRepository::submit_for_verification(&state.pool, auth.user_id).await {
-        Ok(profile) => (StatusCode::OK, Json(serde_json::json!({
+        Ok(profile) => {
-            "status": profile.status,
+            // Invalidate company profile cache
-            "message": "Profile submitted for verification"
+            let cache_key = format!("profile:company:{}", auth.user_id);
-        }))).into_response(),
+            let mut redis = state.redis.clone();
+            let _ = redis.del::<_, ()>(&cache_key).await;
+            (StatusCode::OK, Json(serde_json::json!({
+                "status": profile.status,
+                "message": "Profile submitted for verification"
+            }))).into_response()
+        }
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
 }
@@ -119,11 +150,30 @@ async fn list_jobs(
 
     let page = q.page.unwrap_or(1);
     let limit = q.limit.unwrap_or(20);
+    let status_filter = q.status.as_deref().unwrap_or("");
+
+    // Build cache key
+    let cache_key = format!("jobs:company:{}:{}:{}:{}", company.id, page, limit, status_filter);
+    let mut redis = state.redis.clone();
+
+    // Try cache first
+    if let Ok(cached) = redis.get::<_, String>(&cache_key).await {
+        tracing::debug!("Cache hit for company jobs: {}", cache_key);
+        if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&cached) {
+            return (StatusCode::OK, Json(parsed)).into_response();
+        }
+    }
+
     match JobRepository::list_by_company_id(&state.pool, company.id, q.status, page, limit).await {
-        Ok(jobs) => (StatusCode::OK, Json(serde_json::json!({
+        Ok(jobs) => {
-            "data": jobs,
+            let response = serde_json::json!({
-            "pagination": { "page": page, "limit": limit }
+                "data": jobs,
-        }))).into_response(),
+                "pagination": { "page": page, "limit": limit }
+            });
+            // Cache for 5 minutes
+            let _: Result<(), _> = redis.set_ex(&cache_key, &serde_json::to_string(&response).unwrap_or_default(), 300).await;
+            (StatusCode::OK, Json(response)).into_response()
+        }
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
 }
@@ -190,7 +240,17 @@ async fn create_job(
     };
 
     match JobRepository::create(&state.pool, db_payload).await {
-        Ok(job) => (StatusCode::CREATED, Json(job)).into_response(),
+        Ok(job) => {
+            // Invalidate company's job list cache
+            let mut redis = state.redis.clone();
+            let pattern = format!("jobs:company:{}:*", company.id);
+            if let Ok(keys) = redis.keys::<_, Vec<String>>(pattern).await {
+                if !keys.is_empty() {
+                    let _ = redis.del::<_, ()>(keys).await;
+                }
+            }
+            (StatusCode::CREATED, Json(job)).into_response()
+        }
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
 }
@@ -229,7 +289,17 @@ async fn update_job(
     };
 
     match JobRepository::update(&state.pool, job.id, payload).await {
-        Ok(updated) => (StatusCode::OK, Json(updated)).into_response(),
+        Ok(updated) => {
+            // Invalidate company job list cache
+            let mut redis = state.redis.clone();
+            let pattern = format!("jobs:company:{}:*", company.id);
+            if let Ok(keys) = redis.keys::<_, Vec<String>>(pattern).await {
+                if !keys.is_empty() {
+                    let _ = redis.del::<_, ()>(keys).await;
+                }
+            }
+            (StatusCode::OK, Json(updated)).into_response()
+        }
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
 }
@@ -282,6 +352,14 @@ async fn submit_job(
                 serde_json::json!([]),
             )
             .await;
+            // Invalidate company job list cache
+            let mut redis = state.redis.clone();
+            let pattern = format!("jobs:company:{}:*", company.id);
+            if let Ok(keys) = redis.keys::<_, Vec<String>>(pattern).await {
+                if !keys.is_empty() {
+                    let _ = redis.del::<_, ()>(keys).await;
+                }
+            }
             (StatusCode::OK, Json(updated)).into_response()
         }
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
@@ -305,7 +383,17 @@ async fn close_job(
     };
 
     match JobRepository::update_status(&state.pool, job.id, "CLOSED").await {
-        Ok(updated) => (StatusCode::OK, Json(updated)).into_response(),
+        Ok(updated) => {
+            // Invalidate company job list cache
+            let mut redis = state.redis.clone();
+            let pattern = format!("jobs:company:{}:*", company.id);
+            if let Ok(keys) = redis.keys::<_, Vec<String>>(pattern).await {
+                if !keys.is_empty() {
+                    let _ = redis.del::<_, ()>(keys).await;
+                }
+            }
+            (StatusCode::OK, Json(updated)).into_response()
+        }
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
 }
@@ -366,14 +454,28 @@ async fn update_application_status(
     match ApplicationRepository::update_status(&state.pool, app.id, &payload.status).await {
         Ok(updated) => {
             // Notify applicant of status change (ignore failures)
-            let applicant_info = sqlx::query_as::<_, (String, String)>(
+            let applicant_info = sqlx::query_as::<_, (String, String, Uuid)>(
-                "SELECT CONCAT(u.first_name, ' ', u.last_name) AS name, u.email, u.phone FROM users u WHERE u.id = $1",
+                "SELECT CONCAT(u.first_name, ' ', u.last_name) AS name, u.email, u.id FROM users u WHERE u.id = $1",
             )
             .bind(app.applicant_user_id)
             .fetch_optional(&state.pool)
             .await;
-            if let Ok(Some((name, email))) = applicant_info {
+            if let Ok(Some((name, email, applicant_uuid))) = applicant_info {
                 let _ = state.mail.send_application_status_email(&email, &name, &job.title, &payload.status).await;
+
+                // Send in-app notification to job seeker
+                sqlx::query(
+                    r#"INSERT INTO notifications (user_id, title, body, type, reference_id)
+                       VALUES ($1, $2, $3, $4, $5)"#,
+                )
+                .bind(applicant_uuid)
+                .bind(format!("Application Status: {}", payload.status))
+                .bind(format!("Your application for '{}' has been {}.", job.title, payload.status.to_lowercase()))
+                .bind("APPLICATION")
+                .bind(app.id)
+                .execute(&state.pool)
+                .await
+                .ok();
             }
             (StatusCode::OK, Json(updated)).into_response()
         }
@@ -381,6 +483,96 @@ async fn update_application_status(
     }
 }
 
+async fn upload_documents(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    mut multipart: Multipart,
+) -> impl IntoResponse {
+    let company = match CompanyRepository::get_by_user_id(&state.pool, auth.user_id).await {
+        Ok(Some(c)) => c,
+        Ok(None) => return (StatusCode::NOT_FOUND, "Company profile not found").into_response(),
+        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
+    };
+
+    let mut uploaded_urls: Vec<String> = Vec::new();
+
+    while let Ok(Some(field)) = multipart.next_field().await {
+        let name = field.name().unwrap_or("").to_string();
+        if name != "documents" && name != "files" && name != "file" {
+            continue;
+        }
+
+        let content_type = field.content_type()
+            .unwrap_or("application/octet-stream")
+            .to_string();
+
+        let ext = if let Some(fname) = field.file_name() {
+            fname.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) => return (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": format!("Failed to read file: {}", e) }))).into_response(),
+        };
+
+        if data.is_empty() {
+            continue;
+        }
+
+        if data.len() > 10 * 1024 * 1024 {
+            return (StatusCode::PAYLOAD_TOO_LARGE, Json(serde_json::json!({ "error": "File too large. Maximum 10 MB per file." }))).into_response();
+        }
+
+        let data_len = data.len();
+        let url = match state.storage
+            .upload("company_documents", &ext, data, &content_type)
+            .await
+        {
+            Ok(u) => u,
+            Err(e) => {
+                tracing::error!("B2 upload failed for company {}: {}", company.id, e);
+                return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "File upload failed" }))).into_response();
+            }
+        };
+
+        // Persist document record
+        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 as i64)
+        .bind(&content_type)
+        .execute(&state.pool)
+        .await
+        {
+            tracing::error!("Failed to save document record for company {}: {}", company.id, e);
+        }
+
+        uploaded_urls.push(url);
+    }
+
+    if uploaded_urls.is_empty() {
+        return (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": "No valid document files provided. Send multipart fields named 'documents'." }))).into_response();
+    }
+
+    (StatusCode::OK, Json(serde_json::json!({
+        "documents": uploaded_urls,
+        "count": uploaded_urls.len()
+    }))).into_response()
+}
+
 async fn view_contact(
     State(state): State<AppState>,
     Path(id): Path<Uuid>,

apps/companies/src/main.rs

@@ -1,6 +1,7 @@
 mod handlers;
 
 use axum::{routing::get, Router};
+use cache::RedisPool;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@@ -9,7 +10,9 @@ use sqlx::PgPool;
 #[derive(Clone)]
 pub struct AppState {
     pub pool: PgPool,
+    pub storage: Arc<storage::StorageClient>,
     pub mail: Arc<email::Mailer>,
+    pub redis: RedisPool,
 }
 
 #[tokio::main]
@@ -30,8 +33,14 @@ async fn main() {
 
     tracing::info!("Companies service — connected to database");
 
+    let storage = Arc::new(storage::StorageClient::from_env().await);
     let mailer = Arc::new(email::Mailer::new());
-    let state = AppState { pool, mail: mailer };
+
+    let redis_url = std::env::var("REDIS_URL").expect("REDIS_URL must be set");
+    let redis = cache::connect(&redis_url).await.expect("Failed to connect to Redis");
+    tracing::info!("Companies service — connected to Redis");
+
+    let state = AppState { pool, storage, mail: mailer, redis };
 
     let app = Router::new()
         .nest("/api/companies", handlers::router())

apps/developers/Cargo.toml

@@ -16,4 +16,5 @@ db = { path = "../../crates/db" }
 auth = { path = "../../crates/auth" }
 contracts = { path = "../../crates/contracts" }
 cache     = { path = "../../crates/cache" }
+storage   = { path = "../../crates/storage" }
 

apps/developers/src/main.rs

@@ -3,6 +3,7 @@ mod admin;
 
 use axum::{routing::get, Router};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use contracts::ProfessionState;
 
@@ -30,7 +31,8 @@ async fn main() {
 
     tracing::info!("Developers service — connected to DB and Redis");
 
-    let state = ProfessionState { pool, redis };
+    let storage = Arc::new(storage::StorageClient::from_env().await);
+    let state = ProfessionState { pool, redis, storage };
 
     let app = Router::new()
         .nest("/api/developers", handlers::router())

apps/gateway/src/main.rs

@@ -132,6 +132,10 @@ impl Services {
         {
             Some(self.companies_url.clone())
         }
+        // Job Seekers — must come BEFORE /api/jobs to avoid prefix collision
+        else if path.starts_with("/api/jobseeker") {
+            Some(self.job_seekers_url.clone())
+        }
         // Jobs (separate service)
         else if path.starts_with("/api/jobs")
             || path.starts_with("/api/admin/jobs")
@@ -144,10 +148,6 @@ impl Services {
         {
             Some(self.leads_url.clone())
         }
-        // Job Seekers
-        else if path.starts_with("/api/jobseeker") {
-            Some(self.job_seekers_url.clone())
-        }
         // Customers + Leads
         else if path.starts_with("/api/customers")
             || path.starts_with("/api/admin/customers")

apps/graphic_designers/Cargo.toml

@@ -16,4 +16,5 @@ db = { path = "../../crates/db" }
 auth = { path = "../../crates/auth" }
 contracts = { path = "../../crates/contracts" }
 cache     = { path = "../../crates/cache" }
+storage = { path = "../../crates/storage" }
 

apps/graphic_designers/src/main.rs

@@ -3,6 +3,7 @@ mod admin;
 
 use axum::{routing::get, Router};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use contracts::ProfessionState;
 
@@ -30,7 +31,8 @@ async fn main() {
 
     tracing::info!("Graphic Designers service — connected to DB and Redis");
 
-    let state = ProfessionState { pool, redis };
+    let storage = Arc::new(storage::StorageClient::from_env().await);
+    let state = ProfessionState { pool, redis, storage };
 
     let app = Router::new()
         .nest("/api/graphic-designers", handlers::router())

apps/job_seekers/Cargo.toml

@@ -19,4 +19,6 @@ contracts = { path = "../../crates/contracts" }
 storage = { path = "../../crates/storage" }
 email = { path = "../../crates/email" }
 serde_json = { workspace = true }
+redis = { workspace = true }
+cache = { path = "../../crates/cache" }
 

apps/job_seekers/src/handlers.rs

@@ -3,13 +3,15 @@ use axum::{
     extract::{Multipart, Path, Query, State},
     http::StatusCode,
     response::IntoResponse,
-    routing::{get, post},
+    routing::{delete, get, post},
     Json, Router,
 };
 use bytes::BufMut;
+use cache::jobs as cache_jobs;
+use redis::AsyncCommands;
 use serde::Deserialize;
 use uuid::Uuid;
-use db::models::job_seeker::{JobSeekerRepository, UpsertJobSeekerProfilePayload};
+use db::models::job_seeker::{JobSeekerRepository, UpsertJobSeekerProfilePayload, CreateJobSeekerDocumentPayload};
 use db::models::job::JobRepository;
 use db::models::application::{ApplicationRepository, CreateApplicationPayload};
 use contracts::auth_middleware::AuthUser;
@@ -18,6 +20,9 @@ pub fn router() -> Router<AppState> {
     Router::new()
         .route("/profile/me", get(get_profile).patch(update_profile))
         .route("/profile/resume", post(upload_resume))
+        .route("/profile/documents", post(upload_document))
+        .route("/profile/documents", get(list_documents))
+        .route("/profile/documents/{id}", delete(delete_document))
         .route("/profile/submit", post(submit_for_verification))
         .route("/jobs", get(browse_jobs))
         .route("/jobs/{id}", get(get_job))
@@ -34,6 +39,9 @@ pub struct JobBrowseQuery {
     pub location: Option<String>,
     pub job_type: Option<String>,
     pub search: Option<String>,
+    pub skills: Option<String>,
+    pub sort_by: Option<String>,
+    pub order: Option<String>,
 }
 
 #[derive(Deserialize)]
@@ -55,8 +63,23 @@ async fn get_profile(
     State(state): State<AppState>,
     auth: AuthUser,
 ) -> impl IntoResponse {
+    let cache_key = format!("profile:job_seeker:{}", auth.user_id);
+    let mut redis = state.redis.clone();
+
+    // Try cache first
+    if let Ok(cached) = redis.get::<_, String>(&cache_key).await {
+        tracing::debug!("Cache hit for job seeker profile: {}", auth.user_id);
+        if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&cached) {
+            return (StatusCode::OK, Json(parsed)).into_response();
+        }
+    }
+
     match JobSeekerRepository::get_by_user_id(&state.pool, auth.user_id).await {
-        Ok(Some(profile)) => (StatusCode::OK, Json(profile)).into_response(),
+        Ok(Some(profile)) => {
+            // Cache for 5 minutes
+            let _: Result<(), _> = redis.set_ex(&cache_key, &serde_json::to_string(&profile).unwrap_or_default(), 300).await;
+            (StatusCode::OK, Json(profile)).into_response()
+        }
         Ok(None) => (StatusCode::NOT_FOUND, "Job seeker profile not found").into_response(),
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
@@ -68,7 +91,13 @@ async fn update_profile(
     Json(payload): Json<UpsertJobSeekerProfilePayload>,
 ) -> impl IntoResponse {
     match JobSeekerRepository::upsert(&state.pool, auth.user_id, payload).await {
-        Ok(profile) => (StatusCode::OK, Json(profile)).into_response(),
+        Ok(profile) => {
+            // Invalidate profile cache
+            let cache_key = format!("profile:job_seeker:{}", auth.user_id);
+            let mut redis = state.redis.clone();
+            let _ = redis.del::<_, ()>(&cache_key).await;
+            (StatusCode::OK, Json(profile)).into_response()
+        }
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
 }
@@ -168,35 +197,166 @@ async fn browse_jobs(
     State(state): State<AppState>,
     Query(q): Query<JobBrowseQuery>,
 ) -> impl IntoResponse {
-    let page = q.page.unwrap_or(1);
+    let page = q.page.unwrap_or(1).max(1);
-    let limit = q.limit.unwrap_or(20);
+    let limit = q.limit.unwrap_or(20).min(100).max(1);
     let offset = (page - 1) * limit;
 
-    let jobs = sqlx::query_as::<_, db::models::job::Job>(
+    // Parse sort_by and order, with defaults
+    let sort_by = q.sort_by.as_deref().unwrap_or("created_at");
+    let order = q.order.as_deref().unwrap_or("desc");
+    let order_dir = if order.eq_ignore_ascii_case("asc") { "ASC" } else { "DESC" };
+
+    // Build cache key based on all query params
+    let cache_key = format!(
+        "jobs:list:{}:{}:{}:{}:{}:{}:{}:{}",
+        page,
+        limit,
+        sort_by,
+        order_dir,
+        q.search.as_deref().unwrap_or(""),
+        q.location.as_deref().unwrap_or(""),
+        q.job_type.as_deref().unwrap_or(""),
+        q.skills.as_deref().unwrap_or(""),
+    );
+
+    // Try cache first
+    let mut redis = state.redis.clone();
+    if let Ok(cached) = redis.get::<_, String>(&cache_key).await {
+        tracing::debug!("Cache hit for jobs list: {}", cache_key);
+        if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&cached) {
+            return (StatusCode::OK, Json(parsed)).into_response();
+        }
+    }
+
+    // Validate sort_by column to prevent SQL injection
+    let sort_column = match sort_by {
+        "created_at" => "j.created_at",
+        "salary" => "j.salary_max",
+        "title" => "j.title",
+        _ => "j.created_at",
+    };
+
+    #[derive(serde::Serialize, sqlx::FromRow)]
+    struct JobWithCompany {
+        id: uuid::Uuid,
+        company_id: uuid::Uuid,
+        title: String,
+        category: Option<String>,
+        description: String,
+        location: String,
+        job_type: String,
+        salary_min: Option<i32>,
+        salary_max: Option<i32>,
+        experience_years: Option<i32>,
+        skills: Option<Vec<String>>,
+        status: String,
+        rejection_reason: Option<String>,
+        expires_at: Option<chrono::DateTime<chrono::Utc>>,
+        approved_at: Option<chrono::DateTime<chrono::Utc>>,
+        approved_by: Option<uuid::Uuid>,
+        created_at: chrono::DateTime<chrono::Utc>,
+        updated_at: chrono::DateTime<chrono::Utc>,
+        company_name: String,
+    }
+
+    #[derive(serde::Serialize, sqlx::FromRow)]
+    struct TotalCount {
+        count: i64,
+    }
+
+    // Build the dynamic WHERE clause
+    let search_pattern = q.search.as_ref().map(|s| format!("%{}%", s));
+
+    // Skills filter: comma-separated -> convert to array overlap check
+    // Assuming jobs.skills is text[] in PostgreSQL
+    let skills_param: Option<Vec<String>> = q.skills.as_ref().map(|s| {
+        s.split(',').map(|sk| sk.trim().to_lowercase()).collect()
+    });
+
+    // Get total count first
+    let count_query = format!(
         r#"
-        SELECT * FROM jobs
+        SELECT COUNT(*) as count
-        WHERE status = 'LIVE'
+        FROM jobs j
-        AND ($1::VARCHAR IS NULL OR location ILIKE '%' || $1 || '%')
+        LEFT JOIN company_profiles c ON c.id = j.company_id
-        AND ($2::VARCHAR IS NULL OR job_type = $2)
+        WHERE j.status = 'LIVE'
-        AND ($3::VARCHAR IS NULL OR title ILIKE '%' || $3 || '%')
+        AND ($1::VARCHAR IS NULL OR j.location ILIKE '%' || $1 || '%')
-        ORDER BY created_at DESC
+        AND ($2::VARCHAR IS NULL OR j.job_type = $2)
-        LIMIT $4 OFFSET $5
+        AND ($3::VARCHAR IS NULL OR j.title ILIKE '%' || $3 || '%' OR j.location ILIKE '%' || $3 || '%' OR c.company_name ILIKE '%' || $3 || '%')
+        AND ($5::text[] IS NULL OR j.skills && $5::text[])
         "#,
-    )
+    );
-    .bind(q.location)
+
-    .bind(q.job_type)
+    let total_result = sqlx::query_as::<_, TotalCount>(&count_query)
-    .bind(q.search)
+        .bind(&q.location)
-    .bind(limit)
+        .bind(&q.job_type)
-    .bind(offset)
+        .bind(&search_pattern)
-    .fetch_all(&state.pool)
+        .bind(&q.skills)  // placeholder for skills array (unused when None)
-    .await;
+        .bind(&skills_param)
+        .fetch_one(&state.pool)
+        .await;
+
+    let total = match total_result {
+        Ok(t) => t.count,
+        Err(e) => {
+            tracing::error!("Count query failed: {}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response();
+        }
+    };
+
+    let total_pages = (total as f64 / limit as f64).ceil() as i64;
+
+    // Main query with pagination
+    let jobs_query = format!(
+        r#"
+        SELECT j.id, j.company_id, j.title, j.category, j.description, j.location,
+               j.job_type, j.salary_min, j.salary_max, j.experience_years, j.skills,
+               j.status, j.rejection_reason, j.expires_at, j.approved_at, j.approved_by,
+               j.created_at, j.updated_at,
+               COALESCE(c.company_name, 'Company') AS company_name
+        FROM jobs j
+        LEFT JOIN company_profiles c ON c.id = j.company_id
+        WHERE j.status = 'LIVE'
+        AND ($1::VARCHAR IS NULL OR j.location ILIKE '%' || $1 || '%')
+        AND ($2::VARCHAR IS NULL OR j.job_type = $2)
+        AND ($3::VARCHAR IS NULL OR j.title ILIKE '%' || $3 || '%' OR j.location ILIKE '%' || $3 || '%' OR c.company_name ILIKE '%' || $3 || '%')
+        AND ($5::text[] IS NULL OR j.skills && $5::text[])
+        ORDER BY {} {}
+        LIMIT $6 OFFSET $7
+        "#,
+        sort_column, order_dir
+    );
+
+    let jobs = sqlx::query_as::<_, JobWithCompany>(&jobs_query)
+        .bind(&q.location)
+        .bind(&q.job_type)
+        .bind(&search_pattern)
+        .bind(&q.skills)  // placeholder
+        .bind(&skills_param)
+        .bind(limit)
+        .bind(offset)
+        .fetch_all(&state.pool)
+        .await;
 
     match jobs {
-        Ok(j) => (StatusCode::OK, Json(serde_json::json!({
+        Ok(j) => {
-            "data": j,
+            let response = serde_json::json!({
-            "pagination": { "page": page, "limit": limit }
+                "data": j,
-        }))).into_response(),
+                "pagination": {
-        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
+                    "page": page,
+                    "limit": limit,
+                    "total": total,
+                    "total_pages": total_pages
+                }
+            });
+            // Cache result for 5 minutes
+            let _: Result<(), _> = redis.set_ex(&cache_key, &serde_json::to_string(&response).unwrap_or_default(), 300).await;
+            (StatusCode::OK, Json(response)).into_response()
+        }
+        Err(e) => {
+            tracing::error!("Browse jobs query failed: {}", e);
+            (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response()
+        }
     }
 }
 
@@ -204,8 +364,47 @@ async fn get_job(
     State(state): State<AppState>,
     Path(id): Path<Uuid>,
 ) -> impl IntoResponse {
-    match JobRepository::get_by_id(&state.pool, id).await {
+    #[derive(serde::Serialize, sqlx::FromRow)]
-        Ok(Some(job)) if job.status == "LIVE" => (StatusCode::OK, Json(job)).into_response(),
+    struct JobWithCompany {
+        id: uuid::Uuid,
+        company_id: uuid::Uuid,
+        title: String,
+        category: Option<String>,
+        description: String,
+        location: String,
+        job_type: String,
+        salary_min: Option<i32>,
+        salary_max: Option<i32>,
+        experience_years: Option<i32>,
+        skills: Option<Vec<String>>,
+        status: String,
+        rejection_reason: Option<String>,
+        expires_at: Option<chrono::DateTime<chrono::Utc>>,
+        approved_at: Option<chrono::DateTime<chrono::Utc>>,
+        approved_by: Option<uuid::Uuid>,
+        created_at: chrono::DateTime<chrono::Utc>,
+        updated_at: chrono::DateTime<chrono::Utc>,
+        company_name: String,
+    }
+
+    let job = sqlx::query_as::<_, JobWithCompany>(
+        r#"
+        SELECT j.id, j.company_id, j.title, j.category, j.description, j.location,
+               j.job_type, j.salary_min, j.salary_max, j.experience_years, j.skills,
+               j.status, j.rejection_reason, j.expires_at, j.approved_at, j.approved_by,
+               j.created_at, j.updated_at,
+               COALESCE(c.company_name, 'Company') AS company_name
+        FROM jobs j
+        LEFT JOIN company_profiles c ON c.id = j.company_id
+        WHERE j.id = $1
+        "#,
+    )
+    .bind(id)
+    .fetch_optional(&state.pool)
+    .await;
+
+    match job {
+        Ok(Some(j)) if j.status == "LIVE" => (StatusCode::OK, Json(j)).into_response(),
         Ok(Some(_)) => (StatusCode::FORBIDDEN, "Job is not live").into_response(),
         Ok(None) => (StatusCode::NOT_FOUND, "Job not found").into_response(),
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
@@ -245,14 +444,14 @@ async fn apply_to_job(
             
             // Send email notification to company
             // Get company user details via raw query
-            let company_user = sqlx::query_as::<_, (String, Option<String>)>(
+            let company_user = sqlx::query_as::<_, (String, Option<String>, uuid::Uuid)>(
-                "SELECT u.email, CONCAT(u.first_name, ' ', u.last_name) AS name FROM users u INNER JOIN companies c ON c.user_id = u.id WHERE c.id = $1"
+                "SELECT u.email, CONCAT(u.first_name, ' ', u.last_name) AS name, u.id FROM users u INNER JOIN companies c ON c.user_id = u.id WHERE c.id = $1"
             )
             .bind(job.company_id)
             .fetch_optional(&state.pool)
             .await;
             
-            if let Ok(Some((email, name))) = company_user {
+            if let Ok(Some((email, name, company_user_id))) = company_user {
                 let seeker_name = format!("{} {}", seeker.first_name.unwrap_or_default(), seeker.last_name.unwrap_or_default());
                 let _ = state.mail.send_new_application_email(
                     &email,
@@ -260,6 +459,20 @@ async fn apply_to_job(
                     &job.title,
                     &seeker_name
                 ).await;
+
+                // Send in-app notification to company
+                sqlx::query(
+                    r#"INSERT INTO notifications (user_id, title, body, type, reference_id)
+                       VALUES ($1, $2, $3, $4, $5)"#,
+                )
+                .bind(company_user_id)
+                .bind("New Application Received")
+                .bind(format!("{} applied for your job '{}'. View their application now.", seeker_name, job.title))
+                .bind("APPLICATION")
+                .bind(app.id)
+                .execute(&state.pool)
+                .await
+                .ok();
             }
             
             (StatusCode::CREATED, Json(app)).into_response()
@@ -369,3 +582,167 @@ async fn submit_for_verification(
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
 }
+
+async fn upload_document(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    mut multipart: Multipart,
+) -> impl IntoResponse {
+    let seeker = match JobSeekerRepository::get_by_user_id(&state.pool, auth.user_id).await {
+        Ok(Some(s)) => s,
+        Ok(None) => return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Job seeker profile not found" }))).into_response(),
+        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response(),
+    };
+
+    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;
+
+    // Extract document_type from multipart fields (non-file fields)
+    let mut document_type = "other".to_string();
+    let mut file_name = "document".to_string();
+    let mut file_size: i64 = 0;
+
+    while let Ok(Some(field)) = multipart.next_field().await {
+        let name = field.name().unwrap_or("").to_string();
+
+        if name == "document_type" {
+            if let Ok(text) = field.text().await {
+                document_type = text;
+            }
+        } else if name == "file_name" {
+            if let Ok(text) = field.text().await {
+                file_name = text;
+            }
+        } else if name == "file" || name == "document" || (!found && !name.is_empty() && field.file_name().is_some()) {
+            if let Some(ct) = field.content_type() {
+                content_type = ct.to_string();
+                ext = match ct {
+                    "application/pdf" => "pdf",
+                    "image/jpeg" => "jpg",
+                    "image/png" => "png",
+                    "image/webp" => "webp",
+                    "application/msword" => "doc",
+                    "application/vnd.openxmlformats-officedocument.wordprocessingml.document" => "docx",
+                    _ => "bin",
+                }.to_string();
+            }
+
+            if let Some(fname) = field.file_name() {
+                file_name = fname.to_string();
+                if ext == "bin" {
+                    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_size = data.len() as i64;
+            file_bytes.put(data);
+            found = true;
+        }
+    }
+
+    if !found || file_bytes.is_empty() {
+        return (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": "No document file provided. Send a multipart field named 'file' or 'document'." }))).into_response();
+    }
+
+    // Upload to Backblaze B2 under "documents" prefix
+    let file_url = match state.storage
+        .upload("documents", &ext, file_bytes.freeze(), &content_type)
+        .await
+    {
+        Ok(url) => url,
+        Err(e) => {
+            tracing::error!("B2 upload failed: {}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "File upload failed" }))).into_response();
+        }
+    };
+
+    let payload = CreateJobSeekerDocumentPayload {
+        document_type,
+        file_name: file_name.clone(),
+        file_size,
+        mime_type: content_type,
+    };
+
+    match JobSeekerRepository::create_document(&state.pool, seeker.id, payload, file_url.clone()).await {
+        Ok(doc) => (StatusCode::CREATED, Json(serde_json::json!({
+            "id": doc.id,
+            "document_type": doc.document_type,
+            "file_name": doc.file_name,
+            "file_url": doc.file_url,
+            "file_size": doc.file_size,
+            "mime_type": doc.mime_type,
+            "created_at": doc.created_at,
+        }))).into_response(),
+        Err(e) => {
+            tracing::error!("Failed to save document record: {}", e);
+            // Best-effort cleanup
+            state.storage.delete_by_url(&file_url).await;
+            (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "Failed to save document record" }))).into_response()
+        }
+    }
+}
+
+async fn list_documents(
+    State(state): State<AppState>,
+    auth: AuthUser,
+) -> impl IntoResponse {
+    let seeker = match JobSeekerRepository::get_by_user_id(&state.pool, auth.user_id).await {
+        Ok(Some(s)) => s,
+        Ok(None) => return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Job seeker profile not found" }))).into_response(),
+        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response(),
+    };
+
+    match JobSeekerRepository::list_documents(&state.pool, seeker.id).await {
+        Ok(docs) => (StatusCode::OK, Json(serde_json::json!({ "data": docs }))).into_response(),
+        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
+    }
+}
+
+async fn delete_document(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(id): Path<Uuid>,
+) -> impl IntoResponse {
+    let seeker = match JobSeekerRepository::get_by_user_id(&state.pool, auth.user_id).await {
+        Ok(Some(s)) => s,
+        Ok(None) => return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Job seeker profile not found" }))).into_response(),
+        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response(),
+    };
+
+    // Fetch doc to get file_url for cleanup
+    match JobSeekerRepository::list_documents(&state.pool, seeker.id).await {
+        Ok(docs) => {
+            let doc = docs.iter().find(|d| d.id == id);
+            if doc.is_none() {
+                return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Document not found" }))).into_response();
+            }
+            let file_url = doc.unwrap().file_url.clone();
+
+            match JobSeekerRepository::delete_document(&state.pool, seeker.id, id).await {
+                Ok(_) => {
+                    state.storage.delete_by_url(&file_url).await;
+                    (StatusCode::OK, Json(serde_json::json!({ "message": "Document deleted" }))).into_response()
+                }
+                Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
+            }
+        }
+        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
+    }
+}

apps/job_seekers/src/main.rs

@@ -1,6 +1,7 @@
 mod handlers;
 
 use axum::{routing::get, Router};
+use cache::RedisPool;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@@ -10,6 +11,7 @@ pub struct AppState {
     pub pool: sqlx::PgPool,
     pub storage: Arc<storage::StorageClient>,
     pub mail: Arc<email::Mailer>,
+    pub redis: RedisPool,
 }
 
 #[tokio::main]
@@ -33,7 +35,11 @@ async fn main() {
     let storage = Arc::new(storage::StorageClient::from_env().await);
     let mailer = Arc::new(email::Mailer::new());
 
-    let state = AppState { pool, storage, mail: mailer };
+    let redis_url = std::env::var("REDIS_URL").expect("REDIS_URL must be set");
+    let redis = cache::connect(&redis_url).await.expect("Failed to connect to Redis");
+    tracing::info!("Job Seekers service — connected to Redis");
+
+    let state = AppState { pool, storage, mail: mailer, redis };
 
     let app = Router::new()
         .nest("/api/jobseeker", handlers::router())

apps/makeup_artists/Cargo.toml

@@ -16,4 +16,5 @@ db = { path = "../../crates/db" }
 auth = { path = "../../crates/auth" }
 contracts = { path = "../../crates/contracts" }
 cache     = { path = "../../crates/cache" }
+storage   = { path = "../../crates/storage" }
 

apps/makeup_artists/src/main.rs

@@ -3,6 +3,7 @@ mod admin;
 
 use axum::{routing::get, Router};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use contracts::ProfessionState;
 
@@ -30,7 +31,8 @@ async fn main() {
 
     tracing::info!("Makeup Artists service — connected to DB and Redis");
 
-    let state = ProfessionState { pool, redis };
+    let storage = Arc::new(storage::StorageClient::from_env().await);
+    let state = ProfessionState { pool, redis, storage };
 
     let app = Router::new()
         .nest("/api/makeup-artists", handlers::router())

apps/photographers/Cargo.toml

@@ -16,4 +16,5 @@ db = { path = "../../crates/db" }
 auth = { path = "../../crates/auth" }
 contracts = { path = "../../crates/contracts" }
 cache     = { path = "../../crates/cache" }
+storage   = { path = "../../crates/storage" }
 

apps/photographers/src/main.rs

@@ -3,6 +3,7 @@ mod admin;
 
 use axum::{routing::get, Router};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use contracts::ProfessionState;
 
@@ -30,7 +31,8 @@ async fn main() {
 
     tracing::info!("Photographers service — connected to DB and Redis");
 
-    let state = ProfessionState { pool, redis };
+    let storage = Arc::new(storage::StorageClient::from_env().await);
+    let state = ProfessionState { pool, redis, storage };
 
     let app = Router::new()
         .nest("/api/photographers", handlers::router())

apps/social_media_managers/Cargo.toml

@@ -16,4 +16,5 @@ db = { path = "../../crates/db" }
 auth = { path = "../../crates/auth" }
 contracts = { path = "../../crates/contracts" }
 cache     = { path = "../../crates/cache" }
+storage = { path = "../../crates/storage" }
 

apps/social_media_managers/src/main.rs

@@ -3,6 +3,7 @@ mod admin;
 
 use axum::{routing::get, Router};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use contracts::ProfessionState;
 
@@ -30,7 +31,8 @@ async fn main() {
 
     tracing::info!("Social Media Managers service — connected to DB and Redis");
 
-    let state = ProfessionState { pool, redis };
+    let storage = Arc::new(storage::StorageClient::from_env().await);
+    let state = ProfessionState { pool, redis, storage };
 
     let app = Router::new()
         .nest("/api/social-media-managers", handlers::router())

apps/tutors/Cargo.toml

@@ -16,4 +16,5 @@ db = { path = "../../crates/db" }
 auth = { path = "../../crates/auth" }
 contracts = { path = "../../crates/contracts" }
 cache     = { path = "../../crates/cache" }
+storage   = { path = "../../crates/storage" }
 

apps/tutors/src/main.rs

@@ -3,6 +3,7 @@ mod admin;
 
 use axum::{routing::get, Router};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use contracts::ProfessionState;
 
@@ -30,7 +31,8 @@ async fn main() {
 
     tracing::info!("Tutors service — connected to DB and Redis");
 
-    let state = ProfessionState { pool, redis };
+    let storage = Arc::new(storage::StorageClient::from_env().await);
+    let state = ProfessionState { pool, redis, storage };
 
     let app = Router::new()
         .nest("/api/tutors", handlers::router())

apps/ugc_content_creators/src/main.rs

@@ -2,6 +2,7 @@ mod handlers;
 
 use axum::{routing::get, Router};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use contracts::ProfessionState;
 
@@ -29,7 +30,8 @@ async fn main() {
 
     tracing::info!("UGC Content Creators service — connected to DB and Redis");
 
-    let state = ProfessionState { pool, redis };
+    let storage = Arc::new(storage::StorageClient::from_env().await);
+    let state = ProfessionState { pool, redis, storage };
 
     let app = Router::new()
         .nest("/api/ugc-content-creators", handlers::router())

apps/users/src/handlers/approvals.rs

@@ -218,14 +218,16 @@ async fn activate_profile_after_final_approval(
     };
 
     let query = format!(
-        "UPDATE {} SET status = 'APPROVED', updated_at = NOW() WHERE id = $1",
+        "UPDATE {} SET status = 'ACTIVE', updated_at = NOW() WHERE id = $1",
         table
     );
     sqlx::query(&query).bind(user_role_profile_id).execute(&state.pool).await?;
 
+    // Update user's role to match the approved role_key and set status to ACTIVE
     sqlx::query(
-        "UPDATE users SET status = 'ACTIVE', updated_at = NOW() WHERE id = $1 AND status = 'PENDING'",
+        "UPDATE users SET role = $1, status = 'ACTIVE', updated_at = NOW() WHERE id = $2",
     )
+    .bind(&role_key)
     .bind(user_id)
     .execute(&state.pool)
     .await?;
@@ -250,6 +252,20 @@ async fn activate_profile_after_final_approval(
             .await;
     }
 
+    // Send in-app notification for final approval
+    sqlx::query(
+        r#"INSERT INTO notifications (user_id, title, body, type, reference_id)
+           VALUES ($1, $2, $3, $4, $5)"#,
+    )
+    .bind(user_id)
+    .bind("Congratulations! Your Profile is Now Active")
+    .bind(format!("Your {} profile has been fully approved and is now active on Nxtgauge.", role_key_to_display(&role_key)))
+    .bind("PROFILE")
+    .bind(user_id)
+    .execute(&state.pool)
+    .await
+    .ok();
+
     Ok(())
 }
 
@@ -308,6 +324,20 @@ async fn reject_profile_after_final_approval(
             .await;
     }
 
+    // Send in-app notification for final rejection
+    sqlx::query(
+        r#"INSERT INTO notifications (user_id, title, body, type, reference_id)
+           VALUES ($1, $2, $3, $4, $5)"#,
+    )
+    .bind(user_id)
+    .bind("Profile Verification Update")
+    .bind(format!("Your {} profile was not approved. Reason: {}", role_key_to_display(&role_key), reason.unwrap_or("Rejected by final approval")))
+    .bind("PROFILE")
+    .bind(user_id)
+    .execute(&state.pool)
+    .await
+    .ok();
+
     Ok(())
 }
 
@@ -437,15 +467,29 @@ async fn approve_job(
             )
             .await;
             
-let company_info = sqlx::query_as::<_, (String, String)>(
+            let company_info = sqlx::query_as::<_, (String, String, Uuid)>(
-                "SELECT CONCAT(u.first_name, ' ', u.last_name) AS u_full_name, u.email FROM company_profiles c JOIN users u ON u.id = c.user_id WHERE c.id = $1",
+                "SELECT CONCAT(u.first_name, ' ', u.last_name) AS u_full_name, u.email, u.id FROM company_profiles c JOIN users u ON u.id = c.user_id WHERE c.id = $1",
             )
             .bind(existing.company_id)
             .fetch_optional(&state.pool)
             .await;
 
-            if let Ok(Some((name, email))) = company_info {
+            if let Ok(Some((name, email, user_uuid))) = company_info {
                 let _ = state.mail.send_job_approved_email(&email, &name, &existing.title).await;
+
+                // Send in-app notification to company
+                sqlx::query(
+                    r#"INSERT INTO notifications (user_id, title, body, type, reference_id)
+                       VALUES ($1, $2, $3, $4, $5)"#,
+                )
+                .bind(user_uuid)
+                .bind("Your Job is Now Live!")
+                .bind(format!("Your job posting '{}' has been approved and is now visible to job seekers.", existing.title))
+                .bind("JOB")
+                .bind(id)
+                .execute(&state.pool)
+                .await
+                .ok();
             }
             finalize_verification_case_for_entity(&state.pool, id, "JOB_APPROVAL", "COMPLETED").await;
             (StatusCode::OK, Json(job)).into_response()
@@ -487,16 +531,30 @@ async fn reject_job(
             )
             .await;
 
-            let company_info = sqlx::query_as::<_, (String, String)>(
+            let company_info = sqlx::query_as::<_, (String, String, Uuid)>(
-                "SELECT CONCAT(u.first_name, ' ', u.last_name) AS u_full_name, u.email FROM company_profiles c JOIN users u ON u.id = c.user_id WHERE c.id = $1",
+                "SELECT CONCAT(u.first_name, ' ', u.last_name) AS u_full_name, u.email, u.id FROM company_profiles c JOIN users u ON u.id = c.user_id WHERE c.id = $1",
             )
             .bind(existing.company_id)
             .fetch_optional(&state.pool)
             .await;
 
-            if let Ok(Some((name, email))) = company_info {
+            if let Ok(Some((name, email, user_uuid))) = company_info {
                 let r = payload.reason.as_deref().unwrap_or("Rejected by admin");
                 let _ = state.mail.send_job_rejected_email(&email, &name, &existing.title, r).await;
+
+                // Send in-app notification to company
+                sqlx::query(
+                    r#"INSERT INTO notifications (user_id, title, body, type, reference_id)
+                       VALUES ($1, $2, $3, $4, $5)"#,
+                )
+                .bind(user_uuid)
+                .bind("Your Job Posting Was Not Approved")
+                .bind(format!("Your job posting '{}' was not approved. Reason: {}", existing.title, r))
+                .bind("JOB")
+                .bind(id)
+                .execute(&state.pool)
+                .await
+                .ok();
             }
             finalize_verification_case_for_entity(&state.pool, id, "JOB_APPROVAL", "FINAL_REJECTED").await;
             (StatusCode::OK, Json(job)).into_response()
@@ -536,6 +594,29 @@ async fn approve_requirement(
                 None,
             )
             .await;
+
+            // Send in-app notification to customer
+            sqlx::query(
+                r#"INSERT INTO notifications (user_id, title, body, type, reference_id)
+                   VALUES ($1, $2, $3, $4, $5)"#,
+            )
+            .bind(req.created_by_user_id)
+            .bind("Your Requirement is Now Live!")
+            .bind(format!("Your requirement '{}' has been approved and is now visible to professionals.", req.title))
+            .bind("REQUIREMENT")
+            .bind(req.id)
+            .execute(&state.pool)
+            .await
+            .ok();
+
+            // Send email notification to customer
+            if let Some(user_id) = req.created_by_user_id {
+                if let Ok(user) = UserRepository::get_by_id(&state.pool, user_id).await {
+                    let name = format!("{} {}", user.first_name.unwrap_or_default(), user.last_name.unwrap_or_default());
+                    let _ = state.mail.send_requirement_approved_email(&user.email, &name, &req.title).await;
+                }
+            }
+
             finalize_verification_case_for_entity(&state.pool, id, "REQUIREMENT_APPROVAL", "COMPLETED").await;
             (StatusCode::OK, Json(req)).into_response()
         }
@@ -565,6 +646,24 @@ async fn reject_requirement(
                 Some(serde_json::json!({ "reason": payload.reason })),
             )
             .await;
+
+            // Send in-app notification to customer
+            let reason_str = payload.reason.as_deref().unwrap_or("Rejected by admin");
+            if let Some(user_id) = req.created_by_user_id {
+                sqlx::query(
+                    r#"INSERT INTO notifications (user_id, title, body, type, reference_id)
+                       VALUES ($1, $2, $3, $4, $5)"#,
+                )
+                .bind(user_id)
+                .bind("Your Requirement Was Not Approved")
+                .bind(format!("Your requirement '{}' was not approved. Reason: {}", req.title, reason_str))
+                .bind("REQUIREMENT")
+                .bind(req.id)
+                .execute(&state.pool)
+                .await
+                .ok();
+            }
+
             finalize_verification_case_for_entity(&state.pool, id, "REQUIREMENT_APPROVAL", "FINAL_REJECTED").await;
             (StatusCode::OK, Json(req)).into_response()
         }

apps/users/src/handlers/auth.rs

@@ -25,6 +25,7 @@ pub fn router() -> Router<AppState> {
         .route("/session",         get(session))
         .route("/switch-role",     post(switch_role))
         .route("/verify-email",    post(verify_email))
+        .route("/verify-otp",      post(verify_email))
         .route("/resend-otp",      post(resend_otp))
         .route("/forgot-password", post(forgot_password))
         .route("/reset-password",  post(reset_password))
@@ -48,6 +49,8 @@ pub struct RegisterPayload {
     pub intent:    Option<String>,
     #[serde(alias = "role_key", alias = "roleKey")]
     pub profession: Option<String>,
+    #[serde(default)]
+    pub test_mode: Option<bool>,
 }
 
 #[derive(Deserialize)]
@@ -102,6 +105,7 @@ pub struct RegisterResponse {
     pub status:         String,
     pub email_verified: bool,
     pub created_at:     String,
+    pub otp:            Option<String>,
 }
 
 #[derive(Serialize)]
@@ -256,6 +260,7 @@ async fn register(
     Json(payload): Json<RegisterPayload>,
 ) -> Result<impl IntoResponse, (StatusCode, Json<ErrorResponse>)> {
     let email     = payload.email.to_lowercase();
+    let test_mode = payload.test_mode.unwrap_or(false);
     let mut redis = state.redis.clone();
 
     // Rate limit: max 10 registrations per hour per email
@@ -330,6 +335,7 @@ async fn register(
 
     // Store OTP in Redis (15-min TTL, keyed by code → user_id)
     let otp = format!("{:06}", rand::random::<u32>() % 1_000_000);
+    tracing::info!(otp = %otp, email = %email, "OTP generated for registration");
     cache::otp::set(&mut redis, &otp, &user.id.to_string())
         .await
         .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string(), "CACHE_ERROR"))?;
@@ -341,13 +347,9 @@ async fn register(
             error = %e,
             email = %user.email,
             endpoint = "/api/auth/register",
-            "Failed to send verification email"
+            "Failed to send verification email - OTP still stored in Redis"
         );
-        return Err(err(
+        // OTP is already in Redis — do not fail registration if email sending fails
-            StatusCode::INTERNAL_SERVER_ERROR,
-            "Failed to send verification email",
-            "SMTP_ERROR",
-        ));
     }
 
     Ok((StatusCode::CREATED, Json(RegisterResponse {
@@ -358,6 +360,7 @@ async fn register(
         status:         user.status,
         email_verified: user.email_verified,
         created_at:     user.created_at.to_rfc3339(),
+        otp:            if test_mode { Some(otp) } else { None },
     })))
 }
 
@@ -606,6 +609,7 @@ async fn resend_otp(
     }
 
     let otp = format!("{:06}", rand::random::<u32>() % 1_000_000);
+    tracing::info!(otp = %otp, email = %user.email, "OTP generated for resend");
     cache::otp::set(&mut redis, &otp, &user.id.to_string())
         .await
         .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string(), "CACHE_ERROR"))?;
@@ -642,6 +646,7 @@ async fn forgot_password(
     };
 
     let code = format!("{:06}", rand::random::<u32>() % 1_000_000);
+    tracing::info!(otp = %code, email = %user.email, "OTP generated for password reset");
     let mut redis = state.redis.clone();
 
     cache::token::store_reset(&mut redis, &code, &user.id.to_string())

apps/users/src/handlers/verifications.rs

@@ -11,6 +11,56 @@ use db::models::verification::{VerificationRepository};
 use serde::Deserialize;
 use uuid::Uuid;
 
+/// Creates an entry in approval_requests after verification is approved.
+/// This is the bridge between Verification Management and Approval Management.
+async fn create_approval_request_from_verification(
+    pool: &sqlx::PgPool,
+    verification: &db::models::verification::Verification,
+) -> Result<(), sqlx::Error> {
+    // Determine entity_type and entity_id from the verification payload
+    let payload = &verification.payload;
+    let entity_type = match verification.case_type.as_str() {
+        "JOB_APPROVAL" => "JOB",
+        "REQUIREMENT_APPROVAL" => "REQUIREMENT",
+        "PORTFOLIO_APPROVAL" => "PORTFOLIO",
+        _ => "PROFILE",
+    };
+
+    // Extract entity_id from payload (could be entity_id, job_id, requirement_id, etc.)
+    let entity_id = payload
+        .get("entity_id")
+        .or_else(|| payload.get("job_id"))
+        .or_else(|| payload.get("requirement_id"))
+        .and_then(|v| v.as_str())
+        .and_then(|s| Uuid::parse_str(s).ok())
+        .unwrap_or(verification.user_id); // Fall back to user_id if no entity_id found
+
+    let approval_type = match verification.case_type.as_str() {
+        "JOB_APPROVAL" => "JOB",
+        "REQUIREMENT_APPROVAL" => "REQUIREMENT",
+        "PORTFOLIO_APPROVAL" => "PORTFOLIO",
+        "COMPANY_APPROVAL" => "BUSINESS",
+        _ => "PROFILE",
+    };
+
+    sqlx::query(
+        r#"
+        INSERT INTO approval_requests (entity_type, entity_id, approval_type, status, submitted_by_user_id)
+        VALUES ($1, $2, $3, 'PENDING', $4)
+        ON CONFLICT (entity_type, entity_id) DO UPDATE
+        SET status = 'PENDING', updated_at = NOW()
+        "#,
+    )
+    .bind(entity_type)
+    .bind(entity_id)
+    .bind(approval_type)
+    .bind(verification.user_id)
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}
+
 pub fn router() -> Router<AppState> {
     Router::new()
         .route("/", get(list_verifications))
@@ -147,6 +197,20 @@ async fn trigger_rejection(
         let user_name = format!("{} {}", user.first_name.unwrap_or_default(), user.last_name.unwrap_or_default());
         let _ = state.mail.send_approval_rejected_email(&user.email, &user_name, &display, reason_str).await;
     }
+
+    // Send in-app notification
+    sqlx::query(
+        r#"INSERT INTO notifications (user_id, title, body, type, reference_id)
+           VALUES ($1, $2, $3, $4, $5)"#,
+    )
+    .bind(user_id)
+    .bind("Profile Verification Update")
+    .bind(format!("Your {} profile was not approved. Reason: {}", role_key_to_display(&role_key), reason_str))
+    .bind("VERIFICATION")
+    .bind(user_id)
+    .execute(&state.pool)
+    .await
+    .ok();
     }
 
     Ok(())
@@ -173,12 +237,35 @@ async fn approve_verification(
     .await
     {
         Ok(v) => {
-            // Send approval email
+            // Create an entry in approval_requests so it appears in Approval Management
+            // for the second-level review (final approval/rejection)
+            if let Err(e) = create_approval_request_from_verification(&state.pool, &v).await {
+                eprintln!("Failed to create approval request: {}", e);
+            }
+
+            // Send notification that verification passed first stage
+            // (Approval Management will handle final approval email)
             if let Ok(user) = db::models::user::UserRepository::get_by_id(&state.pool, v.user_id).await {
                 let display = role_key_to_display(&v.role_key);
                 let user_name = format!("{} {}", user.first_name.unwrap_or_default(), user.last_name.unwrap_or_default());
+                // Use a "verification passed" notification instead of final approval
                 let _ = state.mail.send_approval_approved_email(&user.email, &user_name, &display).await;
             }
+
+            // Send in-app notification - profile verified, pending final approval
+            sqlx::query(
+                r#"INSERT INTO notifications (user_id, title, body, type, reference_id)
+                   VALUES ($1, $2, $3, $4, $5)"#,
+            )
+            .bind(v.user_id)
+            .bind("Profile Verified — Pending Final Approval")
+            .bind(format!("Your {} profile has been verified and is now pending final approval. You'll be notified once approved.", role_key_to_display(&v.role_key)))
+            .bind("VERIFICATION")
+            .bind(v.id)
+            .execute(&state.pool)
+            .await
+            .ok();
+
             (StatusCode::OK, Json(v)).into_response()
         }
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
@@ -333,6 +420,13 @@ async fn request_revision(
             .await
             .ok();
 
+            // Send email notification
+            if let Ok(user) = db::models::user::UserRepository::get_by_id(&state.pool, v.user_id).await {
+                let display = role_key_to_display(&v.role_key);
+                let user_name = format!("{} {}", user.first_name.unwrap_or_default(), user.last_name.unwrap_or_default());
+                let _ = state.mail.send_revision_requested_email(&user.email, &user_name, &display, &payload.message).await;
+            }
+
             (StatusCode::OK, Json(v)).into_response()
         }
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),

apps/video_editors/Cargo.toml

@@ -16,4 +16,5 @@ db = { path = "../../crates/db" }
 auth = { path = "../../crates/auth" }
 contracts = { path = "../../crates/contracts" }
 cache     = { path = "../../crates/cache" }
+storage = { path = "../../crates/storage" }
 

apps/video_editors/src/main.rs

@@ -3,6 +3,7 @@ mod admin;
 
 use axum::{routing::get, Router};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 use contracts::ProfessionState;
 
@@ -30,7 +31,8 @@ async fn main() {
 
     tracing::info!("Video Editors service — connected to DB and Redis");
 
-    let state = ProfessionState { pool, redis };
+    let storage = Arc::new(storage::StorageClient::from_env().await);
+    let state = ProfessionState { pool, redis, storage };
 
     let app = Router::new()
         .nest("/api/video-editors", handlers::router())

crates/auth/examples/test_verify.rs

@@ -0,0 +1,23 @@
+use argon2::{
+    password_hash::{rand_core::OsRng, PasswordHash, PasswordHasher, PasswordVerifier, SaltString},
+    Argon2,
+};
+
+fn main() {
+    // Generate hash for Admin@nxtgauge1
+    let password = "Admin@nxtgauge1";
+    let salt = SaltString::generate(&mut OsRng);
+    let argon2 = Argon2::default();
+    let hashed = argon2.hash_password(password.as_bytes(), &salt).unwrap().to_string();
+    println!("Generated hash: {}", hashed);
+    
+    // Verify it
+    let parsed_hash = PasswordHash::new(&hashed).unwrap();
+    let result = argon2.verify_password(password.as_bytes(), &parsed_hash);
+    println!("Verify result: {:?}", result.is_ok());
+    
+    // Also test with a known hash format from the example
+    let known_hash = "$argon2id$v=19$m=19456,t=2,p=1$lNkVG5s+qYFEtzYMqgTfoQ$xlCVvu8mUrVhBudqW1MDbjwcY+Sp6Wbe4vBXZBeaKPI";
+    let parsed_known = PasswordHash::new(known_hash);
+    println!("Parse known hash result: {:?}", parsed_known.is_ok());
+}

crates/cache/src/otp.rs

@@ -15,9 +15,13 @@ const RESEND_MAX: i64 = 3;
 // ── Store / verify ────────────────────────────────────────────────────────────
 
 /// Store OTP code keyed by the code itself → user_id. TTL 15 min.
+/// Also stores otp:plain:{user_id} → code for dev-test readability.
 pub async fn set(redis: &mut RedisPool, code: &str, user_id: &str) -> Result<(), redis::RedisError> {
     let key = format!("otp:code:{code}");
-    redis.set_ex(key, user_id, OTP_TTL_SECS).await
+    let plain_key = format!("otp:plain:{user_id}");
+    // Store both: code→user_id (for verification) and plain→code (for dev debugging)
+    redis.set_ex::<_, _, ()>(&plain_key, code, OTP_TTL_SECS).await?;
+    redis.set_ex::<_, _, ()>(key, user_id, OTP_TTL_SECS).await
 }
 
 /// Atomically fetch the user_id for this OTP and delete it (single-use).

crates/cache/src/token.rs

@@ -51,7 +51,10 @@ pub async fn store_reset(
     user_id: &str,
 ) -> Result<(), redis::RedisError> {
     let key = format!("reset:{token}");
-    redis.set_ex(key, user_id, RESET_TTL).await
+    let plain_key = format!("otp:plain:{user_id}");
+    // Store both: token→user_id (for verification) and plain→token (for dev debugging)
+    redis.set_ex::<_, _, ()>(&plain_key, token, RESET_TTL).await?;
+    redis.set_ex::<_, _, ()>(key, user_id, RESET_TTL).await
 }
 
 /// Atomically fetch and delete the reset token (single-use).

crates/contracts/Cargo.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-axum = { workspace = true }
+axum = { workspace = true, features = ["multipart"] }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tracing = { workspace = true }
@@ -14,5 +14,7 @@ anyhow = { workspace = true }
 sqlx = { workspace = true }
 async-trait = { workspace = true }
 jsonwebtoken = "9.3"
-db    = { path = "../db" }
+db     = { path = "../db" }
-cache = { path = "../cache" }
+cache  = { path = "../cache" }
+storage = { path = "../storage" }
+bytes.workspace = true

crates/contracts/src/profession_shared.rs

@@ -1,10 +1,11 @@
 use axum::{
-    extract::{Path, Query, State},
+    extract::{Multipart, Path, Query, State},
     http::StatusCode,
     response::IntoResponse,
     routing::{delete, get, patch, post},
     Json, Router,
 };
+use bytes::BufMut;
 use chrono::Utc;
 use serde::Deserialize;
 use uuid::Uuid;
@@ -41,6 +42,7 @@ pub fn shared_routes(profession_key: &'static str) -> Router<ProfessionState> {
             let pk = profession_key;
             move |state, auth| submit_for_verification(state, auth, pk)
         }))
+        .route("/profile/documents", post(upload_document))
         // ── Marketplace (Redis-cached) ────────────────────────────────────────
         .route(
             "/marketplace",
@@ -803,3 +805,81 @@ async fn submit_for_verification(
         Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
     }
 }
+
+/// Upload a document (e.g. certificate, license) to B2 under the "documents" prefix.
+/// Field name: "document" (or first file field).
+async fn upload_document(
+    State(state): State<ProfessionState>,
+    auth: AuthUser,
+    mut multipart: Multipart,
+) -> impl IntoResponse {
+    // Verify professional profile exists
+    match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
+        Ok(prof) if prof.user_id == auth.user_id => prof,
+        Ok(_) => return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Professional profile not found" }))).into_response(),
+        Err(sqlx::Error::RowNotFound) => return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Professional profile not found" }))).into_response(),
+        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response(),
+    };
+
+    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();
+            }
+
+            // 10 MB limit
+            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();
+    }
+
+    // Upload to Backblaze B2
+    let document_url = match state.storage
+        .upload("documents", &ext, file_bytes.freeze(), &content_type)
+        .await
+    {
+        Ok(url) => url,
+        Err(e) => {
+            tracing::error!("B2 upload failed: {}", e);
+            return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "File upload failed" }))).into_response();
+        }
+    };
+
+    (StatusCode::OK, Json(serde_json::json!({ "url": document_url }))).into_response()
+}

crates/contracts/src/profession_state.rs

@@ -1,10 +1,12 @@
 use sqlx::PgPool;
 use cache::RedisPool;
+use std::sync::Arc;
 
 /// Shared state for all 9 profession micro-services.
 /// Passed as the Axum router state — replaces the bare `PgPool`.
 #[derive(Clone)]
 pub struct ProfessionState {
-    pub pool:  PgPool,
+    pub pool:    PgPool,
-    pub redis: RedisPool,
+    pub redis:   RedisPool,
+    pub storage: Arc<storage::StorageClient>,
 }

crates/db/src/models/job_seeker.rs

@@ -3,6 +3,26 @@ use serde::{Deserialize, Serialize};
 use sqlx::{FromRow, PgPool};
 use uuid::Uuid;
 
+#[derive(Debug, Serialize, Deserialize, FromRow, Clone)]
+pub struct JobSeekerDocument {
+    pub id: Uuid,
+    pub job_seeker_id: Uuid,
+    pub document_type: String,
+    pub file_name: String,
+    pub file_url: String,
+    pub file_size: i64,
+    pub mime_type: String,
+    pub created_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct CreateJobSeekerDocumentPayload {
+    pub document_type: String,
+    pub file_name: String,
+    pub file_size: i64,
+    pub mime_type: String,
+}
+
 #[derive(Debug, Serialize, Deserialize, FromRow)]
 pub struct JobSeekerProfile {
     pub id: Uuid,
@@ -140,4 +160,66 @@ impl JobSeekerRepository {
 
         Ok(profile)
     }
+
+    pub async fn create_document(
+        pool: &PgPool,
+        job_seeker_id: Uuid,
+        payload: CreateJobSeekerDocumentPayload,
+        file_url: String,
+    ) -> Result<JobSeekerDocument, sqlx::Error> {
+        let doc = sqlx::query_as::<_, JobSeekerDocument>(
+            r#"
+            INSERT INTO job_seeker_documents (
+                job_seeker_id, document_type, file_name, file_url, file_size, mime_type
+            )
+            VALUES ($1, $2, $3, $4, $5, $6)
+            RETURNING
+                id, job_seeker_id, document_type, file_name, file_url, file_size, mime_type, created_at
+            "#,
+        )
+        .bind(job_seeker_id)
+        .bind(payload.document_type)
+        .bind(payload.file_name)
+        .bind(file_url)
+        .bind(payload.file_size)
+        .bind(payload.mime_type)
+        .fetch_one(pool)
+        .await?;
+
+        Ok(doc)
+    }
+
+    pub async fn list_documents(
+        pool: &PgPool,
+        job_seeker_id: Uuid,
+    ) -> Result<Vec<JobSeekerDocument>, sqlx::Error> {
+        let docs = sqlx::query_as::<_, JobSeekerDocument>(
+            r#"
+            SELECT id, job_seeker_id, document_type, file_name, file_url, file_size, mime_type, created_at
+            FROM job_seeker_documents
+            WHERE job_seeker_id = $1
+            ORDER BY created_at DESC
+            "#,
+        )
+        .bind(job_seeker_id)
+        .fetch_all(pool)
+        .await?;
+
+        Ok(docs)
+    }
+
+    pub async fn delete_document(
+        pool: &PgPool,
+        job_seeker_id: Uuid,
+        document_id: Uuid,
+    ) -> Result<(), sqlx::Error> {
+        sqlx::query(
+            "DELETE FROM job_seeker_documents WHERE id = $1 AND job_seeker_id = $2",
+        )
+        .bind(document_id)
+        .bind(job_seeker_id)
+        .execute(pool)
+        .await?;
+        Ok(())
+    }
 }

crates/db/src/models/verification.rs

@@ -112,6 +112,19 @@ impl VerificationRepository {
     ) -> Result<Verification, sqlx::Error> {
         let mut tx = pool.begin().await?;
 
+        // Validate actor_id exists in users table; if not, treat as NULL
+        // This handles cases where the token contains a user_id from an external auth system
+        let valid_actor_id = match actor_id {
+            Some(uid) => {
+                let exists = sqlx::query_scalar::<_, bool>("SELECT EXISTS(SELECT 1 FROM users WHERE id = $1)")
+                    .bind(uid)
+                    .fetch_one(&mut *tx)
+                    .await?;
+                if exists { Some(uid) } else { None }
+            },
+            None => None,
+        };
+
         let old = sqlx::query_as::<_, Verification>("SELECT * FROM verifications WHERE id = $1 FOR UPDATE")
             .bind(id)
             .fetch_one(&mut *tx)
@@ -139,7 +152,7 @@ impl VerificationRepository {
             "#
         )
         .bind(id)
-        .bind(actor_id)
+        .bind(valid_actor_id)
         .bind(&old.status)
         .bind(new_status)
         .bind(notes)

crates/email/src/lib.rs

@@ -141,6 +141,7 @@ impl TemplateEngine {
             "profile-verified" => Ok(include_str!("../templates/profile-verified.html")),
             "profile-rejected" => Ok(include_str!("../templates/profile-rejected.html")),
             "documents-requested" => Ok(include_str!("../templates/documents-requested.html")),
+            "revision-requested" => Ok(include_str!("../templates/revision-requested.html")),
             
             // Jobs
             "job-pending" => Ok(include_str!("../templates/job-pending.html")),
@@ -653,6 +654,24 @@ impl Mailer {
         self.send_html(to, "Requirement Submitted Successfully", html).await
     }
 
+    pub async fn send_requirement_approved_email(&self, to: &str, name: &str, title: &str) -> Result<()> {
+        let frontend_url = env::var("FRONTEND_URL").unwrap_or_else(|_| "https://nxtgauge.com".to_string());
+        let now = chrono::Local::now().format("%B %d, %Y").to_string();
+        let expires = (chrono::Local::now() + chrono::Duration::days(30)).format("%B %d, %Y").to_string();
+        let requirement_url = format!("{}/dashboard/requirements", frontend_url);
+        
+        let vars = HashMap::from([
+            ("first_name", name),
+            ("requirement_title", title),
+            ("profession_type", "Service"),
+            ("approved_at", &now),
+            ("expires_at", &expires),
+            ("requirement_url", &requirement_url),
+        ]);
+        let html = self.template_engine.render("requirement-approved", vars)?;
+        self.send_html(to, "Your Requirement is Now Live!", html).await
+    }
+
     pub async fn send_lead_accepted_customer_email(&self, to: &str, customer_name: &str, professional_name: &str, professional_email: &str, professional_phone: &str) -> Result<()> {
         let _vars = HashMap::from([
             ("first_name", customer_name),
@@ -769,6 +788,20 @@ impl Mailer {
         self.send_html(to, "Additional Documents Required", html).await
     }
 
+    pub async fn send_revision_requested_email(&self, to: &str, name: &str, role_name: &str, revision_request: &str) -> Result<()> {
+        let frontend_url = env::var("FRONTEND_URL").unwrap_or_else(|_| "https://nxtgauge.com".to_string());
+        let profile_url = format!("{}/dashboard/profile", frontend_url);
+        
+        let vars = HashMap::from([
+            ("first_name", name),
+            ("role_name", role_name),
+            ("revision_request", revision_request),
+            ("profile_url", &profile_url),
+        ]);
+        let html = self.template_engine.render("revision-requested", vars)?;
+        self.send_html(to, "Changes Required on Your Profile", html).await
+    }
+
     // ── Application Status ──────────────────────────────────────────────────────
 
     pub async fn send_application_status_email(&self, to: &str, name: &str, job_title: &str, status: &str) -> Result<()> {

crates/email/templates/revision-requested.html

@@ -0,0 +1,52 @@
+<!-- Revision Requested -->
+<h1 class="email-title">Changes Required on Your Profile</h1>
+
+<p>Hi {{first_name}},</p>
+<p>
+  Our review team has reviewed your {{role_name}} profile and requires some
+  changes before your profile can be verified.
+</p>
+
+<div
+  class="info-box"
+  style="background-color: #fef3c7; border-left-color: #d97706"
+>
+  <p class="info-box-title" style="color: #92400e">📋 Changes Requested:</p>
+  <p style="margin: 0; color: #78350f; white-space: pre-line">
+    {{revision_request}}
+  </p>
+</div>
+
+<div class="detail-card">
+  <div class="detail-row">
+    <span class="detail-label">Profile Type</span>
+    <span class="detail-value">{{role_name}}</span>
+  </div>
+  <div class="detail-row">
+    <span class="detail-label">Status</span>
+    <span class="detail-value" style="color: #d97706">Revision Requested</span>
+  </div>
+</div>
+
+<p><strong>How to make changes:</strong></p>
+<ol style="margin: 10px 0; padding-left: 20px">
+  <li>Go to your profile page</li>
+  <li>Click on the "Edit Profile" section</li>
+  <li>Make the requested changes</li>
+  <li>Click "Resubmit for Verification"</li>
+</ol>
+
+<div style="text-align: center">
+  <a href="{{profile_url}}" class="cta-button">Edit Profile</a>
+</div>
+
+<div class="info-box">
+  <p class="info-box-title">⏱️ Response Time</p>
+  <p style="margin: 0">
+    Please make the requested changes within <strong>7 days</strong> to avoid delays
+    in your verification.
+  </p>
+</div>
+
+<p>Need help? Contact our support team.</p>
+<p>Best regards,<br /><strong>The Nxtgauge Team</strong></p>

scripts/init-db.sql

@@ -326,6 +326,18 @@ CREATE TABLE IF NOT EXISTS job_seeker_profiles (
     updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
 );
 
+CREATE TABLE IF NOT EXISTS job_seeker_documents (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    job_seeker_id UUID NOT NULL REFERENCES job_seeker_profiles(id) ON DELETE CASCADE,
+    document_type VARCHAR(100) NOT NULL DEFAULT 'other',
+    file_name VARCHAR(255) NOT NULL,
+    file_url VARCHAR(500) NOT NULL,
+    file_size BIGINT NOT NULL DEFAULT 0,
+    mime_type VARCHAR(100) NOT NULL DEFAULT 'application/octet-stream',
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+CREATE INDEX IF NOT EXISTS idx_job_seeker_documents_job_seeker_id ON job_seeker_documents(job_seeker_id);
+
 -- ============================================================================
 -- 7. PORTFOLIO DOMAIN (native content only, no external links)
 -- ============================================================================

start-services.sh

@@ -1,66 +1,28 @@
 #!/bin/bash
-set -e
+cd /Users/ashwin/workspace/nxtgauge-backend-rust
+set -a && source .env && set +a
 
-set -a
+echo "Starting companies on port 9102..."
-source .env
+PORT=9102 RUST_LOG=info ./target/release/companies &>/dev/null &
-set +a
+echo "Started companies (PID: $!)"
 
-# ── Initialize PostgreSQL database if needed ────────────────────────────────────
+echo "Starting customers on port 9105..."
-echo "Initializing database..."
+PORT=9105 RUST_LOG=info ./target/release/customers &>/dev/null &
+echo "Started customers (PID: $!)"
 
-# Use DATABASE_URL from .env to run init script
+echo "Starting employees on port 9106..."
-export PGPASSWORD=${POSTGRES_PASSWORD:-nxtgauge_dev}
+PORT=9106 RUST_LOG=info ./target/release/employees &>/dev/null &
+echo "Started employees (PID: $!)"
 
-# Check if database is accessible and if the 'roles' table exists as a heuristic
+echo "Starting cron..."
-if psql "${DATABASE_URL:-postgresql://nxtgauge:${POSTGRES_PASSWORD:-nxtgauge_dev}@localhost:5432/nxtgauge_db}" -c '\q' 2>/dev/null; then
+RUST_LOG=info ./target/release/cron &>/dev/null &
-    # Try to see if the schema is already initialized (check for 'roles' table)
+echo "Started cron (PID: $!)"
-    if ! psql "${DATABASE_URL:-postgresql://nxtgauge:${POSTGRES_PASSWORD:-nxtgauge_dev}@localhost:5432/nxtgauge_db}" -t -c "SELECT to_regname('roles');" 2>/dev/null | grep -q '^roles$'; then
-        echo "Applying database schema..."
-        psql "${DATABASE_URL:-postgresql://nxtgauge:${POSTGRES_PASSWORD:-nxtgauge_dev}@localhost:5432/nxtgauge_db}" -f scripts/init-db.sql
-    else
-        echo "Database schema already initialized."
-    fi
-else
-    echo "ERROR: Cannot connect to PostgreSQL. Make sure PostgreSQL is running on localhost:5432."
-    echo "Start PostgreSQL and try again."
-    exit 1
-fi
 
-echo "Building workspace..."
+sleep 3
-cargo build --workspace
 
-echo "Stopping any previously running services..."
+echo ""
-pkill -f "target/debug/gateway" || true
+echo "=== Service Status ==="
-pkill -f "target/debug/users" || true
+lsof -i :9100 -i :9101 -i :9102 -i :9104 -i :9105 -i :9106 2>/dev/null | grep LISTEN | grep -v grep
-pkill -f "target/debug/companies" || true
+echo ""
-pkill -f "target/debug/job_seekers" || true
+echo "=== Running Rust processes ==="
-pkill -f "target/debug/customers" || true
+ps aux | grep -E "target/release/(companies|customers|employees|cron)" | grep -v grep
-pkill -f "target/debug/photographers" || true
-pkill -f "target/debug/makeup_artists" || true
-pkill -f "target/debug/tutors" || true
-pkill -f "target/debug/developers" || true
-pkill -f "target/debug/video_editors" || true
-pkill -f "target/debug/graphic_designers" || true
-pkill -f "target/debug/social_media_managers" || true
-pkill -f "target/debug/fitness_trainers" || true
-pkill -f "target/debug/catering_services" || true
-pkill -f "target/debug/ugc_content_creators" || true
-pkill -f "target/debug/employees" || true
-
-apps=(
-  "gateway" "users" "companies" "job_seekers" "customers"
-  "photographers" "makeup_artists" "tutors" "developers" "video_editors"
-  "graphic_designers" "social_media_managers" "fitness_trainers" "catering_services"
-  "ugc_content_creators" "employees"
-)
-
-for app in "${apps[@]}"; do
-  if [[ -x "./target/debug/$app" ]]; then
-    echo "Starting $app..."
-    nohup ./target/debug/$app > "$app.log" 2>&1 &
-  else
-    echo "Skipping $app (binary not found at ./target/debug/$app)"
-  fi
-done
-
-echo "All available services booted up!"