diff --git a/backend/src/admin.rs b/backend/src/admin.rs new file mode 100644 index 000000000..a72a0c846 --- /dev/null +++ b/backend/src/admin.rs @@ -0,0 +1,305 @@ +use axum::{ + extract::{Path, Query, State}, + http::StatusCode, + response::IntoResponse, + Json, +}; +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; +use tracing::{error, info}; +use uuid::Uuid; + +use crate::api::AppState; + +// --- Request/Response Types --- + +#[derive(Debug, Deserialize)] +pub struct UsersQuery { + pub page: Option, + pub page_size: Option, + pub kyc_status: Option, +} + +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct UserRow { + pub id: Uuid, + pub wallet_address: String, + pub kyc_status: String, + pub created_at: DateTime, +} + +#[derive(Debug, Serialize)] +pub struct UsersListResponse { + pub data: Vec, + pub page: i64, + pub page_size: i64, + pub total: i64, +} + +#[derive(Debug, Deserialize)] +pub struct UpdateKycRequest { + pub kyc_status: String, + pub admin_notes: Option, +} + +#[derive(Debug, Serialize)] +pub struct UpdateKycResponse { + pub wallet_address: String, + pub kyc_status: String, + pub updated_at: DateTime, +} + +#[derive(Serialize)] +struct ApiError { + error: String, +} + +// --- Admin Handlers --- + +/// GET /api/admin/users +/// Returns a paginated list of registered users and their KYC statuses +pub async fn list_users( + State(state): State>, + Query(query): Query, +) -> impl IntoResponse { + // Pagination defaults + let page = query.page.unwrap_or(1).max(1); + let page_size = query.page_size.unwrap_or(20).clamp(1, 100); + let offset = (page - 1) * page_size; + + info!( + page = page, + page_size = page_size, + kyc_status_filter = ?query.kyc_status, + "Admin: Listing users" + ); + + // Count total users matching the filter + let total: i64 = match sqlx::query_scalar( + r#" + SELECT COUNT(*) + FROM users + WHERE ($1::text IS NULL OR kyc_status::text = $1) + "#, + ) + .bind(query.kyc_status.as_deref()) + .fetch_one(&state.db_pool) + .await + { + Ok(count) => count, + Err(e) => { + error!(error = %e, "Admin: Failed to count users"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ApiError { + error: "Failed to retrieve user count".to_string(), + }), + ) + .into_response(); + } + }; + + // Fetch paginated users + let users: Vec = match sqlx::query_as::<_, UserRow>( + r#" + SELECT id, wallet_address, kyc_status::text as kyc_status, created_at + FROM users + WHERE ($1::text IS NULL OR kyc_status::text = $1) + ORDER BY created_at DESC + LIMIT $2 OFFSET $3 + "#, + ) + .bind(query.kyc_status.as_deref()) + .bind(page_size) + .bind(offset) + .fetch_all(&state.db_pool) + .await + { + Ok(rows) => rows, + Err(e) => { + error!(error = %e, "Admin: Failed to fetch users"); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ApiError { + error: "Failed to retrieve users".to_string(), + }), + ) + .into_response(); + } + }; + + info!( + total_users = total, + returned_count = users.len(), + "Admin: Successfully retrieved users" + ); + + ( + StatusCode::OK, + Json(UsersListResponse { + data: users, + page, + page_size, + total, + }), + ) + .into_response() +} + +/// PUT /api/admin/users/:address/kyc +/// Allows overriding KYC status (Approve/Reject) directly in PostgreSQL +pub async fn update_user_kyc( + State(state): State>, + Path(address): Path, + Json(payload): Json, +) -> impl IntoResponse { + // Validate KYC status + let valid_statuses = ["pending", "approved", "rejected", "submitted"]; + if !valid_statuses.contains(&payload.kyc_status.as_str()) { + error!( + wallet_address = %address, + attempted_status = %payload.kyc_status, + "Admin: Invalid KYC status attempted" + ); + return ( + StatusCode::BAD_REQUEST, + Json(ApiError { + error: format!( + "Invalid KYC status. Must be one of: {}", + valid_statuses.join(", ") + ), + }), + ) + .into_response(); + } + + info!( + wallet_address = %address, + new_status = %payload.kyc_status, + admin_notes = ?payload.admin_notes, + "Admin: Updating user KYC status" + ); + + // Check if user exists + let user_exists: bool = match sqlx::query_scalar( + r#" + SELECT EXISTS(SELECT 1 FROM users WHERE wallet_address = $1) + "#, + ) + .bind(&address) + .fetch_one(&state.db_pool) + .await + { + Ok(exists) => exists, + Err(e) => { + error!( + wallet_address = %address, + error = %e, + "Admin: Database error checking user existence" + ); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ApiError { + error: "Database error".to_string(), + }), + ) + .into_response(); + } + }; + + if !user_exists { + error!( + wallet_address = %address, + "Admin: Attempted to update KYC for non-existent user" + ); + return ( + StatusCode::NOT_FOUND, + Json(ApiError { + error: "User not found".to_string(), + }), + ) + .into_response(); + } + + // Update KYC status + let updated_at = Utc::now(); + match sqlx::query( + r#" + UPDATE users + SET kyc_status = $1::kyc_status + WHERE wallet_address = $2 + "#, + ) + .bind(&payload.kyc_status) + .bind(&address) + .execute(&state.db_pool) + .await + { + Ok(result) => { + if result.rows_affected() == 0 { + error!( + wallet_address = %address, + "Admin: No rows affected when updating KYC status" + ); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ApiError { + error: "Failed to update KYC status".to_string(), + }), + ) + .into_response(); + } + } + Err(e) => { + error!( + wallet_address = %address, + kyc_status = %payload.kyc_status, + error = %e, + "Admin: Failed to update KYC status" + ); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(ApiError { + error: format!("Failed to update KYC status: {}", e), + }), + ) + .into_response(); + } + } + + // Log the admin action for auditing + info!( + wallet_address = %address, + kyc_status = %payload.kyc_status, + admin_notes = ?payload.admin_notes, + updated_at = %updated_at, + "Admin: Successfully updated user KYC status" + ); + + // Broadcast KYC update event via WebSocket + let event = crate::ws::KycUpdateEvent { + wallet_address: address.clone(), + kyc_status: payload.kyc_status.clone(), + event_type: "admin_override".to_string(), + }; + + if let Err(e) = state.kyc_tx.send(event) { + error!( + error = %e, + wallet_address = %address, + "Admin: Failed to broadcast KYC update event" + ); + // Don't fail the request if broadcast fails + } + + ( + StatusCode::OK, + Json(UpdateKycResponse { + wallet_address: address, + kyc_status: payload.kyc_status, + updated_at, + }), + ) + .into_response() +} diff --git a/backend/src/api.rs b/backend/src/api.rs index 7ba8df428..ca3e7b80a 100644 --- a/backend/src/api.rs +++ b/backend/src/api.rs @@ -4,7 +4,7 @@ use axum::{ http::{header::HeaderName, HeaderValue}, middleware::from_fn, response::IntoResponse, - routing::{get, post}, + routing::{get, post, put}, Json, Router, }; use chrono::{DateTime, Utc}; @@ -15,7 +15,8 @@ use tower_http::cors::{Any, CorsLayer}; use tracing::error; use uuid::Uuid; -use crate::auth::signature_auth_middleware; +use crate::admin::{list_users, update_user_kyc}; +use crate::auth::{jwt_auth_middleware, signature_auth_middleware}; use crate::cache::PlanCache; use crate::kyc_webhook::kyc_webhook_handler; use crate::metrics::{latency_middleware, metrics_handler}; @@ -122,6 +123,12 @@ pub fn create_router(state: Arc) -> Router { .route("/api/plans/payout", post(trigger_payout)) .route_layer(from_fn(signature_auth_middleware)); + // Admin routes requiring JWT authentication + let admin_routes = Router::new() + .route("/api/admin/users", get(list_users)) + .route("/api/admin/users/:address/kyc", put(update_user_kyc)) + .route_layer(from_fn(jwt_auth_middleware)); + // Public or admin routes let public_routes = Router::new() .route("/api/plans", get(get_plans)) @@ -136,6 +143,7 @@ pub fn create_router(state: Arc) -> Router { Router::new() .merge(user_routes) + .merge(admin_routes) .merge(public_routes) .route("/metrics", get(metrics_handler)) .layer(from_fn(latency_middleware)) diff --git a/backend/src/lib.rs b/backend/src/lib.rs index cf3cc970c..cb0f6b591 100644 --- a/backend/src/lib.rs +++ b/backend/src/lib.rs @@ -1,3 +1,4 @@ +pub mod admin; pub mod api; pub mod auth; pub mod cache;