add: aggr click event, pub get_payout

Author: samparsky

primitives/src/sentry.rs

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

sentry/src/analytics_recorder.rs

@@ -7,7 +7,7 @@ use redis;
 use redis::aio::MultiplexedConnection;
 use slog::{error, Logger};
 
-fn get_payout(channel: &Channel, event: &Event) -> BigNum {
+pub fn get_payout(channel: &Channel, event: &Event) -> BigNum {
     match event {
         Event::Impression { .. } => channel.spec.min_per_impression.clone(),
         Event::Click { .. } => {

sentry/src/db/analytics.rs

@@ -114,6 +114,8 @@ pub async fn get_analytics(
         applied_limit,
     );
 
+    println!("{}", sql_query );
+
     // execute query
     pool.run(move |connection| async move {
         match connection.prepare(&sql_query).await {

sentry/src/db/event_aggregate.rs

@@ -158,16 +158,13 @@ struct EventData {
     earner: Option<ValidatorId>,
     event_count: BigNum,
     event_payout: BigNum,
-    created: DateTime<Utc>,
 }
 
 pub async fn insert_event_aggregate(
     pool: &DbPool,
     channel_id: &ChannelId,
     event: &EventAggregate,
 ) -> Result<bool, RunError<bb8_postgres::tokio_postgres::Error>> {
-    let created = Utc::now();
-
     let mut data: Vec<EventData> = Vec::new();
 
     for (event_type, aggr) in &event.events {
@@ -183,7 +180,6 @@ pub async fn insert_event_aggregate(
                     earner: Some(earner.clone()),
                     event_count: event_count.to_owned(),
                     event_payout: event_payout.clone(),
-                    created,
                 });
 
                 // total sum
@@ -197,7 +193,6 @@ pub async fn insert_event_aggregate(
                 earner: None,
                 event_count: total_event_counts,
                 event_payout: total_event_payouts,
-                created,
             });
         }
     }
@@ -211,10 +206,12 @@ pub async fn insert_event_aggregate(
                     Err(e) => return Err((e, connection))
                 };
 
+                let created = Utc::now(); // time discrepancy
+
                 let writer = BinaryCopyInWriter::new(sink, &[Type::VARCHAR, Type::TIMESTAMPTZ, Type::VARCHAR, Type::VARCHAR, Type::VARCHAR, Type::VARCHAR]);
                 pin_mut!(writer);
                 for item in data {
-                    if let Err(e) = writer.as_mut().write(&[&item.id, &item.created, &item.event_type, &item.event_count, &item.event_payout, &item.earner]).await {
+                    if let Err(e) = writer.as_mut().write(&[&item.id, &created, &item.event_type, &item.event_count, &item.event_payout, &item.earner]).await {
                         err = Some(e);
                         break;
                     }

sentry/src/event_reducer.rs

@@ -1,17 +1,26 @@
 use primitives::sentry::{AggregateEvents, Event, EventAggregate};
-use primitives::{Channel, ValidatorId};
+use primitives::{Channel, ValidatorId, BigNum};
+use crate::analytics_recorder::get_payout;
 
 // @TODO: Remove attribute once we use this function!
 #[allow(dead_code)]
 pub(crate) fn reduce(channel: &Channel, initial_aggr: &mut EventAggregate, ev: &Event) {
     match ev {
         Event::Impression { publisher, .. } => {
             let impression = initial_aggr.events.get("IMPRESSION");
-
-            let merge = merge_impression_ev(impression, &publisher, &channel);
+            let payout = get_payout(&channel, &ev);
+            let merge = merge_impression_ev(impression, &publisher, &payout);
 
             initial_aggr.events.insert("IMPRESSION".to_owned(), merge);
         }
+        Event::Click { publisher, .. } => {
+            let clicks = initial_aggr.events.get("CLICK");
+            let payout = get_payout(&channel, &ev);
+            let merge = merge_impression_ev(clicks, &publisher, &payout);
+
+            initial_aggr.events.insert("CLICK".to_owned(), merge);
+
+        }
         Event::Close => {
             let creator = channel.creator.clone();
             let close_event = AggregateEvents {
@@ -29,7 +38,7 @@ pub(crate) fn reduce(channel: &Channel, initial_aggr: &mut EventAggregate, ev: &
 fn merge_impression_ev(
     impression: Option<&AggregateEvents>,
     earner: &ValidatorId,
-    channel: &Channel,
+    payout: &BigNum
 ) -> AggregateEvents {
     let mut impression = impression.map(Clone::clone).unwrap_or_default();
 
@@ -45,7 +54,7 @@ fn merge_impression_ev(
         .event_payouts
         .entry(earner.clone())
         .or_insert_with(|| 0.into());
-    *event_payouts += &channel.spec.min_per_impression;
+    *event_payouts += payout;
 
     impression
 }