nxtgauge-backend-rust/crates/contracts/src/profession_shared.rs

use axum::{
    extract::{Path, Query, State},
    http::StatusCode,
    response::IntoResponse,
    routing::{delete, get, patch, post},
    Json, Router,
};
use chrono::Utc;
use serde::Deserialize;
use uuid::Uuid;
use db::models::lead_request::{CreateLeadRequestPayload, LeadRequestRepository};
use db::models::professional::{
    CreatePortfolioItemPayload,
    CreateServicePayload,
    ProfessionalRepository,
    UpdatePortfolioItemPayload,
    UpdateServicePayload,
};
use db::models::requirement::RequirementRepository;
use crate::auth_middleware::AuthUser;
use crate::ProfessionState;

#[derive(Deserialize)]
pub struct PaginationQuery {
    pub page: Option<i64>,
    pub limit: Option<i64>,
}

#[derive(Deserialize)]
pub struct LeadRequestPayload {
    pub requirement_id: Uuid,
}

/// Build the shared Router that every profession service merges into its own Router.
/// `profession_key` must be a `'static str` matching the role key, e.g. `"PHOTOGRAPHER"`.
pub fn shared_routes(profession_key: &'static str) -> Router<ProfessionState> {
    Router::new()
        // ── Marketplace (Redis-cached) ────────────────────────────────────────
        .route(
            "/marketplace",
            get(move |State(state): State<ProfessionState>, Query(q): Query<PaginationQuery>| async move {
                let page  = q.page.unwrap_or(1);
                let limit = q.limit.unwrap_or(20);

                // Try cache first
                let mut redis = state.redis.clone();
                if let Ok(Some(cached)) = cache::jobs::get_marketplace(&mut redis, profession_key, page, limit).await {
                    if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(&cached) {
                        return (StatusCode::OK, Json(parsed)).into_response();
                    }
                }

                match ProfessionalRepository::get_marketplace(&state.pool, profession_key, page, limit).await {
                    Ok(items) => {
                        let body = serde_json::json!({
                            "data": items,
                            "pagination": { "page": page, "limit": limit }
                        });
                        // Write to cache (best-effort)
                        let _ = cache::jobs::set_marketplace(
                            &mut redis,
                            profession_key,
                            page,
                            limit,
                            &body.to_string(),
                        ).await;
                        (StatusCode::OK, Json(body)).into_response()
                    }
                    Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
                }
            }),
        )
        .route("/marketplace/{id}", get(get_requirement))
        // ── Lead Requests ────────────────────────────────────────────────────
        .route(
            "/leads/request",
            post(
                move |state: State<ProfessionState>, auth: AuthUser, payload: Json<LeadRequestPayload>| async move {
                    send_lead_request(state, auth, payload, profession_key).await
                },
            ),
        )
        .route("/leads/requests/me", get(my_requests))
        .route("/leads/requests/{id}", delete(cancel_request))
        .route("/leads/accepted/me", get(accepted_leads))
        .route("/leads/accepted/{id}", get(accepted_lead_detail))
        // ── Portfolio ────────────────────────────────────────────────────────
        .route("/portfolio/me", get(list_portfolio).post(create_portfolio_item))
        .route("/portfolio/me/{id}", patch(update_portfolio_item).delete(delete_portfolio_item))
        // ── Services ─────────────────────────────────────────────────────────
        .route("/services/me", get(list_services).post(create_service))
        .route("/services/me/{id}", patch(update_service).delete(delete_service))
        // ── Wallet ───────────────────────────────────────────────────────────
        .route("/wallet/me", get(wallet_balance))
        .route("/wallet/me/ledger", get(wallet_ledger))
        .route("/wallet/me/invoices", get(wallet_invoices))
        .route("/wallet/me/invoices/{id}", get(wallet_invoice_detail))
}

// ── Handlers ──────────────────────────────────────────────────────────────────

