Adds manage scope to Open API. (#304)

fluidex · Sep 2, 2021 · 911ad53 · 911ad53
1 parent 7abc269
commit 911ad53
Show file tree

Hide file tree

Showing 5 changed files with 131 additions and 11 deletions.
diff --git a/src/bin/openapi.rs b/src/bin/openapi.rs
@@ -1,11 +1,12 @@
 use actix_web::{App, HttpServer};
+use dingir_exchange::openapi::manage::market;
 use dingir_exchange::openapi::personal_history::my_internal_txs;
 use dingir_exchange::openapi::public_history::{order_trades, recent_trades};
 use dingir_exchange::openapi::tradingview::{chart_config, history, search_symbols, symbols, ticker, unix_timestamp};
 use dingir_exchange::openapi::user::get_user;
 use dingir_exchange::restapi::state::{AppCache, AppState};
 use fluidex_common::non_blocking_tracing;
-use paperclip::actix::web::{self};
+use paperclip::actix::web::{self, HttpResponse};
 use paperclip::actix::{api_v2_operation, OpenApiExt};
 use sqlx::postgres::Postgres;
 use sqlx::Pool;
@@ -43,7 +44,9 @@ async fn main() -> std::io::Result<()> {
         config,
     });
 
-    HttpServer::new(move || {
+    let workers = user_map.config.workers;
+
+    let server = HttpServer::new(move || {
         App::new()
             .app_data(user_map.clone())
             .app_data(AppCache::new())
@@ -63,14 +66,29 @@ async fn main() -> std::io::Result<()> {
                             .route("/search", web::get().to(search_symbols))
                             .route("/symbols", web::get().to(symbols))
                             .route("/history", web::get().to(history)),
-                    ),
+                    )
+                    .service(if user_map.manage_channel.is_some() {
+                        web::scope("/manage").service(
+                            web::scope("/market")
+                                .route("/reload", web::post().to(market::reload))
+                                .route("/tradepairs", web::post().to(market::add_pair))
+                                .route("/assets", web::post().to(market::add_assets)),
+                        )
+                    } else {
+                        web::scope("/manage")
+                            .service(web::resource("/").to(|| HttpResponse::Forbidden().body(String::from("No manage endpoint"))))
+                    }),
             )
             .with_json_spec_at("/api/spec")
             .build()
-    })
-    .bind("0.0.0.0:50054")?
-    .run()
-    .await
+    });
+
+    let server = match workers {
+        Some(wr) => server.workers(wr),
+        None => server,
+    };
+
+    server.bind("0.0.0.0:50054")?.run().await
 }
 
 #[api_v2_operation]

diff --git a/src/config.rs b/src/config.rs
@@ -1,10 +1,11 @@
 use config_rs::{Config, File};
 use fluidex_common::rust_decimal::Decimal;
