update: insert event aggregate total

Author: samparsky

sentry/Cargo.toml

@@ -23,6 +23,8 @@ regex = "1"
 redis = { version = "0.14", features = ["tokio-rt-core"] }
 bb8 = { git = "https://github.com/khuey/bb8" }
 bb8-postgres = { git = "https://github.com/khuey/bb8", features = ["with-chrono-0_4", "with-serde_json-1"] }
+postgres-types = { version = "0.1.0", features = ["derive"] }
+
 # Migrations
 migrant_lib = { version = "0.27", features = ["d-postgres"] }
 # Logger

sentry/migrations/20190806011140_initial-tables/up.sql

@@ -30,9 +30,10 @@ CREATE TABLE event_aggregates
     channel_id VARCHAR(66)              NOT NULL REFERENCES channels (id) ON DELETE RESTRICT,
     created    TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
     event_type VARCHAR(255)             NOT NULL,
-    earner     VARCHAR(255)             NOT NULL,
+    earner     VARCHAR(255)                    ,
     event_counts VARCHAR                   NOT NULL,
     event_payouts VARCHAR                  NOT NULL,
 );
 
 CREATE INDEX idx_event_aggregates_created ON event_aggregates (created);
+CREATE INDEX idx_event_aggregates_channel ON event_aggregates (created);

sentry/src/db/event_aggregate.rs

@@ -1,9 +1,12 @@
 use crate::db::DbPool;
 use bb8::RunError;
-use bb8_postgres::tokio_postgres::types::ToSql;
+// use bb8_postgres::tokio_postgres::types::{ToSql, FromSql};
 use chrono::{DateTime, Utc};
 use primitives::sentry::EventAggregate;
 use primitives::{ChannelId, ValidatorId};
+use primitives::BigNum;
+use std::ops::Add;
+use postgres_types::{ToSql, FromSql};
 
 pub async fn list_event_aggregates(
     pool: &DbPool,
@@ -53,6 +56,16 @@ pub async fn list_event_aggregates(
     Ok(event_aggregates)
 }
 
+#[derive(Debug, ToSql, FromSql)]
+struct EventData {
+    id: String,
+    event_type: String,
+    earner: Option<String>,
+    event_count: String,
+    event_payout: String,
+}
+
+
 pub async fn insert_event_aggregate(
     pool: &DbPool,
     channel_id: &ChannelId,
@@ -62,21 +75,29 @@ pub async fn insert_event_aggregate(
     let mut index = 0;
     let id = channel_id.to_string();
 
-    let mut data: Vec<String> = Vec::new();
+    let mut data: Vec<EventData> = Vec::new();
 
     for (event_type, aggr) in &event.events {
         if let Some(event_counts) = &aggr.event_counts {
-            for (earner, value) in event_counts {
-                let event_count = value.to_string();
-                let event_payout = aggr.event_payouts[earner].to_string();
-
+            let mut total_event_counts: BigNum = 0.into();
+            let mut total_event_payouts: BigNum = 0.into();
+            for (earner, event_count) in event_counts {
+                let event_payout = aggr.event_payouts[earner].clone();
                 data.extend(vec![
-                    id.clone(),
-                    event_type.clone(),
-                    earner.clone(),
-                    event_count,
-                    event_payout,
+                    EventData {
+                        id: id.clone(),
+                        event_type: event_type.clone(),
+                        earner: Some(earner.clone()),
+                        event_count: event_count.to_string(),
+                        event_payout: event_payout.to_string(),
+                    }
+                    
                 ]);
+
+                // total sum 
+                total_event_counts = event_count.add(&total_event_counts);
+                total_event_payouts = total_event_payouts.add(event_payout);
+
                 //
                 // this is a work around for bulk inserts
                 // rust-postgres does not have native support for bulk inserts
@@ -95,6 +116,28 @@ pub async fn insert_event_aggregate(
                 ));
                 index += 5;
             }
+            // extend with
+            data.extend(vec![
+                EventData {
+                    id: id.clone(),
+                    event_type: event_type.clone(),
+                    earner: None,
+                    event_count: total_event_counts.to_string(),
+                    event_payout: total_event_payouts.to_string(),
+                }
+            ]);
+
+            values.push(format!(
+                "(${}, ${}, ${}, ${}, ${})",
+                index + 1,
+                index + 2,
+                index + 3,
+                index + 4,
+                index + 5
+            ));
+            index += 5;
+
+
         }
     }