async fn get_requirement(
    State(state): State<ProfessionState>,
    _auth: AuthUser,
    Path(id): Path<Uuid>,
) -> impl IntoResponse {
    match RequirementRepository::get_by_id(&state.pool, id).await {
        Ok(Some(req)) if req.status == "OPEN" => (StatusCode::OK, Json(req)).into_response(),
        Ok(Some(_)) => (StatusCode::FORBIDDEN, "Requirement is not open").into_response(),
        Ok(None)    => (StatusCode::NOT_FOUND, "Requirement not found").into_response(),
        Err(e)      => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

async fn send_lead_request(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Json(payload): Json<LeadRequestPayload>,
    profession_key: &'static str,
) -> impl IntoResponse {
    let mut redis = state.redis.clone();

    // ── Rate limit: max 5 lead requests per hour per professional ─────────────
    let allowed = cache::rate_limit::check_lead(&mut redis, &auth.user_id.to_string())
        .await
        .unwrap_or(true);
    if !allowed {
        return (StatusCode::TOO_MANY_REQUESTS, "Too many lead requests. Try again later.").into_response();
    }

    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, "Professional profile not found").into_response(),
    };

    match is_professional_profile_approved(&state.pool, auth.user_id, profession_key).await {
        Ok(true) => {}
        Ok(false) => {
            return (
                StatusCode::FORBIDDEN,
                "Professional profile approval is required before sending lead requests",
            )
                .into_response()
        }
        Err(e) => {
            tracing::error!("Failed to check profile approval: {}", e);
            return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to validate profile approval").into_response();
        }
    }

    // ── Deduplication: one lead per requirement per professional (24 h) ────────
    let duplicate = cache::lead::is_duplicate(
        &mut redis,
        &prof.id.to_string(),
        &payload.requirement_id.to_string(),
    )
    .await
    .unwrap_or(false);

    if duplicate {
        return (StatusCode::CONFLICT, "You have already sent a lead request for this requirement").into_response();
    }

    let req = match RequirementRepository::get_by_id(&state.pool, payload.requirement_id).await {
        Ok(Some(r)) if r.status == "OPEN" => r,
        Ok(Some(_)) => return (StatusCode::BAD_REQUEST, "Requirement is not open").into_response(),
        _           => return (StatusCode::NOT_FOUND, "Requirement not found").into_response(),
    };

    if req.request_count >= 20 {
        return (StatusCode::CONFLICT, "Requirement reached max requests").into_response();
    }

    let wallet = match ProfessionalRepository::get_wallet(&state.pool, auth.user_id).await {
        Ok(w)  => w,
        Err(_) => return (StatusCode::BAD_REQUEST, "Wallet not found").into_response(),
    };

    if wallet.balance < 25 {
        return (StatusCode::PAYMENT_REQUIRED, "Insufficient Tracecoin balance").into_response();
    }

    let db_payload = CreateLeadRequestPayload {
        requirement_id:  req.id,
        professional_id: prof.id,
        expires_at: Utc::now() + chrono::Duration::hours(24),
    };

    match LeadRequestRepository::create(&state.pool, db_payload).await {
        Ok(lead) => {
            let reserved = ProfessionalRepository::try_reserve_tracecoins(
                &state.pool,
                auth.user_id,
                lead.tracecoins_reserved,
                lead.id,
            )
            .await;

            match reserved {
                Ok(true) => {}
                Ok(false) => {
                    let _ = LeadRequestRepository::update_status(&state.pool, lead.id, "CANCELLED").await;
                    return (StatusCode::PAYMENT_REQUIRED, "Insufficient Tracecoin balance").into_response();
                }
                Err(e) => {
                    let _ = LeadRequestRepository::update_status(&state.pool, lead.id, "CANCELLED").await;
                    return (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response();
                }
            }

            let _ = RequirementRepository::increment_request_count(&state.pool, req.id).await;
            // Mark dedup in Redis so this professional can't spam the same requirement
            let _ = cache::lead::mark_sent(
                &mut redis,
                &prof.id.to_string(),
                &payload.requirement_id.to_string(),
            )
            .await;
            (StatusCode::CREATED, Json(lead)).into_response()
        }
        Err(e) => {
            if e.to_string().contains("unique") {
                (StatusCode::CONFLICT, "Already requested this lead").into_response()
            } else {
                (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response()
            }
        }
    }
}

async fn is_professional_profile_approved(
    pool: &sqlx::PgPool,
    user_id: Uuid,
    profession_key: &str,
) -> Result<bool, sqlx::Error> {
    let status = match profession_key {
        "PHOTOGRAPHER" => {
            sqlx::query_scalar::<_, String>("SELECT status FROM photographer_profiles WHERE user_id = $1")
                .bind(user_id)
                .fetch_optional(pool)
                .await?
        }
        "MAKEUP_ARTIST" => {
            sqlx::query_scalar::<_, String>("SELECT status FROM makeup_artist_profiles WHERE user_id = $1")
                .bind(user_id)
                .fetch_optional(pool)
                .await?
        }
        "TUTOR" => {
            sqlx::query_scalar::<_, String>("SELECT status FROM tutor_profiles WHERE user_id = $1")
                .bind(user_id)
                .fetch_optional(pool)
                .await?
        }
        "DEVELOPER" => {
            sqlx::query_scalar::<_, String>("SELECT status FROM developer_profiles WHERE user_id = $1")
                .bind(user_id)
                .fetch_optional(pool)
                .await?
        }
        "VIDEO_EDITOR" => {
            sqlx::query_scalar::<_, String>("SELECT status FROM video_editor_profiles WHERE user_id = $1")
                .bind(user_id)
                .fetch_optional(pool)
                .await?
        }
        "GRAPHIC_DESIGNER" => {
            sqlx::query_scalar::<_, String>("SELECT status FROM graphic_designer_profiles WHERE user_id = $1")
                .bind(user_id)
                .fetch_optional(pool)
                .await?
        }
        "SOCIAL_MEDIA_MANAGER" => {
            sqlx::query_scalar::<_, String>("SELECT status FROM social_media_manager_profiles WHERE user_id = $1")
                .bind(user_id)
                .fetch_optional(pool)
                .await?
        }
        "FITNESS_TRAINER" => {
            sqlx::query_scalar::<_, String>("SELECT status FROM fitness_trainer_profiles WHERE user_id = $1")
                .bind(user_id)
                .fetch_optional(pool)
                .await?
        }
        "CATERING_SERVICES" => {
            sqlx::query_scalar::<_, String>("SELECT status FROM catering_service_profiles WHERE user_id = $1")
                .bind(user_id)
                .fetch_optional(pool)
                .await?
        }
        "UGC_CONTENT_CREATOR" => {
            sqlx::query_scalar::<_, String>("SELECT status FROM ugc_content_creator_profiles WHERE user_id = $1")
                .bind(user_id)
                .fetch_optional(pool)
                .await?
        }
        _ => None,
    };

    Ok(matches!(status.as_deref(), Some("APPROVED")))
}

async fn list_portfolio(State(state): State<ProfessionState>, auth: AuthUser) -> impl IntoResponse {
    match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(prof) => match ProfessionalRepository::get_portfolio(&state.pool, prof.id).await {
            Ok(items) => (StatusCode::OK, Json(serde_json::json!({ "data": items }))).into_response(),
            Err(e)    => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
        },
        Err(_) => (StatusCode::NOT_FOUND, "Professional profile not found").into_response(),
    }
}

async fn list_services(State(state): State<ProfessionState>, auth: AuthUser) -> impl IntoResponse {
    match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(prof) => match ProfessionalRepository::get_services(&state.pool, prof.id).await {
            Ok(items) => (StatusCode::OK, Json(serde_json::json!({ "data": items }))).into_response(),
            Err(e)    => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
        },
        Err(_) => (StatusCode::NOT_FOUND, "Professional profile not found").into_response(),
    }
}