+use paperclip::actix::Apiv2Schema;
 use serde::de;
 use serde::{Deserialize, Serialize};
 use std::str::FromStr;
 
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default, Apiv2Schema)]
 #[serde(default)]
 pub struct Asset {
     pub id: String,
@@ -25,7 +26,7 @@ pub struct MarketUnit {
     pub prec: u32,
 }
 
-#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Apiv2Schema)]
 #[serde(default)]
 pub struct Market {
     pub name: String,

diff --git a/src/openapi/manage.rs b/src/openapi/manage.rs
@@ -0,0 +1,100 @@
+use crate::restapi::{state, types};
+use crate::storage;
+use actix_web::error::InternalError;
+use actix_web::http::StatusCode;
+use futures::future::OptionFuture;
+use orchestra::rpc::exchange::*;
+use paperclip::actix::api_v2_operation;
+use paperclip::actix::web;
+
+pub mod market {
+    use super::*;
+
+    async fn do_reload(app_state: &state::AppState) -> Result<&'static str, actix_web::Error> {
+        let mut rpc_cli = matchengine_client::MatchengineClient::new(app_state.manage_channel.as_ref().unwrap().clone());
+
+        if let Err(e) = rpc_cli.reload_markets(ReloadMarketsRequest { from_scratch: false }).await {
+            return Err(InternalError::new(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR).into());
+        }
+
+        Ok("done")
+    }
+
+    #[api_v2_operation]
+    pub async fn add_assets(
+        req: web::Json<types::NewAssetReq>,
+        app_state: web::Data<state::AppState>,
+    ) -> Result<&'static str, actix_web::Error> {
+        let assets_req = req.into_inner();
+
+        for asset in &assets_req.assets {
+            log::debug!("Add asset {:?}", asset);
+            if let Err(e) = storage::config::persist_asset_to_db(&app_state.db, asset, false).await {
+                return Err(InternalError::new(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR).into());
+            }
+        }
+
+        if !assets_req.not_reload {
+            do_reload(&app_state.into_inner()).await
+        } else {
+            Ok("done")
+        }
+    }
+
+    #[api_v2_operation]
+    pub async fn reload(app_state: web::Data<state::AppState>) -> Result<&'static str, actix_web::Error> {
+        do_reload(&app_state.into_inner()).await
+    }
+
+    #[api_v2_operation]
+    pub async fn add_pair(
+        req: web::Json<types::NewTradePairReq>,
+        app_state: web::Data<state::AppState>,
+    ) -> Result<&'static str, actix_web::Error> {
+        let trade_pair = req.into_inner();
+
+        if let Some(asset) = trade_pair.asset_base.as_ref() {
+            if asset.id != trade_pair.market.base {
+                return Err(InternalError::new("Base asset not match".to_owned(), StatusCode::BAD_REQUEST).into());
+            }
+        }
+
+        if let Some(asset) = trade_pair.asset_quote.as_ref() {
+            if asset.id != trade_pair.market.quote {
+                return Err(InternalError::new("Quote asset not match".to_owned(), StatusCode::BAD_REQUEST).into());
+            }
+        }
+
+        if let Some(Err(e)) = OptionFuture::from(
+            trade_pair
+                .asset_base
+                .as_ref()
+                .map(|base_asset| storage::config::persist_asset_to_db(&app_state.db, base_asset, false)),
+        )
+        .await
+        {
+            return Err(InternalError::new(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR).into());
+        }
+
+        if let Some(Err(e)) = OptionFuture::from(
+            trade_pair
+                .asset_quote
+                .as_ref()
+                .map(|quote_asset| storage::config::persist_asset_to_db(&app_state.db, quote_asset, false)),
+        )
+        .await
+        {
+            return Err(InternalError::new(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR).into());
+        }
+
+        if let Err(e) = storage::config::persist_market_to_db(&app_state.db, &trade_pair.market).await {
+            return Err(InternalError::new(e.to_string(), StatusCode::INTERNAL_SERVER_ERROR).into());
+        }
+
+        if !trade_pair.not_reload {
+            do_reload(&app_state.into_inner()).await
+        } else {
+            Ok("done")
+        }
+    }
+}
diff --git a/src/openapi/mod.rs b/src/openapi/mod.rs
@@ -1,3 +1,4 @@
+pub mod manage;
 pub mod personal_history;
 pub mod public_history;
 pub mod tradingview;

diff --git a/src/restapi/types.rs b/src/restapi/types.rs
@@ -58,14 +58,14 @@ pub struct OrderTradeResult {
     pub trades: Vec<MarketTrade>,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Apiv2Schema)]
 pub struct NewAssetReq {
     pub assets: Vec<Asset>,
     #[serde(default)]
     pub not_reload: bool,
 }
 
-#[derive(Serialize, Deserialize, Default)]
+#[derive(Serialize, Deserialize, Default, Apiv2Schema)]
 pub struct NewTradePairReq {
     pub market: Market,
     #[serde(default)]