add: spawn advanced analytics recorder

Author: samparsky

primitives/src/sentry.rs

@@ -30,7 +30,7 @@ pub struct ApproveStateValidatorMessage {
 }
 
 #[serde(tag = "type", rename_all = "SCREAMING_SNAKE_CASE")]
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub enum Event {
     #[serde(rename_all = "camelCase")]
     Impression {
@@ -92,7 +92,7 @@ impl fmt::Display for Event {
     }
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct Earner {
     #[serde(rename = "publisher")]
     pub address: String,

sentry/src/analytics_recorder.rs

@@ -23,10 +23,10 @@ fn get_payout(channel: &Channel, event: &Event) -> BigNum {
 
 pub async fn record(
     mut conn: MultiplexedConnection,
-    channel: &Channel,
-    session: &Session,
-    events: &[Event],
-    logger: &Logger,
+    channel: Channel,
+    session: Session,
+    events: Vec<Event>,
+    logger: Logger,
 ) {
     let mut db = redis::pipe();
 
@@ -47,7 +47,7 @@ pub async fn record(
                 referrer,
             } => {
                 let divisor = BigNum::from(10u64.pow(18));
-                let pay_amount = get_payout(channel, event)
+                let pay_amount = get_payout(&channel, event)
                     .div_floor(&divisor)
                     .to_u64()
                     .expect("should always have a payout");

sentry/src/db/analytics.rs

@@ -25,15 +25,18 @@ pub enum AnalyticsType {
 
 pub async fn advertiser_channel_ids(
     pool: &DbPool,
-    creator: &ValidatorId
+    creator: &ValidatorId,
 ) -> Result<Vec<ChannelId>, RunError<bb8_postgres::tokio_postgres::Error>> {
     pool.run(move |connection| async move {
-        match connection.prepare("SELECT id FROM channels WHERE creator = {}").await {
+        match connection
+            .prepare("SELECT id FROM channels WHERE creator = {}")
+            .await
+        {
             Ok(stmt) => match connection.query(&stmt, &[creator]).await {
                 Ok(rows) => {
                     let channel_ids: Vec<ChannelId> = rows.iter().map(ChannelId::from).collect();
                     Ok((channel_ids, connection))
-                },
+                }
                 Err(e) => Err((e, connection)),
             },
             Err(e) => Err((e, connection)),

sentry/src/event_aggregator.rs

@@ -1,4 +1,5 @@
 use crate::access::check_access;
+use crate::analytics_recorder;
 use crate::db::event_aggregate::insert_event_aggregate;
 use crate::db::get_channel_by_id;
 use crate::db::DbPool;
@@ -8,17 +9,16 @@ use crate::ResponseError;
 use crate::Session;
 use async_std::sync::RwLock;
 use chrono::Utc;
+use lazy_static::lazy_static;
 use primitives::adapter::Adapter;
 use primitives::sentry::{Event, EventAggregate};
 use primitives::{Channel, ChannelId};
 use slog::{error, Logger};
 use std::collections::HashMap;
+use std::env;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::time::delay_for;
-use lazy_static::lazy_static;
-use std::env;
-use crate::analytics_recorder;
 
 lazy_static! {
     pub static ref ANALYTICS_RECORDER: String =
@@ -74,6 +74,7 @@ impl EventAggregator {
         let recorder = self.recorder.clone();
         let aggr_throttle = app.config.aggr_throttle;
         let dbpool = app.pool.clone();
+        let redis = app.redis.clone();
         let logger = app.logger.clone();
 
         let mut channel_recorder = self.recorder.write().await;
@@ -135,14 +136,21 @@ impl EventAggregator {
             return Err(ResponseError::BadRequest(e.to_string()));
         }
 
-        if ANALYTICS_RECORDER.ne(&"false".to_string()) {
-            analytics_recorder::record(app.redis.clone(), &record.channel, &session, &events, &app.logger).await
-        }
-
         events
             .iter()
             .for_each(|ev| event_reducer::reduce(&record.channel, &mut record.aggregate, ev));
 
+        if ANALYTICS_RECORDER.ne(&"false".to_string()) {
+            let logger = app.logger.clone();
+            tokio::spawn(analytics_recorder::record(
+                redis.clone(),
+                record.channel.clone(),
+                session.clone(),
+                events.to_owned().to_vec(),
+                logger,
+            ));
+        }
+
         // drop write access to RwLock
         // this is required to prevent a deadlock in store
         drop(channel_recorder);

sentry/src/lib.rs

@@ -18,7 +18,7 @@ use primitives::adapter::Adapter;
 use primitives::{Config, ValidatorId};
 use redis::aio::MultiplexedConnection;
 use regex::Regex;
-use routes::analytics::{advertiser_analytics, analytics, publisher_analytics, advanced_analytics};
+use routes::analytics::{advanced_analytics, advertiser_analytics, analytics, publisher_analytics};
 use routes::cfg::config;
 use routes::channel::{
     channel_list, create_channel, create_validator_messages, insert_events, last_approved,
@@ -141,7 +141,7 @@ impl<A: Adapter + 'static> Application<A> {
             ("/channel/list", &Method::GET) => channel_list(req, &self).await,
 
             ("/analytics", &Method::GET) => analytics(req, &self).await,
-            ("/analytics/advanced", &Method::GET) =>  {
+            ("/analytics/advanced", &Method::GET) => {
                 let req = match chain(req, &self, vec![Box::new(auth_required_middleware)]).await {
                     Ok(req) => req,
                     Err(error) => {
@@ -150,7 +150,7 @@ impl<A: Adapter + 'static> Application<A> {
                 };
 
                 advanced_analytics(req, &self).await
-            },
+            }
             ("/analytics/for-advertiser", &Method::GET) => {
                 let req = match chain(req, &self, vec![Box::new(auth_required_middleware)]).await {
                     Ok(req) => req,

sentry/src/routes/analytics.rs

@@ -1,4 +1,6 @@
-use crate::db::analytics::{get_analytics, AnalyticsType, advertiser_channel_ids, get_advanced_reports};
+use crate::db::analytics::{
+    advertiser_channel_ids, get_advanced_reports, get_analytics, AnalyticsType,
+};
 use crate::success_response;
 use crate::Application;
 use crate::ResponseError;
@@ -97,7 +99,6 @@ pub async fn process_analytics<A: Adapter>(
         .map_err(|_| ResponseError::BadRequest("error occurred; try again later".to_string()))
 }
 
-
 pub async fn advanced_analytics<A: Adapter>(
     req: Request<Body>,
     app: &Application<A>,
@@ -107,17 +108,11 @@ pub async fn advanced_analytics<A: Adapter>(
 
     let event_type = serde_urlencoded::from_str::<String>(&req.uri().query().unwrap_or(""))?;
 
-    let response = get_advanced_reports(
-        &app.redis,
-        &event_type,
-        &sess.uid,
-        &advertiser_channels
-    )
-    .await
-    .map_err(|_| ResponseError::BadRequest("error occurred; try again later".to_string()))?;
+    let response = get_advanced_reports(&app.redis, &event_type, &sess.uid, &advertiser_channels)
+        .await
+        .map_err(|_| ResponseError::BadRequest("error occurred; try again later".to_string()))?;
 
     Ok(success_response(serde_json::to_string(&response)?))
-
 }
 
 async fn cache(