Merge pull request #56 from Web3Novalabs/volume-analysis

Implement volume analysis endpoint for admin

Author: Akshola00

server/Cargo.lock

@@ -19,7 +19,7 @@ path = "src/main.rs"
 path = "src/lib.rs"
 
 [dependencies]
-axum = "0.8.4"
+axum = { version = "0.8.4", features = ["macros"] }
 serde = {version="1.0.219", features=["derive"]}
 tokio = { version = "1.45.1", features = ["full"] }
 starknet = "0.17.0"

server/Cargo.toml

@@ -12,6 +12,7 @@ pub mod libs {
 
 pub mod routes {
     pub mod admin;
+    pub mod analytics;
     pub mod crowd_funding;
     pub mod groups;
     pub mod health;

server/src/lib.rs

@@ -35,6 +35,7 @@ pub fn router(state: AppState) -> Router {
         .nest("/groups", routes::groups::router())
         .nest("/admin", routes::admin::router())
         .nest("/crowdfunding", routes::crowd_funding::router())
+        .nest("/analytics", routes::analytics::router())
         .with_state(state)
         .layer(cors)
         .fallback(|| async { (StatusCode::UNAUTHORIZED, "UNAUTHORIZED ORIGIN") })

server/src/libs/router.rs

@@ -0,0 +1,12 @@
+use utoipa_axum::{router::OpenApiRouter, routes};
+
+use crate::AppState;
+
+pub mod analytics_admin_routes;
+pub mod analytics_type;
+pub mod analytics_user_routes;
+pub fn router() -> OpenApiRouter<AppState> {
+    let admin_analytics = OpenApiRouter::new().routes(routes!(analytics_admin_routes::get_volume));
+
+    OpenApiRouter::new().nest("/admin", admin_analytics)
+}

server/src/routes/analytics.rs

@@ -0,0 +1,131 @@
+use axum::{
+    Json,
+    extract::{Query, State},
+};
+use bigdecimal::BigDecimal;
+
+use crate::{
+    AppState,
+    libs::{
+        auth::AdminUser,
+        error::{ApiError, map_sqlx_error},
+        utopia::ADMIN_TAG,
+    },
+    routes::analytics::analytics_type::{
+        AnalyticsItem, AnalyticsRow, VolumeDetails, VolumeRequest,
+    }, util::validate_address::validate_date,
+};
+use sqlx::Arguments;
+#[axum::debug_handler]
+#[utoipa::path(
+    method(get),
+    description = "Get volume processed",
+    path = "/volume",
+    responses(
+        (status = OK, description = "Success", body = VolumeDetails),
+        (status = INTERNAL_SERVER_ERROR, description = "Database Error | Failed to get crowd funding", body = ApiError),
+        (status = NOT_FOUND, description = "Crowd Funding Not Found", body = ApiError),
+        (status = BAD_REQUEST, description = "Invalid Crowd Funding Address", body = ApiError),
+    ),
+    params(
+        VolumeRequest,
+    ),
+    tag = ADMIN_TAG,
+    security(
+        ("bearer" = [])
+    )
+)]
+pub async fn get_volume(
+    State(state): State<AppState>,
+    AdminUser(_hello): AdminUser,
+    Query(params): Query<VolumeRequest>,
+) -> Result<Json<Vec<AnalyticsItem>>, ApiError> {
+    let mut base_sql_command = String::from("SELECT *
+        FROM (
+            SELECT token_address, token_amount, created_at::timestamptz AS time, 'group_tx_hashes' AS source
+            FROM group_tx_hashes
+
+            UNION ALL
+
+            SELECT token_address, amount AS token_amount, paid_at::timestamptz AS time, 'payments' AS source
+            FROM payments
+
+            UNION ALL
+
+            SELECT token_address, amount AS token_amount, created_at::timestamptz AS time, 'donations' AS source
+            FROM donations
+
+            UNION ALL
+
+            SELECT token_address, amount AS token_amount, created_at::timestamptz AS time, 'withdrawals' AS source
+            FROM withdrawals
+        ) AS all_tx");
+
+    let mut _i: i32 = 1;
+    let mut args: sqlx::postgres::PgArguments = sqlx::postgres::PgArguments::default();
+
+    if let (Some(from), Some(to)) = (params.from.as_ref(), params.to.as_ref()) {
+        validate_date(&from)?;
+        validate_date(&to)?;
+        base_sql_command.push_str(&format!(
+            " WHERE time BETWEEN ${}::timestamptz AND ${}::timestamptz",
+            _i,
+            _i + 1
+        ));
+        args.add(to.clone())
+            .map_err(|_| ApiError::Internal("Failed to add limit arg"))?;
+        args.add(from.clone())
+            .map_err(|_| ApiError::Internal("Failed to add limit arg"))?;
+        _i += 2;
+    } else if let Some(from) = params.from.as_ref() {
+        validate_date(&from)?;
+        base_sql_command.push_str(&format!(" WHERE time >= ${}::timestamptz", _i));
+        args.add(from.clone())
+            .map_err(|_| ApiError::Internal("Failed to add limit arg"))?;
+        _i += 1;
+    } else if let Some(to) = params.to.as_ref() {
+        validate_date(&to)?;
+        base_sql_command.push_str(&format!(" WHERE time <= ${}::timestamptz", _i));
+        args.add(to.clone())
+            .map_err(|_| ApiError::Internal("Failed to add limit arg"))?;
+        _i += 1;
+    }
+
+    if let Some(token) = params.token.as_ref() {
+        if _i == 1 {
+            base_sql_command.push_str(&format!(" WHERE token_address = ${}", _i));
+        } else {
+            base_sql_command.push_str(&format!(" AND token_address = ${}", _i));
+        }
+        args.add(token.clone())
+            .map_err(|_| ApiError::Internal("Failed to add limit arg"))?;
+        _i += 1;
+    }
+    base_sql_command.push_str(" ORDER BY time DESC;");
+
+    // println!("params: {:?}", &params);
+    // println!("SQL COMMAND: {}", &base_sql_command);
+    // println!("ARGS: {:?}", &args);
+
+    let rows: Vec<AnalyticsRow> = sqlx::query_as_with(&base_sql_command, args)
+        .fetch_all(&state.db)
+        .await
+        .map_err(|e| {
+            dbg!(&e);
+            map_sqlx_error(&e)
+        })?;
+
+    let items: Vec<AnalyticsItem> = rows
+        .into_iter()
+        .map(
+            |(token_address, token_amount, time, source)| AnalyticsItem {
+                token_address: token_address,
+                token_amount: BigDecimal::to_string(&token_amount),
+                time: time.to_rfc3339(),
+                source: source,
+            },
+        )
+        .collect();
+
+    Ok(Json(items))
+}

server/src/routes/analytics/analytics_admin_routes.rs

@@ -0,0 +1,36 @@
+use bigdecimal::BigDecimal;
+use serde::{Deserialize, Serialize};
+use sqlx::types::chrono;
+use utoipa::{IntoParams, ToSchema};
+
+#[derive(Debug, Serialize, ToSchema)]
+pub struct VolumeDetails {}
+
+#[derive(Debug, Serialize, ToSchema)]
+pub struct VolumeRecord {}
+
+#[derive(Debug, Serialize, ToSchema)]
+pub struct VolumeParams {
+    token: String,
+    perios: String,
+    total_inflow: String,
+    total_outflow: String,
+    net_volume: String,
+}
+
+#[derive(Debug, Serialize, ToSchema, Deserialize, IntoParams)]
+pub struct VolumeRequest {
+    pub from: Option<String>,
+    pub to: Option<String>,
+    pub token: Option<String>,
+}
+
+pub type AnalyticsRow = (String, BigDecimal, chrono::DateTime<chrono::Utc>, String);
+
+#[derive(Debug, Serialize, ToSchema)]
+pub struct AnalyticsItem {
+    pub token_address: String,
+    pub token_amount: String,
+    pub time: String,
+    pub source: String,
+}

server/src/routes/analytics/analytics_type.rs

@@ -0,0 +1 @@
+

server/src/routes/analytics/analytics_user_routes.rs

@@ -1,3 +1,4 @@
+use sqlx::types::chrono;
 use validator::ValidationError;
 
 use crate::libs::error::ApiError;
@@ -17,3 +18,10 @@ pub fn validate_address_api_err(address: &str) -> Result<(), ApiError> {
     .then_some(())
     .ok_or(ApiError::BadRequest("INVALID ADDRESS FORMAT"))
 }
+
+pub fn validate_date(date_str: &str) -> Result<(), ApiError> {
+    match chrono::NaiveDate::parse_from_str(date_str, "%Y-%m-%d") {
+        Ok(_) => Ok(()),
+        Err(_) => Err(ApiError::BadRequest("INVALID DATE FORMAT")),
+    }
+}