feat: adapt route to use new views

Author: EvolveArt

pragma-entities/migrations/2023-10-12-232125_add_publishers_table/up.sql

@@ -6,4 +6,4 @@ CREATE TABLE publishers (
     active_key VARCHAR NOT NULL,
     active BOOLEAN NOT NULL,
     PRIMARY KEY (id)
-);
+);

pragma-entities/migrations/2023-10-16-233310_add_account_address/up.sql

@@ -1,3 +1,12 @@
 -- Your SQL goes here
 ALTER TABLE publishers
-ADD COLUMN account_address VARCHAR NOT NULL DEFAULT '';
+ADD COLUMN account_address VARCHAR NOT NULL DEFAULT '';
+
+INSERT INTO publishers (name, master_key, active_key, active, account_address) 
+VALUES (
+    'PRAGMA',
+    '0x05e6361b53afbb451d1326ed4e37aecff9ef68af8318eb3c8dc58bcadfc16705',
+    '0x05e6361b53afbb451d1326ed4e37aecff9ef68af8318eb3c8dc58bcadfc16705', 
+    true, 
+    '0x624EBFB99865079BD58CFCFB925B6F5CE940D6F6E41E118B8A72B7163FB435C'
+);

pragma-entities/migrations/2023-11-24-185951_add_source_index/up.sql

@@ -1,3 +1,3 @@
 -- Your SQL goes here
 CREATE UNIQUE INDEX idx_entries_unique
-  ON entries(pair_id, source, timestamp);
+  ON entries(pair_id, source, timestamp DESC);

pragma-entities/migrations/2024-01-11-123510_add_continuous_aggregates/up.sql

@@ -1,46 +1,49 @@
 -- Your SQL goes here
 CREATE MATERIALIZED VIEW price_1_min_agg
 WITH (timescaledb.continuous, timescaledb.materialized_only = true)
-AS SELECT source,
+AS SELECT 
     pair_id,
     time_bucket('1 min'::interval, timestamp) as bucket,
-    percentile_agg(price)
+    approx_percentile(0.5, percentile_agg(price)) AS median_price,
+    COUNT(DISTINCT source) as num_sources
 FROM entries
-GROUP BY source, pair_id, bucket
+GROUP BY bucket, pair_id
 WITH NO DATA;
 
 SELECT add_continuous_aggregate_policy('price_1_min_agg',
-  start_offset => INTERVAL '3 min',
+  start_offset => NULL,
   end_offset => INTERVAL '1 min',
   schedule_interval => INTERVAL '1 min');
 
 CREATE MATERIALIZED VIEW price_15_min_agg
 WITH (timescaledb.continuous, timescaledb.materialized_only = true)
-AS SELECT source,
+AS SELECT 
     pair_id,
     time_bucket('15 min'::interval, timestamp) as bucket,
-    percentile_agg(price)
+    approx_percentile(0.5, percentile_agg(price)) AS median_price,
+    COUNT(DISTINCT source) as num_sources
 FROM entries
-GROUP BY source, pair_id, bucket
+GROUP BY bucket, pair_id
 WITH NO DATA;
 
 SELECT add_continuous_aggregate_policy('price_15_min_agg',
-  start_offset => INTERVAL '45 min',
+  start_offset => NULL,
   end_offset => INTERVAL '15 min',
   schedule_interval => INTERVAL '15 min');
 
 CREATE MATERIALIZED VIEW price_1_h_agg
 WITH (timescaledb.continuous, timescaledb.materialized_only = true)
-AS SELECT source,
+AS SELECT 
     pair_id,
     time_bucket('1 hour'::interval, timestamp) as bucket,
-    percentile_agg(price)
+    approx_percentile(0.5, percentile_agg(price)) AS median_price,
+    COUNT(DISTINCT source) as num_sources
 FROM entries
-GROUP BY source, pair_id, bucket
+GROUP BY bucket, pair_id
 WITH NO DATA;
 
 SELECT add_continuous_aggregate_policy('price_1_h_agg',
-  start_offset => INTERVAL '3 hours',
+  start_offset => NULL,
   end_offset => INTERVAL '1 hour',
   schedule_interval => INTERVAL '1 hour');
 

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