async fn wallet_balance(State(state): State<ProfessionState>, auth: AuthUser) -> impl IntoResponse {
    match ProfessionalRepository::get_wallet(&state.pool, auth.user_id).await {
        Ok(w)  => (StatusCode::OK, Json(w)).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

// ── Lead request handlers ─────────────────────────────────────────────────────

#[derive(Deserialize)]
struct RequestsQuery {
    page: Option<i64>,
    limit: Option<i64>,
    status: Option<String>,
}

async fn my_requests(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Query(q): Query<RequestsQuery>,
) -> impl IntoResponse {
    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Professional profile not found" }))).into_response(),
    };

    let page  = q.page.unwrap_or(1).max(1);
    let limit = q.limit.unwrap_or(20).clamp(1, 100);
    let offset = (page - 1) * limit;

    #[derive(serde::Serialize, sqlx::FromRow)]
    struct RichLeadReq {
        #[serde(flatten)]
        #[sqlx(flatten)]
        lead: db::models::lead_request::LeadRequest,
        req_title: Option<String>,
        req_profession_key: Option<String>,
        req_location: Option<String>,
        req_budget: Option<i32>,
        customer_name: Option<String>,
        customer_email: Option<String>,
        customer_phone: Option<String>,
    }

    let rows = if let Some(ref status) = q.status {
        sqlx::query_as::<_, RichLeadReq>(
            r#"
            SELECT lr.*, r.title as req_title, r.profession_key as req_profession_key, r.location as req_location, r.budget as req_budget,
                   CASE WHEN lr.status = 'ACCEPTED' THEN u.full_name ELSE NULL END as customer_name,
                   CASE WHEN lr.status = 'ACCEPTED' THEN u.email ELSE NULL END as customer_email,
                   CASE WHEN lr.status = 'ACCEPTED' THEN u.phone ELSE NULL END as customer_phone
            FROM lead_requests lr
            LEFT JOIN requirements r ON r.id = lr.requirement_id
            LEFT JOIN customers c ON c.id = r.customer_id
            LEFT JOIN users u ON u.id = c.user_id
            WHERE lr.professional_id = $1 AND lr.status = $2 
            ORDER BY lr.requested_at DESC LIMIT $3 OFFSET $4
            "#
        )
        .bind(prof.id).bind(status).bind(limit).bind(offset).fetch_all(&state.pool).await
    } else {
        sqlx::query_as::<_, RichLeadReq>(
            r#"
            SELECT lr.*, r.title as req_title, r.profession_key as req_profession_key, r.location as req_location, r.budget as req_budget,
                   CASE WHEN lr.status = 'ACCEPTED' THEN u.full_name ELSE NULL END as customer_name,
                   CASE WHEN lr.status = 'ACCEPTED' THEN u.email ELSE NULL END as customer_email,
                   CASE WHEN lr.status = 'ACCEPTED' THEN u.phone ELSE NULL END as customer_phone
            FROM lead_requests lr
            LEFT JOIN requirements r ON r.id = lr.requirement_id
            LEFT JOIN customers c ON c.id = r.customer_id
            LEFT JOIN users u ON u.id = c.user_id
            WHERE lr.professional_id = $1 
            ORDER BY lr.requested_at DESC LIMIT $2 OFFSET $3
            "#
        )
        .bind(prof.id).bind(limit).bind(offset).fetch_all(&state.pool).await
    };

    let total: i64 = if let Some(ref status) = q.status {
        sqlx::query_scalar("SELECT COUNT(*) FROM lead_requests WHERE professional_id = $1 AND status = $2")
            .bind(prof.id).bind(status).fetch_one(&state.pool).await.unwrap_or(0)
    } else {
        sqlx::query_scalar("SELECT COUNT(*) FROM lead_requests WHERE professional_id = $1")
            .bind(prof.id).fetch_one(&state.pool).await.unwrap_or(0)
    };

    match rows {
        Ok(data) => {
            let total_pages = if total == 0 { 1 } else { (total + limit - 1) / limit };
            (StatusCode::OK, Json(serde_json::json!({
                "data": data,
                "pagination": { "page": page, "limit": limit, "total": total, "total_pages": total_pages }
            }))).into_response()
        }
        Err(e) => {
            tracing::error!("Failed to fetch lead requests: {}", e);
            (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "Failed to fetch requests" }))).into_response()
        }
    }
}

