this brings back balance back to the stats request (#112)

* removed bloom filter temporarily, added some fixes with decimals in payment.rs

* balance is inside the influx query again

* fixed aggregate

scripts/generate-requests-and-stats.sh

@@ -5,4 +5,6 @@
 # https://github.com/INFURA/versus
 # ./ethspam | ./versus --stop-after 100 "http://localhost:8544/" # Pipe into the endpoint ..., add a bearer token and all that
 
-./ethspam http://127.0.0.1:8544/rpc/01H0ZZJDNNEW49FRFS4D9SPR8B | ./versus --concurrency=4 --stop-after 100 http://localhost:8544/rpc/01H0ZZJDNNEW49FRFS4D9SPR8B
+./ethspam http://127.0.0.1:8544/rpc/01H2D5DN4D423VR2KFWBZE46TR | ./versus --concurrency=4 --stop-after 10000 http://localhost:8544/rpc/01H2D5DN4D423VR2KFWBZE46TR
+
+./ethspam http://127.0.0.1:8544/rpc/01H2D5CAP1KF2NKRS30SGATDSD | ./versus --concurrency=4 --stop-after 10000 http://localhost:8544/rpc/01H2D5CAP1KF2NKRS30SGATDSD

scripts/manual-tests/101-balance-referral-stats.sh

@@ -127,4 +127,10 @@ curl \
 
 curl \
 -H "Authorization: Bearer 01H2D5CAQJF7P80222P4ZAFQ26" \
--X GET "127.0.0.1:8544/user/referral/stats/shared-codes"
+-X GET "127.0.0.1:8544/user/referral/stats/shared-codes"
+
+
+# Finally also get some stats
+curl -X GET \
+-H "Authorization: Bearer 01H2D5DN564M4Q2T6PETEZY83Q" \
+"http://localhost:8544/user/stats/detailed?query_start=1686236378&query_window_seconds=3600"

web3_proxy/src/stats/influxdb_queries.rs

@@ -167,6 +167,23 @@ pub async fn query_user_stats<'a>(
         StatType::Detailed => "".to_string(),
     };
 
+    let join_candidates = match stat_response_type {
+        StatType::Aggregated => f!(
+            r#"{:?}"#,
+            vec!["_time", "_measurement", "chain_id", "rpc_secret_key_id"]
+        ),
+        StatType::Detailed => f!(
+            r#"{:?}"#,
+            vec![
+                "_time",
+                "_measurement",
+                "method",
+                "chain_id",
+                "rpc_secret_key_id"
+            ]
+        ),
+    };
+
     let query = f!(r#"
     base = from(bucket: "{bucket}")
         |> range(start: {query_start}, stop: {query_stop})
@@ -174,8 +191,8 @@ pub async fn query_user_stats<'a>(
         |> filter(fn: (r) => r["_measurement"] == "{measurement}")
         {filter_chain_id}
         {drop_method}
-
-    base
+        
+    cumsum = base 
         |> aggregateWindow(every: {query_window_seconds}s, fn: sum, createEmpty: false)
         |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
         |> drop(columns: ["balance"])
@@ -186,7 +203,21 @@ pub async fn query_user_stats<'a>(
         |> sort(columns: ["frontend_requests"], desc: true)
         |> limit(n: 1)
         |> group()
-        |> sort(columns: ["_time", "_measurement", "archive_needed", "chain_id", "error_response", "method", "rpc_secret_key_id"], desc: true)
+        
+    balance = base
+        |> toFloat()
+        |> aggregateWindow(every: {query_window_seconds}s, fn: mean, createEmpty: false)
+        |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
+        |> group(columns: ["_time", "_measurement", "chain_id", "method", "rpc_secret_key_id"])
+        |> mean(column: "balance")
+        |> group()
+        |> sort(columns: ["_time", "_measurement", "chain_id", "method", "rpc_secret_key_id"], desc: true)
+
+    join(
+        tables: {{cumsum, balance}},
+        on: {join_candidates}
+    )
+        |> sort(columns: ["_time", "_measurement", "chain_id", "method", "rpc_secret_key_id"], desc: true)
     "#);
 
     debug!("Raw query to db is: {:#?}", query);