Base routing (#33)

* first draft of routing

* result of rebase

* fix: normalize values before routing

* fix: routing condition

* fix: type

---------

Co-authored-by: 0xevolve <Artevolve@yahoo.com>

Author: graniet

pragma-entities/src/models/currency.rs

@@ -1,3 +1,6 @@
+use super::DieselResult;
+use crate::schema::currencies;
+use diesel::{ExpressionMethods, PgConnection, QueryDsl, RunQueryDsl};
 use utoipa::ToSchema;
 use uuid::Uuid;
 
@@ -9,3 +12,16 @@ pub struct Currency {
     pub is_abstract: bool,
     pub ethereum_address: String,
 }
+
+impl Currency {
+    pub fn get_all(conn: &mut PgConnection) -> DieselResult<Vec<String>> {
+        currencies::table.select(currencies::name).get_results(conn)
+    }
+
+    pub fn get_abstract_all(conn: &mut PgConnection) -> DieselResult<Vec<String>> {
+        currencies::table
+            .select(currencies::name)
+            .filter(currencies::abstract_.eq(true))
+            .get_results(conn)
+    }
+}

pragma-entities/src/models/entry.rs

@@ -47,6 +47,13 @@ impl Entry {
             .get_results(conn)
     }
 
+    pub fn exists(conn: &mut PgConnection, pair_id: String) -> DieselResult<bool> {
+        diesel::select(diesel::dsl::exists(
+            entries::table.filter(entries::pair_id.eq(pair_id)),
+        ))
+        .get_result(conn)
+    }
+
     pub fn get_by_pair_id(conn: &mut PgConnection, pair_id: String) -> DieselResult<Entry> {
         entries::table
             .filter(entries::pair_id.eq(pair_id))

pragma-node/src/handlers/entries/get_entry.rs

@@ -46,11 +46,12 @@ pub async fn get_entry(
         Interval::OneMinute
     };
 
+    let is_routing = params.routing.unwrap_or(false);
+
     // Validate given timestamp
     if timestamp > now {
         return Err(EntryError::InvalidTimestamp);
     }
-
     // Mock strk/eth pair
     if pair_id == "STRK/ETH" {
         return Ok(Json(GetEntryResponse {
@@ -62,14 +63,15 @@ pub async fn get_entry(
         }));
     }
 
-    let entry =
-        entry_repository::get_median_price(&state.pool, pair_id.clone(), interval, timestamp)
-            .await
-            .map_err(|db_error| to_entry_error(db_error, &pair_id))?;
-
-    let decimals = entry_repository::get_decimals(&state.pool, &pair_id)
-        .await
-        .map_err(|db_error| to_entry_error(db_error, &pair_id))?;
+    let (entry, decimals) = entry_repository::routing(
+        &state.pool,
+        pair_id.clone(),
+        interval,
+        timestamp,
+        is_routing,
+    )
+    .await
+    .map_err(|e| to_entry_error(e, &pair_id))?;
 
     Ok(Json(adapt_entry_to_entry_response(
         pair_id, &entry, decimals,
@@ -97,7 +99,7 @@ fn adapt_entry_to_entry_response(
     }
 }
 
-fn to_entry_error(error: InfraError, pair_id: &String) -> EntryError {
+pub(crate) fn to_entry_error(error: InfraError, pair_id: &String) -> EntryError {
     match error {
         InfraError::InternalServerError => EntryError::InternalServerError,
         InfraError::NotFound => EntryError::NotFound(pair_id.to_string()),

pragma-node/src/handlers/entries/mod.rs

@@ -68,7 +68,7 @@ pub struct GetVolatilityResponse {
 /// Query parameters structs
 
 // Define an enum for the allowed intervals
-#[derive(Default, Debug, Deserialize, ToSchema)]
+#[derive(Default, Debug, Deserialize, ToSchema, Clone, Copy)]
 pub enum Interval {
     #[serde(rename = "1min")]
     #[default]
@@ -83,13 +83,15 @@ pub enum Interval {
 pub struct GetEntryParams {
     pub timestamp: Option<u64>,
     pub interval: Option<Interval>,
+    pub routing: Option<bool>,
 }
 
 impl Default for GetEntryParams {
     fn default() -> Self {
         Self {
             timestamp: Some(chrono::Utc::now().timestamp_millis() as u64),
             interval: Some(Interval::default()),
+            routing: Some(false),
         }
     }
 }

pragma-node/src/infra/repositories/entry_repository.rs

@@ -8,10 +8,11 @@ use pragma_entities::dto;
 use pragma_entities::{
     error::{adapt_infra_error, InfraError},
     schema::currencies,
-    Entry, NewEntry,
+    Currency, Entry, NewEntry,
 };
 
 use crate::handlers::entries::Interval;
+use crate::utils::{convert_via_quote, normalize_to_decimals};
 
 #[derive(Deserialize)]
 #[allow(unused)]
@@ -81,6 +82,139 @@ pub struct MedianEntryRaw {
     pub num_sources: i64,
 }
 
+pub async fn routing(
+    pool: &deadpool_diesel::postgres::Pool,
+    pair_id: String,
+    interval: Interval,
+    timestamp: u64,
+    is_routing: bool,
+) -> Result<(MedianEntry, u32), InfraError> {
+    if pair_id_exist(pool, pair_id.clone()).await? || !is_routing {
+        return get_price_decimals(pool, pair_id, interval, timestamp).await;
+    }
+
+    let [base, quote]: [&str; 2] = pair_id
+        .split('/')
+        .collect::<Vec<_>>()
+        .try_into()
+        .map_err(|_| InfraError::InternalServerError)?;
+
+    match find_alternative_pair_price(pool, base, quote, interval, timestamp).await {
+        Ok(result) => Ok(result),
+        Err(_) => Err(InfraError::NotFound),
+    }
+}
+
+fn calculate_rebased_price(
+    base_result: (MedianEntry, u32),
+    quote_result: (MedianEntry, u32),
+) -> Result<(MedianEntry, u32), InfraError> {
+    let (base_entry, base_decimals) = base_result;
+    let (quote_entry, quote_decimals) = quote_result;
+
+    if quote_entry.median_price == BigDecimal::from(0) {
+        return Err(InfraError::InternalServerError);
+    }
+
+    let (rebase_price, decimals) = if base_decimals < quote_decimals {
+        let normalized_base_price =
+            normalize_to_decimals(base_entry.median_price, base_decimals, quote_decimals);
+        (
+            convert_via_quote(
+                normalized_base_price,
+                quote_entry.median_price,
+                quote_decimals,
+            )?,
+            quote_decimals,
+        )
+    } else {
+        let normalized_quote_price =
+            normalize_to_decimals(quote_entry.median_price, quote_decimals, base_decimals);
+        (
+            convert_via_quote(
+                base_entry.median_price,
+                normalized_quote_price,
+                base_decimals,
+            )?,
+            base_decimals,
+        )
+    };
+    let min_timestamp = std::cmp::max(base_entry.time.timestamp(), quote_entry.time.timestamp());
+    let num_sources = std::cmp::max(base_entry.num_sources, quote_entry.num_sources);
+    let new_timestamp =
+        NaiveDateTime::from_timestamp_opt(min_timestamp, 0).ok_or(InfraError::InvalidTimeStamp)?;
+
+    let median_entry = MedianEntry {
+        time: new_timestamp,
+        median_price: rebase_price,
+        num_sources,
+    };
+
+    Ok((median_entry, decimals))
+}
+
+async fn find_alternative_pair_price(
+    pool: &deadpool_diesel::postgres::Pool,
+    base: &str,
+    quote: &str,
+    interval: Interval,
+    timestamp: u64,
+) -> Result<(MedianEntry, u32), InfraError> {
+    let conn = pool.get().await.map_err(adapt_infra_error)?;
+
+    let alternative_currencies = conn
+        .interact(Currency::get_abstract_all)
+        .await
+        .map_err(adapt_infra_error)?
+        .map_err(adapt_infra_error)?;
+
+    for alt_currency in alternative_currencies {
+        let base_alt_pair = format!("{}/{}", base, alt_currency);
+        let alt_quote_pair = format!("{}/{}", alt_currency, quote);
+
+        if pair_id_exist(pool, base_alt_pair.clone()).await?
+            && pair_id_exist(pool, alt_quote_pair.clone()).await?
+        {
+            let base_alt_result =
+                get_price_decimals(pool, base_alt_pair, interval, timestamp).await?;
+            let alt_quote_result =
+                get_price_decimals(pool, alt_quote_pair, interval, timestamp).await?;
+
+            return calculate_rebased_price(base_alt_result, alt_quote_result);
+        }
+    }
+
+    Err(InfraError::NotFound)
+}
+
+async fn pair_id_exist(
+    pool: &deadpool_diesel::postgres::Pool,
+    pair_id: String,
+) -> Result<bool, InfraError> {
+    let conn = pool.get().await.map_err(adapt_infra_error)?;
+
+    let res = conn
+        .interact(move |conn| Entry::exists(conn, pair_id))
+        .await
+        .map_err(adapt_infra_error)?
+        .map_err(adapt_infra_error)?;
+
+    Ok(res)
+}
+
+async fn get_price_decimals(
+    pool: &deadpool_diesel::postgres::Pool,
+    pair_id: String,
+    interval: Interval,
+    timestamp: u64,
+) -> Result<(MedianEntry, u32), InfraError> {
+    let entry = get_median_price(pool, pair_id.clone(), interval, timestamp).await?;
+
+    let decimals = get_decimals(pool, &pair_id).await?;
+
+    Ok((entry, decimals))
+}
+
 pub async fn get_median_price(
     pool: &deadpool_diesel::postgres::Pool,
     pair_id: String,

pragma-node/src/utils/conversion.rs

@@ -0,0 +1,30 @@
+use bigdecimal::BigDecimal;
+use pragma_entities::InfraError;
+
+pub fn convert_via_quote(
+    a_price: BigDecimal,
+    b_price: BigDecimal,
+    output_decimals: u32,
+) -> Result<BigDecimal, InfraError> {
+    if b_price == BigDecimal::from(0) {
+        return Err(InfraError::InternalServerError);
+    }
+
+    let power = BigDecimal::from(10_i64.pow(output_decimals));
+
+    Ok(a_price * power / b_price)
+}
+
+pub fn normalize_to_decimals(
+    value: BigDecimal,
+    original_decimals: u32,
+    target_decimals: u32,
+) -> BigDecimal {
+    if target_decimals >= original_decimals {
+        let power = BigDecimal::from(10_i64.pow(target_decimals - original_decimals));
+        value * power
+    } else {
+        let power = BigDecimal::from(10_i64.pow(original_decimals - target_decimals));
+        value / power
+    }
+}

pragma-node/src/utils/mod.rs

@@ -1,6 +1,8 @@
+pub use conversion::{convert_via_quote, normalize_to_decimals};
 pub use custom_extractors::json_extractor::JsonExtractor;
 pub use custom_extractors::path_extractor::PathExtractor;
 pub use signing::typed_data::TypedData;
 
+mod conversion;
 mod custom_extractors;
 mod signing;