async fn cancel_request(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Path(id): Path<Uuid>,
) -> impl IntoResponse {
    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Professional profile not found" }))).into_response(),
    };

    let lead = match LeadRequestRepository::get_by_id(&state.pool, id).await {
        Ok(Some(l)) => l,
        Ok(None)    => return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Lead request not found" }))).into_response(),
        Err(e)      => return (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response(),
    };

    if lead.professional_id != prof.id {
        return (StatusCode::FORBIDDEN, Json(serde_json::json!({ "error": "Access denied" }))).into_response();
    }

    if lead.status != "PENDING" {
        return (StatusCode::BAD_REQUEST, Json(serde_json::json!({
            "error": format!("Cannot cancel a request with status '{}'", lead.status)
        }))).into_response();
    }

    // Release reserved Tracecoins back to balance
    if lead.tracecoins_reserved > 0 {
        let _ = ProfessionalRepository::try_release_reserved_tracecoins(
            &state.pool,
            auth.user_id,
            lead.tracecoins_reserved,
            lead.id,
            "LEAD_CANCELLED",
        ).await;
    }

    match LeadRequestRepository::update_status(&state.pool, id, "CANCELLED").await {
        Ok(updated) => (StatusCode::OK, Json(updated)).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": e.to_string() }))).into_response(),
    }
}