@@ -1,14 +1,14 @@
 use axum::extract::State;
 use axum::Json;
-use bigdecimal::num_bigint::{BigInt, ToBigInt};
+use bigdecimal::num_bigint::ToBigInt;
 
 use crate::handlers::entries::GetEntryResponse;
 use crate::infra::repositories::entry_repository::{self, MedianEntry};
 use crate::utils::PathExtractor;
 use crate::AppState;
 use pragma_entities::{error::InfraError, EntryError};
 
-use super::utils::{compute_median_price_and_time, currency_pair_to_pair_id};
+use super::utils::currency_pair_to_pair_id;
 
 #[utoipa::path(
         get,
@@ -41,19 +41,14 @@ pub async fn get_entry(
     }
 
     // Get entries from database with given pair id (only the latest one grouped by publisher)
-    let mut entries = entry_repository::get_median_entries(&state.pool, pair_id.clone())
+    let entry = entry_repository::get_median_price(&state.pool, pair_id.clone())
         .await
         .map_err(|db_error| match db_error {
             InfraError::InternalServerError => EntryError::InternalServerError,
             InfraError::NotFound => EntryError::NotFound(pair_id.clone()),
             InfraError::InvalidTimeStamp => EntryError::InvalidTimestamp,
         })?;
 
-    // Error if no entries found
-    if entries.is_empty() {
-        return Err(EntryError::UnknownPairId(pair_id));
-    }
-
     let decimals = entry_repository::get_decimals(&state.pool, &pair_id)
         .await
         .map_err(|db_error| match db_error {
@@ -63,28 +58,25 @@ pub async fn get_entry(
         })?;
 
     Ok(Json(adapt_entry_to_entry_response(
-        pair_id,
-        &mut entries,
-        decimals,
+        pair_id, &entry, decimals,
     )))
 }
 
 fn adapt_entry_to_entry_response(
     pair_id: String,
-    entries: &mut Vec<MedianEntry>,
+    entry: &MedianEntry,
     decimals: u32,
 ) -> GetEntryResponse {
-    let (price, timestamp) = compute_median_price_and_time(entries).unwrap_or_default();
-
     GetEntryResponse {
         pair_id,
-        timestamp: timestamp.timestamp_millis() as u64,
-        num_sources_aggregated: entries.len(),
+        timestamp: entry.time.timestamp_millis() as u64,
+        num_sources_aggregated: entry.num_sources as usize,
         price: format!(
             "0x{}",
-            price
+            entry
+                .median_price
                 .to_bigint()
-                .unwrap_or(BigInt::default())
+                .unwrap_or_default()
                 .to_str_radix(16)
         ),
         decimals,

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

@@ -16,6 +16,7 @@ pub(crate) fn currency_pair_to_pair_id(quote: &str, base: &str) -> String {
 /// The median is computed as the middle value of a sorted list of values.
 /// If the list has an even number of values, the median is computed as the average of the two middle values.
 /// If the list is empty, None is returned.
+#[allow(dead_code)]
 pub(crate) fn compute_median_price_and_time(
     entries: &mut Vec<MedianEntry>,
 ) -> Option<(BigDecimal, NaiveDateTime)> {
@@ -74,27 +75,27 @@ fn test_volatility() {
         MedianEntry {
             time: chrono::NaiveDateTime::from_timestamp_opt(1640995200, 0).unwrap(),
             median_price: bigdecimal::BigDecimal::from(47686),
-            source: "source".to_string(),
+            num_sources: 5,
         },
         MedianEntry {
             time: chrono::NaiveDateTime::from_timestamp_opt(1641081600, 0).unwrap(),
             median_price: bigdecimal::BigDecimal::from(47345),
-            source: "source".to_string(),
+            num_sources: 5,
         },
         MedianEntry {
             time: chrono::NaiveDateTime::from_timestamp_opt(1641168000, 0).unwrap(),
             median_price: bigdecimal::BigDecimal::from(46458),
-            source: "source".to_string(),
+            num_sources: 5,
         },
         MedianEntry {
             time: chrono::NaiveDateTime::from_timestamp_opt(1641254400, 0).unwrap(),
             median_price: bigdecimal::BigDecimal::from(45897),
-            source: "source".to_string(),
+            num_sources: 5,
         },
         MedianEntry {
             time: chrono::NaiveDateTime::from_timestamp_opt(1641340800, 0).unwrap(),
             median_price: bigdecimal::BigDecimal::from(43569),
-            source: "source".to_string(),
+            num_sources: 5,
         },
     ];
 

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

@@ -64,42 +64,43 @@ pub async fn _get_all(
 
 #[derive(Debug, Serialize, Queryable)]
 pub struct MedianEntry {
-    pub source: String,
     pub time: NaiveDateTime,
     pub median_price: BigDecimal,
+    pub num_sources: i32,
 }
 
 #[derive(Serialize, QueryableByName)]
 pub struct MedianEntryRaw {
-    #[diesel(sql_type = diesel::sql_types::Text)]
-    pub source: String,
     #[diesel(sql_type = diesel::sql_types::Timestamp)]
     pub time: NaiveDateTime,
     #[diesel(sql_type = diesel::sql_types::Numeric)]
     pub median_price: BigDecimal,
+    #[diesel(sql_type = diesel::sql_types::Integer)]
+    pub num_sources: i32,
 }
 
-pub async fn get_median_entries(
+pub async fn get_median_price(
     pool: &deadpool_diesel::postgres::Pool,
     pair_id: String,
-) -> Result<Vec<MedianEntry>, InfraError> {
+) -> Result<MedianEntry, InfraError> {
     let conn = pool.get().await.map_err(adapt_infra_error)?;
 
     let raw_sql = r#"
         -- query the materialized realtime view
         SELECT
-            source,
-            MAX(bucket) AS time,
-            approx_percentile(0.5, percentile_agg) AS median_price
+            bucket AS time,
+            median_price,
+            num_sources
         FROM
             price_1_min_agg
         WHERE
             pair_id = $1
         ORDER BY
-            source;
+            time
+        LIMIT 1;
     "#;
 
-    let raw_entries: Vec<MedianEntryRaw> = conn
+    let raw_entry = conn
         .interact(move |conn| {
             diesel::sql_query(raw_sql)
                 .bind::<diesel::sql_types::Text, _>(pair_id)
@@ -109,16 +110,15 @@ pub async fn get_median_entries(
         .map_err(adapt_infra_error)?
         .map_err(adapt_infra_error)?;
 
-    let entries: Vec<MedianEntry> = raw_entries
-        .into_iter()
-        .map(|raw_entry| MedianEntry {
-            time: raw_entry.time,
-            median_price: raw_entry.median_price,
-            source: raw_entry.source,
-        })
-        .collect();
+    let raw_entry = raw_entry.first().ok_or(InfraError::NotFound)?;
 
-    Ok(entries)
+    let entry: MedianEntry = MedianEntry {
+        time: raw_entry.time,
+        median_price: raw_entry.median_price.clone(),
+        num_sources: raw_entry.num_sources,
+    };
+
+    Ok(entry)
 }
 
 pub async fn get_entries_between(
@@ -135,14 +135,16 @@ pub async fn get_entries_between(
 
     let raw_sql = r#"
         SELECT
-            source,
-            "timestamp" AS "time",
-            PERCENTILE_DISC(0.5) WITHIN GROUP(ORDER BY price) AS "median_price"
-        FROM entries
-        WHERE pair_id = $1
-        AND "timestamp" BETWEEN $2 AND $3
-        GROUP BY (timestamp, source)
-        ORDER BY timestamp ASC;
+            bucket AS time,
+            median_price,
+            num_sources
+        FROM price_1_min_agg
+        WHERE 
+            pair_id = $1
+        AND 
+            time BETWEEN $2 AND $3
+        ORDER BY 
+            time DESC;
     "#;
 
     let raw_entries = conn
@@ -162,7 +164,7 @@ pub async fn get_entries_between(
         .map(|raw_entry| MedianEntry {
             time: raw_entry.time,
             median_price: raw_entry.median_price,
-            source: raw_entry.source,
+            num_sources: raw_entry.num_sources,
         })
         .collect();