Merge remote-tracking branch 'origin/master' into feature/zzhang/toki…

…o_v1

Cargo.toml

@@ -30,6 +30,7 @@ sqlx = { git = "https://github.com/launchbadge/sqlx.git", features=["runtime-tok
 chrono = { version = "0.4.19", features = ["serde"] }
 rust_decimal = { version = "1.10.0", features = ["postgres", "diesel", "bytes", "byteorder"] }
 rust_decimal_macros = "1.10.0"
+humantime-serde = "1.0"
 
 ttl_cache = "0.5.1"
 itertools = "0.10.0"

migrations/mock/20210115121017_kline.sql

@@ -1,10 +1,17 @@
 -- Add migration script here
 
-insert into trade_record select time, 'ETH_BTC', 1, random()*0.1, random()*100 
-from generate_series(timestamp '2020-01-02 00:00:00', timestamp '2020-01-03 00:00:00', interval '1 s') as time;
+insert into trade_record select time, 'ETH_USDT', 1, price, amount, price * amount, 'ask'
+from (select time, random()*300 + 1000 as price, random()*10 as amount 
+from generate_series(now() - interval '1 day', now() + interval '1 day', interval '1 s') as time) t;
 
-insert into trade_record select time, 'ETH_BTC', 2, random()*0.1, random()*100 
-from generate_series(timestamp '2020-01-02 00:00:00', timestamp '2020-01-03 00:00:00', interval '2 s') as time;
+insert into trade_record select time, 'ETH_USDT', 1, price, amount, price * amount, 'ask'
+from (select time, random()*200 + 1000 as price, random()*30 as amount 
+from generate_series(now() - interval '1 day', now() + interval '1 day', interval '3 s') as time) t;
 
-insert into trade_record select time, 'ETH_BTC', 3, random()*0.1, random()*100 
-from generate_series(timestamp '2020-01-02 00:00:00', timestamp '2020-01-03 00:00:00', interval '5 s') as time;
+insert into trade_record select time, 'BTC_USDT', 1, price, amount, price * amount, 'ask'
+from (select time, random()*200 + 1000 as price, random()*30 as amount 
+from generate_series(now() - interval '1 day', now() + interval '1 day', interval '3 s') as time) t;
+
+insert into trade_record select time, 'ETH_USDT', 1, price, amount, price * amount, 'ask'
+from (select time, random()*200 + 1000 as price, random()*100 as amount 
+from generate_series(now() - interval '1 day', now() + interval '1 day', interval '15 s') as time) t;

src/bin/restapi.rs

@@ -14,9 +14,9 @@ use std::sync::Mutex;
 use dingir_exchange::restapi;
 
 use restapi::personal_history::my_orders;
-use restapi::public_history::recent_trades;
-use restapi::state::AppState;
-use restapi::tradingview::{chart_config, history, symbols, unix_timestamp};
+use restapi::public_history::{order_trades, recent_trades};
+use restapi::state::{AppCache, AppState};
+use restapi::tradingview::{chart_config, history, symbols, ticker, unix_timestamp};
 use restapi::types::UserInfo;
 
 async fn ping(_req: HttpRequest, _data: web::Data<AppState>) -> impl Responder {
@@ -46,21 +46,28 @@ async fn main() -> std::io::Result<()> {
     let config_file = dotenv::var("CONFIG_FILE").unwrap();
     conf.merge(config_rs::File::with_name(&config_file)).unwrap();
 
+    let restapi_cfg: Option<config_rs::Value> = conf.get("restapi").ok();
+
     let dburl = conf.get_str("db_history").unwrap();
+    log::debug!("Prepared db connection: {}", &dburl);
+
     let user_map = web::Data::new(AppState {
         user_addr_map: Mutex::new(HashMap::new()),
         db: Pool::<Postgres>::connect(&dburl).await.unwrap(),
+        config: restapi_cfg.and_then(|v| v.try_into().ok()).unwrap_or_else(Default::default),
     });
 
-    log::debug!("Prepared db connection: {}", &dburl);
+    let workers = user_map.config.workers;
 
-    HttpServer::new(move || {
-        App::new().app_data(user_map.clone()).service(
+    let server = HttpServer::new(move || {
+        App::new().app_data(user_map.clone()).app_data(AppCache::new()).service(
             web::scope("/restapi")
                 .route("/ping", web::get().to(ping))
                 .route("/user/{id_or_addr}", web::get().to(get_user))
                 .route("/recenttrades/{market}", web::get().to(recent_trades))
+                .route("/ordertrades/{market}/{order_id}", web::get().to(order_trades))
                 .route("/closedorders/{market}/{user_id}", web::get().to(my_orders))
+                .route("/ticker_{ticker_inv}/{market}", web::get().to(ticker))
                 .service(
                     web::scope("/tradingview")
                         .route("/time", web::get().to(unix_timestamp))
@@ -69,8 +76,11 @@ async fn main() -> std::io::Result<()> {
                         .route("/history", web::get().to(history)),
                 ),
         )
-    })
-    .bind(("0.0.0.0", 50053))?
-    .run()
-    .await
+    });
+
+    let server = match workers {
+        Some(wr) => server.workers(wr),
+        None => server,
+    };
+    server.bind(("0.0.0.0", 50053))?.run().await
 }

src/restapi/config.rs

@@ -0,0 +1,35 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
+#[serde(default)]
+pub struct Trading {
+    #[serde(with = "humantime_serde")]
+    pub ticker_update_interval: std::time::Duration,
+    #[serde(with = "humantime_serde")]
+    pub ticker_interval: std::time::Duration,
+}
+
+impl Default for Trading {
+    fn default() -> Self {
+        Trading {
+            ticker_update_interval: std::time::Duration::from_secs(5),
+            ticker_interval: std::time::Duration::from_secs(86_400),
+        }
+    }
+}
+
+#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
+#[serde(default)]
+pub struct Settings {
+    pub workers: Option<usize>,
+    pub trading: Trading,
+}
+
+impl Default for Settings {
+    fn default() -> Self {
+        Settings {
+            workers: None,
+            trading: Default::default(),
+        }
+    }
+}

src/restapi/mod.rs

@@ -1,3 +1,4 @@
+pub mod config;
 pub mod errors;
 pub mod mock;
 pub mod personal_history;

src/restapi/public_history.rs

@@ -2,9 +2,16 @@ use actix_web::{web, HttpRequest, Responder};
 
 use actix_web::web::Json;
 
+use crate::models::{
+    self,
+    tablenames::{TRADEHISTORY, TRADERECORD},
+};
 use core::cmp::min;
 
-use super::{errors::RpcError, state::AppState};
+use super::{errors::RpcError, state::AppState, types};
+use models::{DecimalDbType, TimestampDbType};
+use rust_decimal::prelude::*;
+
 fn check_market_exists(_market: &str) -> bool {
     // TODO
     true
@@ -24,10 +31,69 @@ pub async fn recent_trades(req: HttpRequest, data: web::Data<AppState>) -> impl
     // and more suitable for fetching latest trades on a market.
     // models::TradeHistory is designed for a user to fetch his trades.
 
-    let trade_table = crate::models::tablenames::TRADERECORD;
-    let sql_query = format!("select * from {} where market = $1 order by time desc limit {}", trade_table, limit);
+    let sql_query = format!("select * from {} where market = $1 order by time desc limit {}", TRADERECORD, limit);
 
-    let trades: Vec<crate::models::TradeRecord> = sqlx::query_as(&sql_query).bind(market).fetch_all(&data.db).await?;
+    let trades: Vec<models::TradeRecord> = sqlx::query_as(&sql_query).bind(market).fetch_all(&data.db).await?;
 
     Ok(Json(trades))
 }
+
+#[derive(sqlx::FromRow, Debug, Clone)]
+struct QueriedTradeHistory {
+    pub time: TimestampDbType,
+    pub user_id: i32,
+    pub trade_id: i64,
+    pub order_id: i64,
+    pub price: DecimalDbType,
+    pub amount: DecimalDbType,
+    pub quote_amount: DecimalDbType,
+    pub fee: DecimalDbType,
+}
+
+#[cfg(sqlxverf)]
+fn sqlverf_ticker() {
+    sqlx::query_as!(
+        QueriedTradeHistory,
+        "select time, user_id, trade_id, order_id,
+        price, amount, quote_amount, fee
+        from trade_history where market = $1 and order_id = $2 
+        order by trade_id, time asc",
+        "USDT_ETH",
+        10000,
+    );
+}
+
+pub async fn order_trades(
+    app_state: web::Data<AppState>,
+    web::Path((market_name, order_id)): web::Path<(String, i64)>,
+) -> Result<Json<types::OrderTradeResult>, RpcError> {
+    log::debug!("order_trades market {} order_id {}", market_name, order_id);
+
+    let sql_query = format!(
+        "
+    select time, user_id, trade_id, order_id,
+    price, amount, quote_amount, fee
+    from {} where market = $1 and order_id = $2 
+    order by trade_id, time asc",
+        TRADEHISTORY
+    );
+
+    let trades: Vec<QueriedTradeHistory> = sqlx::query_as(&sql_query)
+        .bind(market_name)
+        .bind(order_id)
+        .fetch_all(&app_state.db)
+        .await?;
+
+    Ok(Json(types::OrderTradeResult {
+        trades: trades
+            .into_iter()
+            .map(|v| types::TradeRecord {
+                time: v.time.timestamp() as i32,
+                amount: v.amount.to_f32().unwrap_or(0.0),
+                quote_amount: v.quote_amount.to_f32().unwrap_or(0.0),
+                price: v.price.to_f32().unwrap_or(0.0),
+                fee: v.fee.to_f32().unwrap_or(0.0),
+            })
+            .collect(),
+    }))
+}

src/restapi/state.rs

@@ -1,9 +1,51 @@
-use super::types::UserInfo;
+use super::config::Settings;
+use super::types::{TickerResult, UserInfo};
 
 use sqlx::postgres::Postgres;
+use std::cell::RefCell;
 use std::collections::HashMap;
 use std::sync::Mutex;
 pub struct AppState {
     pub user_addr_map: Mutex<HashMap<String, UserInfo>>,
     pub db: sqlx::pool::Pool<Postgres>,
+    pub config: Settings,
+}
+
+#[derive(Debug)]
+pub struct TradingData {
+    pub ticker_ret_cache: HashMap<String, TickerResult>,
+}
+
+impl TradingData {
+    pub fn new() -> Self {
+        TradingData {
+            ticker_ret_cache: HashMap::new(),
+        }
+    }
+}
+
+impl Default for TradingData {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+//TLS storage
+#[derive(Debug)]
+pub struct AppCache {
+    pub trading: RefCell<TradingData>,
+}
+
+impl AppCache {
+    pub fn new() -> Self {
+        AppCache {
+            trading: TradingData::new().into(),
+        }
+    }
+}
+
+impl Default for AppCache {
+    fn default() -> Self {
+        Self::new()
+    }
 }