async fn accepted_leads(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Query(q): Query<PaginationQuery>,
) -> impl IntoResponse {
    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Professional profile not found" }))).into_response(),
    };

    let page  = q.page.unwrap_or(1).max(1);
    let limit = q.limit.unwrap_or(20).clamp(1, 100);
    let offset = (page - 1) * limit;

    // Join lead_requests → requirements → customers → users to get full contact info
    let rows = sqlx::query(
        r#"
        SELECT
            lr.id              AS lead_id,
            lr.status,
            lr.requested_at,
            lr.resolved_at,
            r.id               AS requirement_id,
            r.title            AS requirement_title,
            r.description      AS requirement_description,
            r.location         AS requirement_location,
            r.profession_key,
            u.full_name        AS customer_name,
            u.email            AS customer_email,
            u.phone            AS customer_phone
        FROM lead_requests lr
        INNER JOIN requirements r  ON r.id  = lr.requirement_id
        INNER JOIN customers   c  ON c.id  = r.customer_id
        INNER JOIN users       u  ON u.id  = c.user_id
        WHERE lr.professional_id = $1
          AND lr.status = 'ACCEPTED'
        ORDER BY lr.resolved_at DESC
        LIMIT $2 OFFSET $3
        "#
    )
    .bind(prof.id)
    .bind(limit)
    .bind(offset)
    .fetch_all(&state.pool)
    .await;

    let total: i64 = sqlx::query_scalar(
        "SELECT COUNT(*) FROM lead_requests WHERE professional_id = $1 AND status = 'ACCEPTED'"
    )
    .bind(prof.id)
    .fetch_one(&state.pool)
    .await
    .unwrap_or(0);

    match rows {
        Ok(rows) => {
            use sqlx::Row;
            let data: Vec<serde_json::Value> = rows.iter().map(|row| {
                serde_json::json!({
                    "lead_id":                   row.get::<Uuid, _>("lead_id"),
                    "status":                    row.get::<String, _>("status"),
                    "requested_at":              row.get::<chrono::DateTime<chrono::Utc>, _>("requested_at"),
                    "resolved_at":               row.try_get::<chrono::DateTime<chrono::Utc>, _>("resolved_at").ok(),
                    "requirement_id":            row.get::<Uuid, _>("requirement_id"),
                    "requirement_title":         row.get::<String, _>("requirement_title"),
                    "requirement_description":   row.try_get::<String, _>("requirement_description").ok(),
                    "requirement_location":      row.try_get::<String, _>("requirement_location").ok(),
                    "profession_key":            row.get::<String, _>("profession_key"),
                    "customer": {
                        "name":  row.try_get::<String, _>("customer_name").ok(),
                        "email": row.get::<String, _>("customer_email"),
                        "phone": row.try_get::<String, _>("customer_phone").ok(),
                    }
                })
            }).collect();

            let total_pages = if total == 0 { 1 } else { (total + limit - 1) / limit };
            (StatusCode::OK, Json(serde_json::json!({
                "data": data,
                "pagination": { "page": page, "limit": limit, "total": total, "total_pages": total_pages }
            }))).into_response()
        }
        Err(e) => {
            tracing::error!("Failed to fetch accepted leads: {}", e);
            (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "Failed to fetch accepted leads" }))).into_response()
        }
    }
}

