From c570d7df678282f8beeb4c6ee4532e665612f9c8 Mon Sep 17 00:00:00 2001 From: Ashwin Kumar Sivakumar Date: Tue, 21 Jul 2026 04:19:56 +0530 Subject: [PATCH] chore: remove the dead apps/leads microservice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit apps/leads implemented its own, independent "lead request" system (POST /api/leads, /api/lead-requests/send, accept/reject) with a schema that never matched the live one (message vs remarks, no professional_user_id, accepted_at/rejected_at instead of resolved_at — see 20260721030000_create_lead_requests's commit message). Confirmed unreachable: the frontend's live flows use apps/customers' /api/customers/requirements and each profession's /leads/request (crates/contracts::profession_shared), never anything under apps/leads' own paths. Its /api/lead-requests/* endpoints weren't even reachable through the gateway (wrong prefix, never matched /api/leads or /api/admin/leads). Removed: - apps/leads/ entirely, and its Cargo.toml workspace membership - the `leads` docker-compose service, its LEADS_SERVICE_URL env var on gateway, and gateway's depends_on entry - the `leads` entry from both CI build matrices (.gitea/.forgejo) - gateway's leads_url field/routing branch — gateway no longer hard- requires LEADS_SERVICE_URL to boot (.expect() would have panicked once the service was gone); /api/admin/leads now falls through to the customers service, which already had a matching (previously shadowed) branch for it NOTE: this service may still have a live Deployment/Service in nxtgauge-gitops (a separate repo not touched here) — that manifest should be removed too, or the next deploy will reference an image that no CI job builds anymore. --- .forgejo/workflows/build.yaml | 1 - .gitea/workflows/build.yaml | 1 - Cargo.lock | 20 - Cargo.toml | 1 - apps/gateway/src/main.rs | 10 - apps/leads/Cargo.toml | 24 - apps/leads/Dockerfile | 28 -- apps/leads/src/lead_requests.rs | 837 -------------------------------- apps/leads/src/main.rs | 206 -------- docker-compose.yml | 13 - 10 files changed, 1141 deletions(-) delete mode 100644 apps/leads/Cargo.toml delete mode 100644 apps/leads/Dockerfile delete mode 100644 apps/leads/src/lead_requests.rs delete mode 100644 apps/leads/src/main.rs diff --git a/.forgejo/workflows/build.yaml b/.forgejo/workflows/build.yaml index f88e14e..81500a2 100644 --- a/.forgejo/workflows/build.yaml +++ b/.forgejo/workflows/build.yaml @@ -22,7 +22,6 @@ jobs: - users - companies - jobs - - leads - job-seekers - customers - payments diff --git a/.gitea/workflows/build.yaml b/.gitea/workflows/build.yaml index ab72fc2..30d2353 100644 --- a/.gitea/workflows/build.yaml +++ b/.gitea/workflows/build.yaml @@ -57,7 +57,6 @@ jobs: graphic-designers|apps/graphic_designers job-seekers|apps/job_seekers jobs|apps/jobs - leads|apps/leads makeup-artists|apps/makeup_artists payments|apps/payments photographers|apps/photographers diff --git a/Cargo.lock b/Cargo.lock index 25680f4..dcfce75 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2378,26 +2378,6 @@ dependencies = [ "spin 0.9.8", ] -[[package]] -name = "leads" -version = "0.1.0" -dependencies = [ - "anyhow", - "axum", - "chrono", - "contracts", - "jsonwebtoken 10.4.0", - "reqwest", - "serde", - "serde_json", - "sqlx", - "tokio", - "tower-http", - "tracing", - "tracing-subscriber", - "uuid", -] - [[package]] name = "leb128fmt" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 7019a3b..fdcefe5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,7 +26,6 @@ members = [ "apps/employees", "apps/payments", "apps/jobs", - "apps/leads", "crates/db-migrate" ] diff --git a/apps/gateway/src/main.rs b/apps/gateway/src/main.rs index 6d0039f..9a05add 100644 --- a/apps/gateway/src/main.rs +++ b/apps/gateway/src/main.rs @@ -20,7 +20,6 @@ struct Services { users_url: String, companies_url: String, jobs_url: String, - leads_url: String, job_seekers_url: String, customers_url: String, // ── 9 separate profession services ──────────────────────────────────── @@ -50,8 +49,6 @@ impl Services { .expect("COMPANIES_SERVICE_URL must be set"), jobs_url: std::env::var("JOBS_SERVICE_URL") .expect("JOBS_SERVICE_URL must be set"), - leads_url: std::env::var("LEADS_SERVICE_URL") - .expect("LEADS_SERVICE_URL must be set"), job_seekers_url: std::env::var("JOB_SEEKERS_SERVICE_URL") .expect("JOB_SEEKERS_SERVICE_URL must be set"), customers_url: std::env::var("CUSTOMERS_SERVICE_URL") @@ -147,12 +144,6 @@ impl Services { { Some(self.jobs_url.clone()) } - // Leads (separate service) - else if path.starts_with("/api/leads") - || path.starts_with("/api/admin/leads") - { - Some(self.leads_url.clone()) - } // Customers + Leads else if path.starts_with("/api/customers") || path.starts_with("/api/admin/customers") @@ -376,7 +367,6 @@ mod tests { users_url: "http://users".to_string(), companies_url: "http://companies".to_string(), jobs_url: "http://jobs".to_string(), - leads_url: "http://leads".to_string(), job_seekers_url: "http://job-seekers".to_string(), customers_url: "http://customers".to_string(), photographers_url: "http://photographers".to_string(), diff --git a/apps/leads/Cargo.toml b/apps/leads/Cargo.toml deleted file mode 100644 index 253d28a..0000000 --- a/apps/leads/Cargo.toml +++ /dev/null @@ -1,24 +0,0 @@ -[package] -name = "leads" -version = "0.1.0" -edition = "2021" - -[dependencies] -sqlx = { workspace = true } -axum = { workspace = true } -serde = { workspace = true } -serde_json = { workspace = true } -tokio = { workspace = true, features = ["full"] } -tracing = { workspace = true } -tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } -anyhow = { workspace = true } -uuid = { workspace = true } -chrono = { workspace = true } -tower-http = { version = "0.6", features = ["cors", "trace"] } -reqwest = { workspace = true } -contracts = { path = "../../crates/contracts" } -jsonwebtoken = "10" - -[[bin]] -name = "leads" -path = "src/main.rs" diff --git a/apps/leads/Dockerfile b/apps/leads/Dockerfile deleted file mode 100644 index a6137fa..0000000 --- a/apps/leads/Dockerfile +++ /dev/null @@ -1,28 +0,0 @@ -FROM ci.nxtgauge.com/admin/rust:alpine AS builder - -WORKDIR /usr/src/app - -RUN apk add --no-cache musl-dev pkgconfig openssl-dev openssl-libs-static && \ - rustup target add x86_64-unknown-linux-musl - -COPY Cargo.toml Cargo.lock ./ -COPY crates ./crates -COPY apps ./apps - -ENV RUSTFLAGS='-C target-feature=+crt-static' -ENV OPENSSL_STATIC=1 -ENV OPENSSL_DIR=/usr -RUN cargo build --release --bin leads --target x86_64-unknown-linux-musl - -FROM ci.nxtgauge.com/admin/alpine:latest AS runtime - -RUN apk add --no-cache ca-certificates - -RUN adduser -D -u 1000 appuser -WORKDIR /app - -COPY --from=builder /usr/src/app/target/x86_64-unknown-linux-musl/release/leads ./leads - -USER appuser - -CMD ["./leads"] diff --git a/apps/leads/src/lead_requests.rs b/apps/leads/src/lead_requests.rs deleted file mode 100644 index b155a46..0000000 --- a/apps/leads/src/lead_requests.rs +++ /dev/null @@ -1,837 +0,0 @@ -use crate::AppState; -use axum::{ - extract::{Path, Query, State}, - http::StatusCode, - response::IntoResponse, - routing::{get, post}, - Json, Router, -}; -use contracts::auth_middleware::AuthUser; -use serde::{Deserialize, Serialize}; -use sqlx::FromRow; -use std::sync::Arc; -use uuid::Uuid; - -#[derive(Debug, Deserialize)] -pub struct PaginationQuery { - pub page: Option, - pub limit: Option, - pub status: Option, -} - -#[derive(Debug, Deserialize)] -pub struct SendLeadRequestPayload { - pub lead_id: Uuid, - pub message: Option, -} - -#[derive(Debug, Deserialize)] -pub struct SendLeadRequestAiPayload { - pub lead_id: Uuid, - pub profession_key: String, -} - -#[derive(Debug, FromRow)] -pub struct LeadRequestRow { - pub id: Uuid, - pub reference_number: String, - pub lead_id: Uuid, - pub user_role_profile_id: Uuid, - pub customer_user_id: Uuid, - pub status: String, - pub tracecoins_reserved: i32, - pub message: Option, - pub expires_at: chrono::DateTime, - pub accepted_at: Option>, - pub rejected_at: Option>, - pub rejected_reason: Option, - pub created_at: chrono::DateTime, -} - -#[derive(Debug, Serialize)] -pub struct LeadRequestResponse { - pub id: Uuid, - pub reference_number: String, - pub lead_id: Uuid, - pub user_role_profile_id: Uuid, - pub customer_user_id: Uuid, - pub professional_name: Option, - pub professional_role: Option, - pub customer_name: Option, - pub lead_title: Option, - pub status: String, - pub tracecoins_reserved: i32, - pub message: Option, - pub expires_at: chrono::DateTime, - pub accepted_at: Option>, - pub rejected_at: Option>, - pub rejected_reason: Option, - pub created_at: chrono::DateTime, -} - -pub fn router() -> Router> { - Router::new() - .route("/", get(list_lead_requests)) - .route("/send", post(send_lead_request)) - .route("/send-ai", post(send_lead_request_ai)) - .route("/{id}/accept", post(accept_lead_request)) - .route("/{id}/reject", post(reject_lead_request)) - .route("/my-requests", get(my_requests)) - .route("/my-pending", get(my_pending_requests)) - .route("/customer/{lead_id}", get(get_customer_lead_requests)) -} - -fn lead_request_to_response(row: LeadRequestRow) -> LeadRequestResponse { - LeadRequestResponse { - id: row.id, - reference_number: row.reference_number, - lead_id: row.lead_id, - user_role_profile_id: row.user_role_profile_id, - customer_user_id: row.customer_user_id, - professional_name: None, - professional_role: None, - customer_name: None, - lead_title: None, - status: row.status, - tracecoins_reserved: row.tracecoins_reserved, - message: row.message, - expires_at: row.expires_at, - accepted_at: row.accepted_at, - rejected_at: row.rejected_at, - rejected_reason: row.rejected_reason, - created_at: row.created_at, - } -} - -async fn list_lead_requests( - auth: AuthUser, - State(state): State>, - Query(q): Query, -) -> impl IntoResponse { - let _ = auth; // authenticated; admin listing — no ownership filter - let page = q.page.unwrap_or(1); - let limit = q.limit.unwrap_or(20).min(100); - let offset = (page - 1) * limit; - - let requests = match sqlx::query_as::<_, LeadRequestRow>( - r#" - SELECT lr.* FROM lead_requests lr - WHERE ($1::text IS NULL OR lr.status = $1) - ORDER BY lr.created_at DESC - LIMIT $2 OFFSET $3 - "#, - ) - .bind(q.status.as_deref()) - .bind(limit) - .bind(offset) - .fetch_all(&state.pool) - .await - { - Ok(r) => r, - Err(e) => { - tracing::error!("list_lead_requests db error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - let requests: Vec = requests.into_iter().map(lead_request_to_response).collect(); - - (StatusCode::OK, Json(serde_json::json!({ - "data": requests, - "pagination": { "page": page, "limit": limit } - }))).into_response() -} - -async fn send_lead_request( - auth: AuthUser, - State(state): State>, - Json(payload): Json, -) -> impl IntoResponse { - let user_id = auth.user_id; - - let user_role_profile_id = match sqlx::query_scalar::<_, Uuid>( - "SELECT id FROM user_role_profiles WHERE user_id = $1 LIMIT 1" - ) - .bind(user_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(id)) => id, - Ok(None) => return (StatusCode::NOT_FOUND, "Professional profile not found. Please complete your profile first.").into_response(), - Err(e) => { - tracing::error!("send_lead_request profile lookup error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - let lead = match sqlx::query_as::<_, (Uuid, String, Uuid, String, i32)>( - "SELECT id, title, customer_user_id, status, COALESCE(current_acceptances, 0) FROM leads WHERE id = $1" - ) - .bind(payload.lead_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(l)) => l, - Ok(None) => return (StatusCode::NOT_FOUND, "Lead not found").into_response(), - Err(e) => { - tracing::error!("send_lead_request lead lookup error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - if lead.3 != "OPEN" { - return (StatusCode::BAD_REQUEST, "Lead is not open for requests").into_response(); - } - - if lead.4 >= 10 { - return (StatusCode::BAD_REQUEST, "Lead has reached maximum acceptances").into_response(); - } - - let duplicate = match sqlx::query_scalar::<_, Uuid>( - "SELECT id FROM lead_requests WHERE lead_id = $1 AND user_role_profile_id = $2 AND status IN ('PENDING', 'ACCEPTED')" - ) - .bind(payload.lead_id) - .bind(user_role_profile_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(_)) => true, - Ok(None) => false, - Err(e) => { - tracing::error!("send_lead_request duplicate check error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - if duplicate { - return (StatusCode::CONFLICT, "You have already sent a request for this lead").into_response(); - } - - let request_count: (i64,) = match sqlx::query_as( - "SELECT COUNT(*) FROM lead_requests WHERE lead_id = $1 AND status IN ('PENDING', 'ACCEPTED')" - ) - .bind(payload.lead_id) - .fetch_one(&state.pool) - .await - { - Ok(c) => c, - Err(e) => { - tracing::error!("send_lead_request request count error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - if request_count.0 >= 20 { - return (StatusCode::CONFLICT, "Lead has reached maximum requests").into_response(); - } - - let wallet = match sqlx::query_as::<_, (Uuid, i64)>( - "SELECT id, balance FROM tracecoin_wallets WHERE user_id = $1" - ) - .bind(user_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(w)) => w, - Ok(None) => return (StatusCode::BAD_REQUEST, "Wallet not found. Please contact support.").into_response(), - Err(e) => { - tracing::error!("send_lead_request wallet lookup error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - let tracecoins_cost = 25; - if wallet.1 < tracecoins_cost as i64 { - return (StatusCode::PAYMENT_REQUIRED, format!("Insufficient balance. You need at least {} Tracecoins.", tracecoins_cost)).into_response(); - } - - let expires_at = chrono::Utc::now() + chrono::Duration::hours(24); - - let result = sqlx::query_as::<_, LeadRequestRow>( - r#" - INSERT INTO lead_requests (lead_id, user_role_profile_id, customer_user_id, status, tracecoins_reserved, message, expires_at) - VALUES ($1, $2, $3, 'PENDING', $4, $5, $6) - RETURNING * - "# - ) - .bind(payload.lead_id) - .bind(user_role_profile_id) - .bind(lead.2) - .bind(tracecoins_cost) - .bind(&payload.message) - .bind(expires_at) - .fetch_one(&state.pool) - .await; - - match result { - Ok(req) => { - let _ = sqlx::query( - r#" - UPDATE tracecoin_wallets SET - balance = balance - $1, - reserved = COALESCE(reserved, 0) + $1, - updated_at = NOW() - WHERE user_id = $2 - "# - ) - .bind(tracecoins_cost as i64) - .bind(user_id) - .execute(&state.pool) - .await; - - let _ = sqlx::query( - r#" - INSERT INTO notifications (user_id, title, body, notification_type, reference_id) - VALUES ($1, $2, $3, $4, $5) - "# - ) - .bind(lead.2) - .bind("New Lead Request") - .bind("You have a new lead request. Please review and respond within 24 hours.") - .bind("LEAD_REQUEST") - .bind(req.id) - .execute(&state.pool) - .await; - - let response = lead_request_to_response(req); - (StatusCode::CREATED, Json(response)).into_response() - } - Err(e) => { - tracing::error!("send_lead_request insert error: {}", e); - (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response() - } - } -} - -async fn send_lead_request_ai( - auth: AuthUser, - State(state): State>, - Json(payload): Json, -) -> impl IntoResponse { - let user_id = auth.user_id; - - let lead = match sqlx::query_as::<_, (Uuid, String, String, String, String, Option, Option)>( - "SELECT id, title, description, location, profession_key, budget_min, budget_max FROM leads WHERE id = $1" - ) - .bind(payload.lead_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(l)) => l, - Ok(None) => return (StatusCode::NOT_FOUND, "Lead not found").into_response(), - Err(e) => { - tracing::error!("send_lead_request_ai lead lookup error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - if lead.4 != payload.profession_key { - return (StatusCode::BAD_REQUEST, "Lead profession does not match your profile").into_response(); - } - - let user_role_profile_id: Uuid = match sqlx::query_scalar::<_, Uuid>( - "SELECT id FROM user_role_profiles WHERE user_id = $1 LIMIT 1" - ) - .bind(user_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(id)) => id, - Ok(None) => return (StatusCode::NOT_FOUND, "Professional profile not found").into_response(), - Err(e) => { - tracing::error!("send_lead_request_ai profile lookup error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - let existing = match sqlx::query_scalar::<_, Uuid>( - "SELECT id FROM lead_requests WHERE lead_id = $1 AND user_role_profile_id = $2 AND status IN ('PENDING', 'ACCEPTED')" - ) - .bind(payload.lead_id) - .bind(user_role_profile_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(_)) => true, - Ok(None) => false, - Err(e) => { - tracing::error!("send_lead_request_ai duplicate check error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - if existing { - return (StatusCode::CONFLICT, "You have already sent a request for this lead").into_response(); - } - - let wallet = match sqlx::query_as::<_, (Uuid, i64)>( - "SELECT id, balance FROM tracecoin_wallets WHERE user_id = $1" - ) - .bind(user_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(w)) => w, - Ok(None) => return (StatusCode::BAD_REQUEST, "Wallet not found").into_response(), - Err(e) => { - tracing::error!("send_lead_request_ai wallet lookup error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - let tracecoins_cost = 30; - if wallet.1 < tracecoins_cost as i64 { - return (StatusCode::PAYMENT_REQUIRED, format!("Insufficient balance. You need {} Tracecoins.", tracecoins_cost)).into_response(); - } - - let budget = match (lead.5, lead.6) { - (Some(min), Some(max)) => format!("Budget: ₹{}-₹{}", min, max), - (Some(min), None) => format!("Budget: ₹{} onwards", min), - _ => "Budget: Not specified".to_string(), - }; - - let prompt = format!( - "You are a professional {} responding to a potential client's lead/request.\n\n\ - IMPORTANT: Do NOT include phone number, email, or any contact information in your response. \ - Clients pay to view contact details through the platform.\n\n\ - LEAD DETAILS:\n\ - Title: {}\n\ - Description: {}\n\ - Location: {}\n\ - {}\n\n\ - Write a professional, friendly message (max 150 words) expressing your interest and qualifications. \ - Mention relevant experience and ask any clarifying questions. Be concise and compelling.", - payload.profession_key.replace("_", " "), - lead.1, - lead.2, - lead.3, - budget - ); - - let ai_message = match generate_ai_message(&state.http_client, &state.ollama_base_url, &state.ollama_model, &prompt).await { - Ok(msg) => msg, - Err(e) => { - tracing::error!("AI message generation failed: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "AI generation failed").into_response(); - } - }; - - let expires_at = chrono::Utc::now() + chrono::Duration::hours(24); - let customer_id = lead.2.clone(); - - let result = sqlx::query_as::<_, LeadRequestRow>( - r#" - INSERT INTO lead_requests (lead_id, user_role_profile_id, customer_user_id, status, tracecoins_reserved, message, expires_at) - VALUES ($1, $2, $3, 'PENDING', $4, $5, $6) - RETURNING * - "# - ) - .bind(payload.lead_id) - .bind(user_role_profile_id) - .bind(&customer_id) - .bind(tracecoins_cost) - .bind(&ai_message) - .bind(expires_at) - .fetch_one(&state.pool) - .await; - - match result { - Ok(req) => { - let _ = sqlx::query( - r#" - UPDATE tracecoin_wallets SET - balance = balance - $1, - reserved = COALESCE(reserved, 0) + $1, - updated_at = NOW() - WHERE user_id = $2 - "# - ) - .bind(tracecoins_cost as i64) - .bind(user_id) - .execute(&state.pool) - .await; - - let _ = sqlx::query( - r#" - INSERT INTO notifications (user_id, title, body, notification_type, reference_id) - VALUES ($1, $2, $3, $4, $5) - "# - ) - .bind(&customer_id) - .bind("AI Auto-Respond Sent") - .bind("Your AI-assisted response has been sent to the customer.") - .bind("LEAD_REQUEST") - .bind(req.id) - .execute(&state.pool) - .await; - - let response = lead_request_to_response(req); - (StatusCode::CREATED, Json(response)).into_response() - } - Err(e) => { - tracing::error!("send_lead_request_ai insert error: {}", e); - (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response() - } - } -} - -async fn generate_ai_message( - client: &reqwest::Client, - base_url: &str, - model: &str, - prompt: &str, -) -> Result { - #[derive(Serialize)] - struct GenerateRequest<'a> { - model: &'a str, - prompt: String, - stream: bool, - } - - #[derive(Deserialize)] - struct GenerateResponse { - response: String, - } - - let url = format!("{}/api/generate", base_url.trim_end_matches('/')); - let req = GenerateRequest { - model, - prompt: prompt.to_string(), - stream: false, - }; - - let response = client - .post(&url) - .json(&req) - .send() - .await - .map_err(|e| format!("ollama request failed: {}", e))?; - - if !response.status().is_success() { - return Err(format!("ollama returned status: {}", response.status())); - } - - let result: GenerateResponse = response - .json() - .await - .map_err(|e| format!("failed to parse ollama response: {}", e))?; - - Ok(result.response.trim().to_string()) -} - -async fn accept_lead_request( - auth: AuthUser, - State(state): State>, - Path(id): Path, -) -> impl IntoResponse { - let customer_user_id = auth.user_id; - - let request = match sqlx::query_as::<_, LeadRequestRow>( - "SELECT * FROM lead_requests WHERE id = $1 AND status = 'PENDING'" - ) - .bind(id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(r)) => r, - Ok(None) => return (StatusCode::NOT_FOUND, "Lead request not found or already processed").into_response(), - Err(e) => { - tracing::error!("accept_lead_request fetch error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - if request.customer_user_id != customer_user_id { - return (StatusCode::FORBIDDEN, "You are not authorized to accept this request").into_response(); - } - - if request.expires_at < chrono::Utc::now() { - return (StatusCode::BAD_REQUEST, "This request has expired").into_response(); - } - - // Resolve the professional's user_id from their role profile - let professional_user_id = match sqlx::query_scalar::<_, Uuid>( - "SELECT user_id FROM user_role_profiles WHERE id = $1" - ) - .bind(request.user_role_profile_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(uid)) => uid, - Ok(None) => return (StatusCode::NOT_FOUND, "Professional profile not found").into_response(), - Err(e) => { - tracing::error!("accept_lead_request profile lookup error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - let lead_acceptances: (i32,) = match sqlx::query_as( - "SELECT COALESCE(current_acceptances, 0) FROM leads WHERE id = $1" - ) - .bind(request.lead_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(l)) => l, - Ok(None) => return (StatusCode::NOT_FOUND, "Lead not found").into_response(), - Err(e) => { - tracing::error!("accept_lead_request acceptances check error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - if lead_acceptances.0 >= 10 { - return (StatusCode::BAD_REQUEST, "Lead has reached maximum acceptances").into_response(); - } - - let _ = sqlx::query( - "UPDATE lead_requests SET status = 'ACCEPTED', accepted_at = NOW(), updated_at = NOW() WHERE id = $1" - ) - .bind(id) - .execute(&state.pool) - .await; - - let _ = sqlx::query( - "UPDATE leads SET current_acceptances = current_acceptances + 1, updated_at = NOW() WHERE id = $1" - ) - .bind(request.lead_id) - .execute(&state.pool) - .await; - - // Deduct the reserved coins from the professional's wallet - let _ = sqlx::query( - r#" - UPDATE tracecoin_wallets SET - reserved = reserved - $1, - updated_at = NOW() - WHERE user_id = $2 - "# - ) - .bind(request.tracecoins_reserved as i64) - .bind(professional_user_id) - .execute(&state.pool) - .await; - - if lead_acceptances.0 + 1 >= 10 { - let _ = sqlx::query("UPDATE leads SET status = 'CLOSED', updated_at = NOW() WHERE id = $1") - .bind(request.lead_id) - .execute(&state.pool) - .await; - } - - let _ = sqlx::query( - r#" - INSERT INTO notifications (user_id, title, body, notification_type, reference_id) - VALUES ($1, $2, $3, $4, $5) - "# - ) - .bind(professional_user_id) - .bind("Lead Request Accepted") - .bind("Your lead request has been accepted! Contact details have been shared.") - .bind("LEAD_REQUEST") - .bind(id) - .execute(&state.pool) - .await; - - (StatusCode::OK, Json(serde_json::json!({ - "message": "Lead request accepted successfully", - "contact_details_shared": true - }))).into_response() -} - -async fn reject_lead_request( - auth: AuthUser, - State(state): State>, - Path(id): Path, -) -> impl IntoResponse { - let customer_user_id = auth.user_id; - - let request = match sqlx::query_as::<_, LeadRequestRow>( - "SELECT * FROM lead_requests WHERE id = $1 AND status = 'PENDING'" - ) - .bind(id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(r)) => r, - Ok(None) => return (StatusCode::NOT_FOUND, "Lead request not found or already processed").into_response(), - Err(e) => { - tracing::error!("reject_lead_request fetch error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - if request.customer_user_id != customer_user_id { - return (StatusCode::FORBIDDEN, "You are not authorized to reject this request").into_response(); - } - - // Resolve the professional's user_id to refund their coins - let professional_user_id = match sqlx::query_scalar::<_, Uuid>( - "SELECT user_id FROM user_role_profiles WHERE id = $1" - ) - .bind(request.user_role_profile_id) - .fetch_optional(&state.pool) - .await - { - Ok(Some(uid)) => uid, - Ok(None) => return (StatusCode::NOT_FOUND, "Professional profile not found").into_response(), - Err(e) => { - tracing::error!("reject_lead_request profile lookup error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - let _ = sqlx::query( - "UPDATE lead_requests SET status = 'REJECTED', rejected_at = NOW(), rejected_reason = 'Rejected by customer', updated_at = NOW() WHERE id = $1" - ) - .bind(id) - .execute(&state.pool) - .await; - - // Refund reserved coins to the professional's wallet - let _ = sqlx::query( - r#" - UPDATE tracecoin_wallets SET - balance = balance + $1, - reserved = reserved - $1, - updated_at = NOW() - WHERE user_id = $2 - "# - ) - .bind(request.tracecoins_reserved as i64) - .bind(professional_user_id) - .execute(&state.pool) - .await; - - let _ = sqlx::query( - r#" - INSERT INTO notifications (user_id, title, body, notification_type, reference_id) - VALUES ($1, $2, $3, $4, $5) - "# - ) - .bind(professional_user_id) - .bind("Lead Request Rejected") - .bind("Your lead request was not accepted. Tracecoins have been refunded.") - .bind("LEAD_REQUEST") - .bind(id) - .execute(&state.pool) - .await; - - (StatusCode::OK, Json(serde_json::json!({ - "message": "Lead request rejected. Tracecoins refunded.", - "refunded": request.tracecoins_reserved - }))).into_response() -} - -async fn my_requests( - auth: AuthUser, - State(state): State>, - Query(q): Query, -) -> impl IntoResponse { - let user_id = auth.user_id; - let page = q.page.unwrap_or(1); - let limit = q.limit.unwrap_or(20).min(100); - let offset = (page - 1) * limit; - - let requests = match sqlx::query_as::<_, LeadRequestRow>( - r#" - SELECT lr.* FROM lead_requests lr - JOIN user_role_profiles urp ON urp.id = lr.user_role_profile_id - WHERE urp.user_id = $1 - AND ($2::text IS NULL OR lr.status = $2) - ORDER BY lr.created_at DESC - LIMIT $3 OFFSET $4 - "#, - ) - .bind(user_id) - .bind(q.status.as_deref()) - .bind(limit) - .bind(offset) - .fetch_all(&state.pool) - .await - { - Ok(r) => r, - Err(e) => { - tracing::error!("my_requests db error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - let requests: Vec = requests.into_iter().map(lead_request_to_response).collect(); - - (StatusCode::OK, Json(serde_json::json!({ - "data": requests, - "pagination": { "page": page, "limit": limit } - }))).into_response() -} - -async fn my_pending_requests( - auth: AuthUser, - State(state): State>, -) -> impl IntoResponse { - let user_id = auth.user_id; - - let requests = match sqlx::query_as::<_, LeadRequestRow>( - r#" - SELECT lr.* FROM lead_requests lr - WHERE lr.customer_user_id = $1 AND lr.status = 'PENDING' - ORDER BY lr.expires_at ASC - "# - ) - .bind(user_id) - .fetch_all(&state.pool) - .await - { - Ok(r) => r, - Err(e) => { - tracing::error!("my_pending_requests db error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - let requests: Vec = requests.into_iter().map(lead_request_to_response).collect(); - - (StatusCode::OK, Json(serde_json::json!({ - "data": requests - }))).into_response() -} - -async fn get_customer_lead_requests( - auth: AuthUser, - State(state): State>, - Path(lead_id): Path, - Query(q): Query, -) -> impl IntoResponse { - let user_id = auth.user_id; - let page = q.page.unwrap_or(1); - let limit = q.limit.unwrap_or(20).min(100); - let offset = (page - 1) * limit; - - let requests = match sqlx::query_as::<_, LeadRequestRow>( - r#" - SELECT lr.* FROM lead_requests lr - WHERE lr.lead_id = $1 AND lr.customer_user_id = $2 - ORDER BY lr.created_at DESC - LIMIT $3 OFFSET $4 - "#, - ) - .bind(lead_id) - .bind(user_id) - .bind(limit) - .bind(offset) - .fetch_all(&state.pool) - .await - { - Ok(r) => r, - Err(e) => { - tracing::error!("get_customer_lead_requests db error: {}", e); - return (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error").into_response(); - } - }; - - let requests: Vec = requests.into_iter().map(lead_request_to_response).collect(); - - (StatusCode::OK, Json(serde_json::json!({ - "data": requests, - "pagination": { "page": page, "limit": limit } - }))).into_response() -} diff --git a/apps/leads/src/main.rs b/apps/leads/src/main.rs deleted file mode 100644 index 376786b..0000000 --- a/apps/leads/src/main.rs +++ /dev/null @@ -1,206 +0,0 @@ -// retrigger-build-marker -#![allow(dead_code)] - -use axum::{ - extract::State, - http::StatusCode, - routing::{get, post, patch}, - Json, Router, -}; -use reqwest::Client; -use serde::{Deserialize, Serialize}; -use sqlx::PgPool; -use std::net::SocketAddr; -use std::sync::Arc; -use axum::http::HeaderValue; -use tower_http::cors::{Any, CorsLayer}; -use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; - -pub mod lead_requests; - -#[derive(Clone)] -pub struct AppState { - pub pool: PgPool, - pub http_client: reqwest::Client, - pub ollama_base_url: String, - pub ollama_model: String, -} - -#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)] -pub struct Lead { - pub id: uuid::Uuid, - pub title: String, - pub description: String, - pub location: String, - pub profession_key: String, - pub status: String, - pub created_at: chrono::DateTime, -} - -#[derive(Debug, Deserialize)] -pub struct CreateLead { - pub title: String, - pub description: String, - pub location: String, - pub profession_key: String, -} - -#[derive(Debug, Deserialize)] -pub struct UpdateLead { - pub title: Option, - pub description: Option, - pub location: Option, - pub status: Option, -} - -async fn list_leads(State(state): State>) -> Result>, StatusCode> { - let leads = sqlx::query_as::<_, Lead>( - "SELECT id, title, description, location, profession_key, status, created_at FROM leads ORDER BY created_at DESC" - ) - .fetch_all(&state.pool) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - - Ok(Json(leads)) -} - -async fn create_lead( - State(state): State>, - Json(payload): Json, -) -> Result, StatusCode> { - let lead = sqlx::query_as::<_, Lead>( - r#" - INSERT INTO leads (title, description, location, profession_key) - VALUES ($1, $2, $3, $4) - RETURNING id, title, description, location, profession_key, status, created_at - "# - ) - .bind(&payload.title) - .bind(&payload.description) - .bind(&payload.location) - .bind(&payload.profession_key) - .fetch_one(&state.pool) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - - Ok(Json(lead)) -} - -async fn get_lead( - State(state): State>, - axum::extract::Path(id): axum::extract::Path, -) -> Result, StatusCode> { - let lead = sqlx::query_as::<_, Lead>( - "SELECT id, title, description, location, profession_key, status, created_at FROM leads WHERE id = $1" - ) - .bind(id) - .fetch_optional(&state.pool) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? - .ok_or(StatusCode::NOT_FOUND)?; - - Ok(Json(lead)) -} - -async fn health() -> &'static str { - "Leads Service OK" -} - -async fn update_lead( - State(state): State>, - axum::extract::Path(id): axum::extract::Path, - Json(payload): Json, -) -> Result, StatusCode> { - let status = payload.status.as_deref().unwrap_or("OPEN"); - - let lead = sqlx::query_as::<_, Lead>( - r#" - UPDATE leads - SET title = COALESCE($1, title), - description = COALESCE($2, description), - location = COALESCE($3, location), - status = $4, - updated_at = NOW() - WHERE id = $5 - RETURNING id, title, description, location, profession_key, status, created_at - "#, - ) - .bind(&payload.title) - .bind(&payload.description) - .bind(&payload.location) - .bind(status) - .bind(id) - .fetch_optional(&state.pool) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? - .ok_or(StatusCode::NOT_FOUND)?; - - Ok(Json(lead)) -} - -#[tokio::main] -async fn main() { - tracing_subscriber::registry() - .with(tracing_subscriber::EnvFilter::new( - std::env::var("RUST_LOG").unwrap_or_else(|_| "info".into()), - )) - .with(tracing_subscriber::fmt::layer()) - .init(); - - let database_url = std::env::var("DATABASE_URL") - .expect("DATABASE_URL must be set"); - - let pool = sqlx::postgres::PgPoolOptions::new() - .max_connections(10) - .connect(&database_url) - .await - .expect("Failed to connect to database"); - - tracing::info!("Connected to database"); - - let state = Arc::new(AppState { - pool, - http_client: Client::new(), - ollama_base_url: std::env::var("OLLAMA_BASE_URL") - .expect("OLLAMA_BASE_URL must be set"), - ollama_model: std::env::var("OLLAMA_CHAT_MODEL") - .expect("OLLAMA_CHAT_MODEL must be set"), - }); - - let frontend_url: HeaderValue = std::env::var("FRONTEND_URL") - .unwrap_or_else(|_| "http://localhost:3000".to_string()) - .parse() - .expect("FRONTEND_URL is not a valid header value"); - let admin_url: HeaderValue = std::env::var("ADMIN_URL") - .unwrap_or_else(|_| "http://localhost:3001".to_string()) - .parse() - .expect("ADMIN_URL is not a valid header value"); - - let cors = CorsLayer::new() - .allow_origin([frontend_url, admin_url]) - .allow_methods(Any) - .allow_headers(Any); - - let app = Router::new() - .route("/health", get(health)) - .nest("/api", Router::new() - .route("/leads", get(list_leads)) - .route("/leads", post(create_lead)) - .route("/leads/{id}", get(get_lead)) - .route("/leads/{id}", patch(update_lead)) - .nest("/lead-requests", lead_requests::router()) - ) - .layer(cors) - .with_state(state); - - let port: u16 = std::env::var("PORT") - .unwrap_or_else(|_| "9118".to_string()) - .parse() - .expect("PORT must be a valid u16"); - let addr = SocketAddr::from(([0, 0, 0, 0], port)); - - tracing::info!("Leads service listening on {}", addr); - - let listener = tokio::net::TcpListener::bind(&addr).await.unwrap(); - axum::serve(listener, app).await.unwrap(); -} diff --git a/docker-compose.yml b/docker-compose.yml index 252f369..d47a260 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -49,7 +49,6 @@ services: USERS_SERVICE_URL: http://users:9101 COMPANIES_SERVICE_URL: http://companies:9102 JOBS_SERVICE_URL: http://jobs:9103 - LEADS_SERVICE_URL: http://leads:9118 JOB_SEEKERS_SERVICE_URL: http://job-seekers:9104 CUSTOMERS_SERVICE_URL: http://customers:9105 EMPLOYEES_SERVICE_URL: http://employees:9106 @@ -75,8 +74,6 @@ services: condition: service_started jobs: condition: service_started - leads: - condition: service_started job-seekers: condition: service_started customers: @@ -146,16 +143,6 @@ services: postgres: condition: service_healthy - leads: - platform: linux/amd64 - image: ci.nxtgauge.com/ashwin/nxtgauge-rust-leads:high-performance-latest - environment: - PORT: "9118" - DATABASE_URL: postgresql://nxtgauge:nxtgauge_dev@postgres:5432/nxtgauge_db - depends_on: - postgres: - condition: service_healthy - job-seekers: platform: linux/amd64 image: ci.nxtgauge.com/ashwin/nxtgauge-rust-job-seekers:high-performance-latest