add: db/event_aggregate.rs, route impl

Author: samparsky

Cargo.lock

@@ -7,6 +7,7 @@ edition = "2018"
 [dependencies]
 # Futures
 futures = "0.3.1"
+async-std = "1.4.0"
 # Primitives
 primitives = { path = "../primitives", features = ["postgres"] }
 adapter = { version = "0.1", path = "../adapter" }
@@ -25,7 +26,7 @@ bb8-postgres = { git = "https://github.com/khuey/bb8", features = ["with-chrono-
 # Migrations
 migrant_lib = { version = "0.27", features = ["d-postgres"] }
 # Logger
-slog = { version = "^2.2.3" , features = ["max_level_trace"] }
+slog = { version = "^2.2.3", features = ["max_level_trace"] }
 # Serde
 serde = { version = "^1.0", features = ['derive'] }
 serde_json = "^1.0"

sentry/Cargo.toml

@@ -28,8 +28,11 @@ CREATE INDEX idx_validator_messages_msg_state_root ON validator_messages ((msg -
 CREATE TABLE event_aggregates
 (
     channel_id VARCHAR(66)              NOT NULL REFERENCES channels (id) ON DELETE RESTRICT,
-    created    TIMESTAMP WITH TIME ZONE NOT NULL,
-    events     JSONB                    NOT NULL
+    created    TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
+    event_type VARCHAR(255)             NOT NULL,
+    earner     VARCHAR(255)             NOT NULL,
+    event_counts TEXT                   NOT NULL,
+    event_payouts TEXT                  NOT NULL,
 );
 
 CREATE INDEX idx_event_aggregates_created ON event_aggregates (created);

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

@@ -1,9 +1,9 @@
 use crate::db::DbPool;
 use bb8::RunError;
 use bb8_postgres::tokio_postgres::types::ToSql;
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, Utc, Date};
 use primitives::sentry::EventAggregate;
-use primitives::ValidatorId;
+use primitives::{ValidatorId, ChannelId};
 
 pub async fn list_event_aggregates(
     pool: &DbPool,
@@ -52,3 +52,51 @@ pub async fn list_event_aggregates(
 
     Ok(event_aggregates)
 }
+
+pub async fn insert_event_aggregate(
+    pool: &DbPool,
+    channel_id: &ChannelId,
+    event: &EventAggregate
+) -> Result<bool, RunError<bb8_postgres::tokio_postgres::Error>>  {
+
+    let mut inserts : Vec<String> = Vec::new();
+    let mut values = Vec::new();
+    let mut index = 0;
+    let id = channel_id.to_string();
+
+    for (event_type, aggr) in event.events {
+        if let Some(event_counts) = aggr.event_counts {
+            for (earner, value) in event_counts {
+                let data =  vec![id.clone(), event_type.clone(), earner, value.to_string(), aggr.event_payouts[&earner].to_string()];
+                inserts.extend(data);
+                //
+                // this is a work around for bulk inserts
+                // rust-postgres does not have native support for bulk inserts
+                // so we have to manually build up a query string dynamically based on
+                // how many things we want to insert
+                //
+                values.push(format!("(${}, ${}, ${}, ${}, ${})", index+1, index+2, index+3, index+4, index+5));
+                index += 5;
+            }
+        }
+    }
+
+    //    the created field is supplied by postgres Default
+     pool
+        .run(
+            move |connection | {
+                async move {
+                    match connection.prepare(&format!("INSERT INTO event_aggregates (channel_id, event_type, earner, event_counts, event_payouts) values {}", values.join(" ,"))).await {
+                        Ok(stmt) => match connection.execute(&stmt, &inserts.as_slice()).await {
+                            Ok(row) => {
+                                let inserted = row == (index / 5);
+                                Ok((inserted, connection))
+                            },
+                            Err(e) => Err((e, connection)),
+                        },
+                        Err(e) => Err((e, connection)),
+                    }
+                }
+            }
+        ).await
+}

sentry/src/db/event_aggregate.rs

@@ -0,0 +1,69 @@
+use crate::event_reducer;
+use primitives::{ChannelId, Channel};
+use primitives::adapter::Adapter;
+use primitives::sentry::{EventAggregate, Event};
+use std::collections::HashMap;
+use futures::future::BoxFuture;
+use crate::Session;
+use crate::access::check_access;
+use crate::Application;
+use chrono::{Utc, Duration};
+use async_std::stream;
+
+// use futures::
+use async_std::sync::RwLock;
+use std::sync::{Arc};
+
+pub struct EventAggregator {
+    // recorder: HashMap<ChannelId, FnMut(&Session, &str) -> BoxFuture>,
+    aggregate: Arc<RwLock<HashMap<ChannelId, EventAggregate>>>
+}
+
+fn persist(aggr_throttle: i64, channel_id: ChannelId, aggr: Arc<RwLock<HashMap<ChannelId, EventAggregate>>>) {
+    let mut interval = stream::interval(Duration::from_secs(aggr_throttle));
+    while let Some(_) = interval.next().await {
+//        loop through the keys and persist them in the
+//        database
+    }
+}
+
+impl EventAggregator {
+    pub async fn record<A: Adapter>(
+        &self,
+        app: &Application<A>,
+        channel: &Channel,
+        session: &Session,
+        events: &[Event]
+        )
+    {
+        // eventAggrCol
+        // channelsCol
+        // try getting aggr if none create and store new aggr
+    //     redis: &MultiplexedConnection,
+    // session: &Session,
+    // rate_limit: &RateLimit,
+    // channel: &Channel,
+    // events: &[Event]
+        let has_access = check_access(&app.redis, session, &app.config.ip_rate_limit, channel, events).await;
+        if has_access.is_err() {
+            // return the error
+        }
+        let recorder = self.aggregate.write().await.expect("should acquire lock");
+        let mut aggr: EventAggregate = *recorder.get_mut(&channel.id);
+        // if aggr is none
+        // spawn a tokio task for saving to database
+        events.iter().for_each( | ev| event_reducer::reduce(channel, &mut aggr, ev));
+
+        if app.config.aggr_throttle > 0
+            &&
+            Utc::now() < (aggr.created + Duration::seconds(app.config.aggr_throttle as i64))
+        {
+
+        }
+
+
+
+
+
+    }
+}

sentry/src/event_aggregator.rs

@@ -39,6 +39,8 @@ pub mod access;
 mod chain;
 pub mod db;
 pub mod event_reducer;
+pub mod event_aggregator;
+
 
 lazy_static! {
     static ref CHANNEL_GET_BY_ID: Regex =
@@ -50,6 +52,7 @@ lazy_static! {
     static ref CHANNEL_EVENTS_AGGREGATES: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/events-aggregates/?$").expect("The regex should be valid");
     static ref ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");
     static ref ADVERTISER_ANALYTICS_BY_CHANNEL_ID: Regex = Regex::new(r"^/analytics/for-advertiser/0x([a-zA-Z0-9]{64})/?$").expect("The regex should be valid");
+    static ref CREATE_EVENTS_BY_CHANNEL_ID: Regex = Regex::new(r"^/channel/0x([a-zA-Z0-9]{64})/events(/.*)?$").expect("The regex should be valid");
 }
 
 fn auth_required_middleware<'a, A: Adapter>(
@@ -262,6 +265,8 @@ async fn channels_router<A: Adapter + 'static>(
 
         list_channel_event_aggregates(req, app).await
     } else {
+    // else if let (Some(caps), &Method::POST) = (CREATE_EVENTS_BY_CHANNEL_ID.captures(&path) ,method) {
+    //} else {
         Err(ResponseError::NotFound)
     }
 }

sentry/src/lib.rs

@@ -7,7 +7,7 @@ use crate::RouteParams;
 use hex::FromHex;
 use hyper::{Body, Request, Response};
 use primitives::adapter::Adapter;
-use primitives::sentry::SuccessResponse;
+use primitives::sentry::{SuccessResponse, Event};
 use primitives::{Channel, ChannelId};
 use slog::error;
 
@@ -102,6 +102,34 @@ pub async fn last_approved<A: Adapter>(
         .unwrap())
 }
 
+pub async fn events<A: Adapter>(
+    req: Request<Body>,
+    app: &Application<A>,
+) -> Result<Response<Body>, ResponseError> {
+    let ip = if let Some(xforwardedfor) =  req.headers().get("x-forwarded-for") {
+        let ip: Vec<&str> = xforwardedfor.to_str()?.split(',').collect();
+        Some(ip[0])
+    } else if let Some(trueip) = req.headers().get("true-client-ip") {
+        Some(trueip.to_str()?)
+    } else {
+        None
+    };
+
+    let body = hyper::body::to_bytes(req.into_body()).await?;
+    let events = serde_json::from_slice::<Vec<Event>>(&body)?;
+
+    
+    
+
+
+
+    Ok(Response::builder()
+        .header("Content-type", "application/json")
+        .body(serde_json::to_string(&events)?.into())
+        .unwrap())
+
+}
+
 mod channel_list {
     use chrono::serde::ts_seconds::deserialize as ts_seconds;
     use chrono::{DateTime, Utc};