feat(stats): Chart Migration Five (#1147)

PR introduces daily transactions for the main page. It works even on not fully indexed network, because it recalculates data for the whole time segment each time.

- [Chart update blockscout source](https://github.com/blockscout/blockscout/blob/1450388918bd2dc43950dde1dc819dac3edec155/apps/explorer/lib/explorer/history/process.ex#L53)
- [SQL queries to db](https://github.com/blockscout/blockscout/blob/1450388918bd2dc43950dde1dc819dac3edec155/apps/explorer/lib/explorer/chain/transaction/history/historian.ex#L19)

stats/config/charts.json

@@ -111,6 +111,10 @@
             "title": "New transactions",
             "description": "Number of new transactions"
         },
+        "new_txns_window": {
+            "title": "Daily transactions",
+            "description": "Number of transactions yesterday (0:00 - 23:59 UTC). The chart displays daily transactions for the past 30 days."
+        },
         "txns_growth": {
             "title": "Number of transactions",
             "description": "Cumulative transaction growth over time"

stats/config/layout.json

@@ -39,7 +39,8 @@
                 "txns_growth",
                 "new_operational_txns",
                 "operational_txns_growth",
-                "txns_success_rate"
+                "txns_success_rate",
+                "new_txns_window"
             ]
         },
         {

stats/config/update_groups.json

@@ -9,6 +9,7 @@
         "total_txns_group": "0 5 */2 * * * *",
         "total_operational_txns_group": "0 5 1 * * * *",
         "yesterday_txns_group": "0 8 0 * * * *",
+        "new_txns_window_group": "0 8 0 * * * *",
         "active_recurring_accounts_daily_recurrence_60_days_group": "0 0 2 * * * *",
         "active_recurring_accounts_daily_recurrence_90_days_group": "0 20 2 * * * *",
         "active_recurring_accounts_daily_recurrence_120_days_group": "0 40 2 * * * *",

stats/stats-server/src/runtime_setup.rs

@@ -275,6 +275,7 @@ impl RuntimeSetup {
             Arc::new(ActiveRecurringAccountsMonthlyRecurrence120DaysGroup),
             Arc::new(ActiveRecurringAccountsWeeklyRecurrence120DaysGroup),
             Arc::new(ActiveRecurringAccountsYearlyRecurrence120DaysGroup),
+            Arc::new(NewTxnsWindowGroup),
             // singletons but not really (include all resolutions of the same chart)
             Arc::new(AverageBlockRewardsGroup),
             Arc::new(AverageBlockSizeGroup),

stats/stats-server/tests/it/lines.rs

@@ -69,6 +69,10 @@ async fn test_lines_ok() {
 
     let mut enabled_resolutions = enabled_resolutions(line_charts).await;
 
+    // does not return data for latest dates,
+    // so todo: test later with other main page stuff
+    enabled_resolutions.remove("newTxnsWindow");
+
     for line_name in [
         "accountsGrowth",
         "activeAccounts",
@@ -88,6 +92,7 @@ async fn test_lines_ok() {
         "newBlocks",
         "newNativeCoinTransfers",
         "newTxns",
+        // "newTxnsWindow",
         "txnsFee",
         "txnsGrowth",
         // "newOperationalTxns",
@@ -122,7 +127,7 @@ async fn test_lines_ok() {
 
             assert!(
                 !chart_data.is_empty(),
-                "chart '{line_name}' '{resolution}' is empty"
+                "chart data for '{line_name}' '{resolution}' is empty"
             );
 
             let info = chart

stats/stats/src/charts/db_interaction/write.rs

@@ -56,6 +56,14 @@ where
     Ok(())
 }
 
+pub async fn clear_all_chart_data<C: ConnectionTrait>(db: &C, chart_id: i32) -> Result<(), DbErr> {
+    chart_data::Entity::delete_many()
+        .filter(chart_data::Column::ChartId.eq(chart_id))
+        .exec(db)
+        .await?;
+    Ok(())
+}
+
 pub async fn set_last_updated_at<Tz>(
     chart_id: i32,
     db: &DatabaseConnection,

stats/stats/src/charts/lines/mod.rs

@@ -18,6 +18,7 @@ mod new_native_coin_holders;
 mod new_native_coin_transfers;
 mod new_operational_txns;
 mod new_txns;
+mod new_txns_window;
 mod new_verified_contracts;
 mod operational_txns_growth;
 mod txns_fee;
@@ -96,6 +97,7 @@ pub use new_operational_txns::{
 };
 pub(crate) use new_txns::NewTxnsStatement;
 pub use new_txns::{NewTxns, NewTxnsInt, NewTxnsMonthly, NewTxnsWeekly, NewTxnsYearly};
+pub use new_txns_window::NewTxnsWindow;
 pub use new_verified_contracts::{
     NewVerifiedContracts, NewVerifiedContractsMonthly, NewVerifiedContractsWeekly,
     NewVerifiedContractsYearly,

stats/stats/src/charts/lines/new_txns_window.rs

@@ -0,0 +1,199 @@
+//! New transactions for the last N days (usually 30).
+//!
+//! Allowed to work on a non-indexed networks, as it
+//! recalculates whole N day window/range each time.
+//!
+//! Does not include last day, even as incomplete day.
+
+use crate::{
+    data_source::{
+        kinds::{
+            local_db::{
+                parameters::{
+                    update::batching::{
+                        parameters::{BatchMaxDays, ClearAllAndPassStep},
+                        BatchUpdate,
+                    },
+                    DefaultCreate, DefaultQueryVec,
+                },
+                LocalDbChartSource,
+            },
+            remote_db::{RemoteDatabaseSource, RemoteQueryBehaviour, StatementFromRange},
+        },
+        types::BlockscoutMigrations,
+        UpdateContext,
+    },
+    range::UniversalRange,
+    types::{Timespan, TimespanDuration, TimespanValue},
+    utils::day_start,
+    ChartError, ChartProperties, Named,
+};
+
+use chrono::{DateTime, NaiveDate, Utc};
+use entity::sea_orm_active_enums::ChartType;
+use sea_orm::{FromQueryResult, Statement};
+
+use super::NewTxnsStatement;
+
+pub const WINDOW: u64 = 30;
+
+fn new_txns_window_statement(
+    update_day: NaiveDate,
+    completed_migrations: &BlockscoutMigrations,
+) -> Statement {
+    // `update_day` is not included because the data would
+    // be incomplete.
+    let window =
+        day_start(&update_day.saturating_sub(TimespanDuration::from_timespan_repeats(WINDOW)))
+            ..day_start(&update_day);
+    NewTxnsStatement::get_statement(Some(window), completed_migrations)
+}
+
+pub struct NewTxnsWindowQuery;
+
+impl RemoteQueryBehaviour for NewTxnsWindowQuery {
+    type Output = Vec<TimespanValue<NaiveDate, String>>;
+
+    async fn query_data(
+        cx: &UpdateContext<'_>,
+        _range: UniversalRange<DateTime<Utc>>,
+    ) -> Result<Vec<TimespanValue<NaiveDate, String>>, ChartError> {
+        let update_day = cx.time.date_naive();
+        let query = new_txns_window_statement(update_day, &cx.blockscout_applied_migrations);
+        let mut data = TimespanValue::<NaiveDate, String>::find_by_statement(query)
+            .all(cx.blockscout.connection.as_ref())
+            .await
+            .map_err(ChartError::BlockscoutDB)?;
+        // linear time for sorted sequences
+        data.sort_unstable_by(|a, b| a.timespan.cmp(&b.timespan));
+        Ok(data)
+    }
+}
+
+pub type NewTxnsWindowRemote = RemoteDatabaseSource<NewTxnsWindowQuery>;
+
+pub struct Properties;
+
+impl Named for Properties {
+    fn name() -> String {
+        "newTxnsWindow".into()
+    }
+}
+
+impl ChartProperties for Properties {
+    type Resolution = NaiveDate;
+
+    fn chart_type() -> ChartType {
+        ChartType::Line
+    }
+}
+
+pub type NewTxnsWindow = LocalDbChartSource<
+    NewTxnsWindowRemote,
+    (),
+    DefaultCreate<Properties>,
+    BatchUpdate<
+        NewTxnsWindowRemote,
+        (),
+        ClearAllAndPassStep,
+        BatchMaxDays,
+        DefaultQueryVec<Properties>,
+        Properties,
+    >,
+    DefaultQueryVec<Properties>,
+    Properties,
+>;
+
+#[cfg(test)]
+mod tests {
+    use pretty_assertions::assert_eq;
+
+    use super::*;
+    use crate::{
+        data_source::{DataSource, UpdateParameters},
+        query_dispatch::QuerySerialized,
+        tests::{
+            mock_blockscout::{fill_mock_blockscout_data, imitate_reindex},
+            point_construction::dt,
+            simple_test::{chart_output_to_expected, map_str_tuple_to_owned, prepare_chart_test},
+        },
+        utils::MarkedDbConnection,
+    };
+
+    #[tokio::test]
+    #[ignore = "needs database to run"]
+    async fn update_txns_window_clears_and_overwrites() {
+        let (init_time, db, blockscout) =
+            prepare_chart_test::<NewTxnsWindow>("update_txns_window_clears_and_overwrites", None)
+                .await;
+        {
+            let current_date = init_time.date_naive();
+            fill_mock_blockscout_data(&blockscout, current_date).await;
+        }
+        let current_time = dt("2022-12-01T00:00:00").and_utc();
+
+        let mut parameters = UpdateParameters {
+            db: &MarkedDbConnection::from_test_db(&db).unwrap(),
+            blockscout: &MarkedDbConnection::from_test_db(&blockscout).unwrap(),
+            blockscout_applied_migrations: BlockscoutMigrations::latest(),
+            update_time_override: Some(current_time),
+            force_full: false,
+        };
+        let cx = UpdateContext::from_params_now_or_override(parameters.clone());
+        NewTxnsWindow::update_recursively(&cx).await.unwrap();
+        assert_eq!(
+            &chart_output_to_expected(
+                NewTxnsWindow::query_data_static(&cx, UniversalRange::full(), None, false)
+                    .await
+                    .unwrap()
+            ),
+            &map_str_tuple_to_owned(vec![
+                ("2022-11-09", "5"),
+                ("2022-11-10", "12"),
+                ("2022-11-11", "14"),
+                ("2022-11-12", "5"),
+                // update day is not included
+            ]),
+        );
+
+        let current_time = dt("2022-12-10T00:00:00").and_utc();
+        parameters.update_time_override = Some(current_time);
+        let cx = UpdateContext::from_params_now_or_override(parameters.clone());
+        NewTxnsWindow::update_recursively(&cx).await.unwrap();
+        assert_eq!(
+            &chart_output_to_expected(
+                NewTxnsWindow::query_data_static(&cx, UniversalRange::full(), None, false)
+                    .await
+                    .unwrap()
+            ),
+            &map_str_tuple_to_owned(vec![
+                // values outside the window are removed
+                ("2022-11-10", "12"),
+                ("2022-11-11", "14"),
+                ("2022-11-12", "5"),
+                ("2022-12-01", "5"),
+            ]),
+        );
+
+        imitate_reindex(&blockscout, init_time.date_naive()).await;
+
+        let current_time = dt("2022-12-11T00:00:00").and_utc();
+        parameters.update_time_override = Some(current_time);
+        let cx = UpdateContext::from_params_now_or_override(parameters);
+        NewTxnsWindow::update_recursively(&cx).await.unwrap();
+        assert_eq!(
+            &chart_output_to_expected(
+                NewTxnsWindow::query_data_static(&cx, UniversalRange::full(), None, false)
+                    .await
+                    .unwrap()
+            ),
+            &map_str_tuple_to_owned(vec![
+                // values outside the window are removed
+                // new values within the window are added
+                ("2022-11-11", "18"),
+                ("2022-11-12", "5"),
+                ("2022-12-01", "5"),
+            ]),
+        );
+    }
+}

stats/stats/src/data_source/kinds/local_db/parameters/update/batching/parameter_traits.rs

@@ -1,7 +1,7 @@
 use std::future::Future;
 
 use chrono::{DateTime, Utc};
-use sea_orm::DatabaseConnection;
+use sea_orm::{ConnectionTrait, TransactionTrait};
 
 use crate::{
     types::{Timespan, TimespanValue},
@@ -17,8 +17,8 @@ where
     /// Update chart with data from its dependencies.
     ///
     /// Returns how many records were found
-    fn batch_update_values_step_with(
-        db: &DatabaseConnection,
+    fn batch_update_values_step_with<C: ConnectionTrait + TransactionTrait>(
+        db: &C,
         chart_id: i32,
         update_time: DateTime<Utc>,
         min_blockscout_block: i64,

stats/stats/src/data_source/kinds/local_db/parameters/update/batching/parameters/cumulative.rs

@@ -4,7 +4,7 @@ use std::{fmt::Display, marker::PhantomData, ops::Add, str::FromStr};
 
 use chrono::{DateTime, Utc};
 use rust_decimal::prelude::Zero;
-use sea_orm::DatabaseConnection;
+use sea_orm::{ConnectionTrait, TransactionTrait};
 
 use crate::{
     data_source::kinds::local_db::parameters::update::batching::parameter_traits::BatchStepBehaviour,
@@ -29,15 +29,18 @@ where
     <Value as FromStr>::Err: Display,
     ChartProps: ChartProperties,
 {
-    async fn batch_update_values_step_with(
-        db: &DatabaseConnection,
+    async fn batch_update_values_step_with<C>(
+        db: &C,
         chart_id: i32,
         update_time: DateTime<Utc>,
         min_blockscout_block: i64,
         last_accurate_point: TimespanValue<Resolution, String>,
         main_data: Vec<TimespanValue<Resolution, Value>>,
         _resolution_data: (),
-    ) -> Result<usize, ChartError> {
+    ) -> Result<usize, ChartError>
+    where
+        C: ConnectionTrait + TransactionTrait,
+    {
         let partial_sum = last_accurate_point.value.parse::<Value>().map_err(|e| {
             ChartError::Internal(format!(
                 "failed to parse value in chart '{}': {e}",

stats/stats/src/data_source/kinds/local_db/parameters/update/batching/parameters/mock.rs

@@ -1,7 +1,7 @@
 use std::marker::PhantomData;
 
 use chrono::{DateTime, NaiveDate, Utc};
-use sea_orm::DatabaseConnection;
+use sea_orm::{ConnectionTrait, TransactionTrait};
 
 use crate::{
     data_source::kinds::local_db::parameters::update::batching::parameter_traits::BatchStepBehaviour,
@@ -32,15 +32,18 @@ impl<StepsRecorder> BatchStepBehaviour<NaiveDate, Vec<DateValue<String>>, ()>
 where
     StepsRecorder: Recorder<Data = StepInput<Vec<DateValue<String>>, ()>>,
 {
-    async fn batch_update_values_step_with(
-        db: &DatabaseConnection,
+    async fn batch_update_values_step_with<C>(
+        db: &C,
         chart_id: i32,
         update_time: DateTime<Utc>,
         min_blockscout_block: i64,
         last_accurate_point: DateValue<String>,
         main_data: Vec<DateValue<String>>,
         resolution_data: (),
-    ) -> Result<usize, ChartError> {
+    ) -> Result<usize, ChartError>
+    where
+        C: ConnectionTrait + TransactionTrait,
+    {
         StepsRecorder::record(StepInput {
             chart_id,
             update_time,