async fn accepted_lead_detail(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Path(id): Path<Uuid>,
) -> impl IntoResponse {
    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Professional profile not found" }))).into_response(),
    };

    let row = sqlx::query(
        r#"
        SELECT
            lr.id              AS lead_id,
            lr.status,
            lr.tracecoins_reserved,
            lr.requested_at,
            lr.resolved_at,
            r.id               AS requirement_id,
            r.title            AS requirement_title,
            r.description      AS requirement_description,
            r.location         AS requirement_location,
            r.profession_key,
            r.custom_fields,
            u.full_name        AS customer_name,
            u.email            AS customer_email,
            u.phone            AS customer_phone
        FROM lead_requests lr
        INNER JOIN requirements r  ON r.id  = lr.requirement_id
        INNER JOIN customers   c  ON c.id  = r.customer_id
        INNER JOIN users       u  ON u.id  = c.user_id
        WHERE lr.id = $1
          AND lr.professional_id = $2
          AND lr.status = 'ACCEPTED'
        "#
    )
    .bind(id)
    .bind(prof.id)
    .fetch_optional(&state.pool)
    .await;

    match row {
        Ok(Some(row)) => {
            use sqlx::Row;
            let data = serde_json::json!({
                "lead_id":                   row.get::<Uuid, _>("lead_id"),
                "status":                    row.get::<String, _>("status"),
                "tracecoins_reserved":       row.get::<i32, _>("tracecoins_reserved"),
                "requested_at":              row.get::<chrono::DateTime<chrono::Utc>, _>("requested_at"),
                "resolved_at":               row.try_get::<chrono::DateTime<chrono::Utc>, _>("resolved_at").ok(),
                "requirement": {
                    "id":          row.get::<Uuid, _>("requirement_id"),
                    "title":       row.get::<String, _>("requirement_title"),
                    "description": row.try_get::<String, _>("requirement_description").ok(),
                    "location":    row.try_get::<String, _>("requirement_location").ok(),
                    "profession_key": row.get::<String, _>("profession_key"),
                    "custom_fields":  row.try_get::<serde_json::Value, _>("custom_fields").ok(),
                },
                "customer": {
                    "name":  row.try_get::<String, _>("customer_name").ok(),
                    "email": row.get::<String, _>("customer_email"),
                    "phone": row.try_get::<String, _>("customer_phone").ok(),
                }
            });
            (StatusCode::OK, Json(data)).into_response()
        }
        Ok(None) => (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": "Accepted lead not found" }))).into_response(),
        Err(e) => {
            tracing::error!("Failed to fetch accepted lead detail: {}", e);
            (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "Failed to fetch lead detail" }))).into_response()
        }
    }
}

async fn create_portfolio_item(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Json(payload): Json<CreatePortfolioItemPayload>,
) -> impl IntoResponse {
    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, "Professional profile not found").into_response(),
    };

    match ProfessionalRepository::create_portfolio_item(&state.pool, prof.id, payload).await {
        Ok(item) => (StatusCode::CREATED, Json(item)).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

async fn update_portfolio_item(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Path(id): Path<Uuid>,
    Json(payload): Json<UpdatePortfolioItemPayload>,
) -> impl IntoResponse {
    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, "Professional profile not found").into_response(),
    };

    match ProfessionalRepository::update_portfolio_item(&state.pool, prof.id, id, payload).await {
        Ok(Some(item)) => (StatusCode::OK, Json(item)).into_response(),
        Ok(None) => (StatusCode::NOT_FOUND, "Portfolio item not found").into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

async fn delete_portfolio_item(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Path(id): Path<Uuid>,
) -> impl IntoResponse {
    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, "Professional profile not found").into_response(),
    };

    match ProfessionalRepository::delete_portfolio_item(&state.pool, prof.id, id).await {
        Ok(true) => (StatusCode::OK, Json(serde_json::json!({ "message": "Deleted" }))).into_response(),
        Ok(false) => (StatusCode::NOT_FOUND, "Portfolio item not found").into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

async fn create_service(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Json(payload): Json<CreateServicePayload>,
) -> impl IntoResponse {
    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, "Professional profile not found").into_response(),
    };

    match ProfessionalRepository::create_service(&state.pool, prof.id, payload).await {
        Ok(svc) => (StatusCode::CREATED, Json(svc)).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

async fn update_service(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Path(id): Path<Uuid>,
    Json(payload): Json<UpdateServicePayload>,
) -> impl IntoResponse {
    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, "Professional profile not found").into_response(),
    };

    match ProfessionalRepository::update_service(&state.pool, prof.id, id, payload).await {
        Ok(Some(svc)) => (StatusCode::OK, Json(svc)).into_response(),
        Ok(None) => (StatusCode::NOT_FOUND, "Service not found").into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

async fn delete_service(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Path(id): Path<Uuid>,
) -> impl IntoResponse {
    let prof = match ProfessionalRepository::get_by_user_id(&state.pool, auth.user_id).await {
        Ok(p)  => p,
        Err(_) => return (StatusCode::NOT_FOUND, "Professional profile not found").into_response(),
    };

    match ProfessionalRepository::delete_service(&state.pool, prof.id, id).await {
        Ok(true) => (StatusCode::OK, Json(serde_json::json!({ "message": "Deleted" }))).into_response(),
        Ok(false) => (StatusCode::NOT_FOUND, "Service not found").into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

async fn wallet_ledger(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Query(q): Query<PaginationQuery>,
) -> impl IntoResponse {
    let page = q.page.unwrap_or(1);
    let limit = q.limit.unwrap_or(20);

    match ProfessionalRepository::list_wallet_ledger(&state.pool, auth.user_id, page, limit).await {
        Ok(items) => (StatusCode::OK, Json(serde_json::json!({
            "data": items,
            "pagination": { "page": page, "limit": limit }
        }))).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

async fn wallet_invoices(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Query(q): Query<PaginationQuery>,
) -> impl IntoResponse {
    let page = q.page.unwrap_or(1);
    let limit = q.limit.unwrap_or(20);

    match ProfessionalRepository::list_wallet_invoices(&state.pool, auth.user_id, page, limit).await {
        Ok(items) => (StatusCode::OK, Json(serde_json::json!({
            "data": items,
            "pagination": { "page": page, "limit": limit }
        }))).into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}

async fn wallet_invoice_detail(
    State(state): State<ProfessionState>,
    auth: AuthUser,
    Path(id): Path<Uuid>,
) -> impl IntoResponse {
    match ProfessionalRepository::get_invoice_by_id_for_user(&state.pool, auth.user_id, id).await {
        Ok(Some(inv)) => (StatusCode::OK, Json(inv)).into_response(),
        Ok(None) => (StatusCode::NOT_FOUND, "Invoice not found").into_response(),
        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
    }
}