V5: Authenticated POST route to modify/submit campaigns

primitives/src/sentry.rs

@@ -379,6 +379,18 @@ pub mod campaign_create {
             }
         }
     }
+
+    // All editable fields stored in one place, used for checking when a budget is changed
+    // #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+    // pub struct CampaignState {
+    //     pub budget: UnifiedNum,
+    //     pub validators: Validators,
+    //     pub title: Option<String>,
+    //     pub pricing_bounds: Option<PricingBounds>,
+    //     pub event_submission: Option<EventSubmission>,
+    //     pub ad_units: Vec<AdUnit>,
+    //     pub targeting_rules: Rules,
+    // }
 }
 
 #[cfg(feature = "postgres")]

sentry/src/db.rs

@@ -6,7 +6,7 @@ use tokio_postgres::NoTls;
 use lazy_static::lazy_static;
 
 pub mod analytics;
-mod campaign;
+pub mod campaign;
 mod channel;
 pub mod event_aggregate;
 pub mod spendable;

sentry/src/db/campaign.rs

@@ -2,8 +2,6 @@ use crate::db::{DbPool, PoolError};
 use primitives::Campaign;
 use tokio_postgres::types::Json;
 
-// TODO: Remove once we use this fn
-#[allow(dead_code)]
 pub async fn insert_campaign(pool: &DbPool, campaign: &Campaign) -> Result<bool, PoolError> {
     let client = pool.get().await?;
     let ad_units = Json(campaign.ad_units.clone());
@@ -38,9 +36,7 @@ pub async fn insert_campaign(pool: &DbPool, campaign: &Campaign) -> Result<bool,
 /// SELECT id, channel, creator, budget, validators, title, pricing_bounds, event_submission, ad_units, targeting_rules, created, active_from, active_to FROM campaigns
 /// WHERE id = $1
 /// ```
-// TODO: Remove once we use this fn
-#[allow(dead_code)]
-pub async fn fetch_campaign(pool: DbPool, campaign: &Campaign) -> Result<Campaign, PoolError> {
+pub async fn fetch_campaign(pool: &DbPool, campaign: &Campaign) -> Result<Campaign, PoolError> {
     let client = pool.get().await?;
     let statement = client.prepare("SELECT id, channel, creator, budget, validators, title, pricing_bounds, event_submission, ad_units, targeting_rules, created, active_from, active_to FROM campaigns WHERE id = $1").await?;
 
@@ -49,6 +45,51 @@ pub async fn fetch_campaign(pool: DbPool, campaign: &Campaign) -> Result<Campaig
     Ok(Campaign::from(&row))
 }
 
+pub async fn get_campaigns_for_channel(pool: &DbPool, campaign: &Campaign) -> Result<Vec<Campaign>, PoolError> {
+    let client = pool.get().await?;
+    let statement = client.prepare("SELECT id, channel, creator, budget, validators, title, pricing_bounds, event_submission, ad_units, targeting_rules, created, active_from, active_to FROM campaigns WHERE channel_id = $1").await?;
+
+    let row = client.query(&statement, &[&campaign.channel.id()]).await?;
+
+    let campaigns = row.into_iter().map(|c| Campaign::from(&c)).collect();
+    Ok(campaigns)
+}
+
+pub async fn campaign_exists(pool: &DbPool, campaign: &Campaign) -> Result<bool, PoolError> {
+    let client = pool.get().await?;
+    let statement = client
+        .prepare("SELECT EXISTS(SELECT 1 FROM campaigns WHERE id = $1)")
+        .await?;
+
+    let row = client.execute(&statement, &[&campaign.id]).await?;
+
+    let exists = row == 1;
+    Ok(exists)
+}
+
+pub async fn update_campaign_in_db(pool: &DbPool, campaign: &Campaign) -> Result<bool, PoolError> {
+    let client = pool.get().await?;
+    let statement = client
+        .prepare("UPDATE campaigns SET budget = $1, validators = $2, title = $3, pricing_bounds = $4, event_submission = $5, ad_units = $6, targeting_rules = $7 WHERE id = $8")
+        .await?;
+
+    let row = client
+        .execute(&statement, &[
+            &campaign.budget,
+            &campaign.validators,
+            &campaign.title,
+            &campaign.pricing_bounds,
+            &campaign.event_submission,
+            &campaign.ad_units,
+            &campaign.targeting_rules,
+            &campaign.id,
+        ])
+        .await?;
+
+    let exists = row == 1;
+    Ok(exists)
+}
+
 #[cfg(test)]
 mod test {
     use primitives::{campaign::Campaign, util::tests::prep_db::DUMMY_CAMPAIGN};
@@ -72,7 +113,12 @@ mod test {
 
         assert!(is_inserted);
 
-        let fetched_campaign: Campaign = fetch_campaign(db_pool.clone(), &campaign_for_testing)
+        let exists = campaign_exists(&db_pool.clone(), &campaign_for_testing)
+            .await
+            .expect("Should succeed");
+        assert!(exists);
+
+        let fetched_campaign: Campaign = fetch_campaign(&db_pool, &campaign_for_testing)
             .await
             .expect("Should fetch successfully");
 

sentry/src/db/event_aggregate.rs

@@ -3,7 +3,7 @@ use futures::pin_mut;
 use primitives::{
     sentry::{EventAggregate, MessageResponse},
     validator::{ApproveState, Heartbeat, NewState},
-    Address, BigNum, Channel, ChannelId, ValidatorId,
+    Address, BigNum, Channel, ChannelId, ValidatorId, channel_v5::Channel as ChannelV5,
 };
 use std::{convert::TryFrom, ops::Add};
 use tokio_postgres::{
@@ -58,6 +58,31 @@ pub async fn latest_new_state(
         .map_err(PoolError::Backend)
 }
 
+pub async fn latest_new_state_v5(
+    pool: &DbPool,
+    channel: &ChannelV5,
+    state_root: &str,
+) -> Result<Option<MessageResponse<NewState>>, PoolError> {
+    let client = pool.get().await?;
+
+    let select = client.prepare("SELECT \"from\", msg, received FROM validator_messages WHERE channel_id = $1 AND \"from\" = $2 AND msg ->> 'type' = 'NewState' AND msg->> 'stateRoot' = $3 ORDER BY received DESC LIMIT 1").await?;
+    let rows = client
+        .query(
+            &select,
+            &[
+                &channel.id(),
+                &channel.leader,
+                &state_root,
+            ],
+        )
+        .await?;
+
+    rows.get(0)
+        .map(MessageResponse::<NewState>::try_from)
+        .transpose()
+        .map_err(PoolError::Backend)
+}
+
 pub async fn latest_heartbeats(
     pool: &DbPool,
     channel_id: &ChannelId,

sentry/src/lib.rs

@@ -21,6 +21,7 @@ use primitives::{Config, ValidatorId};
 use redis::aio::MultiplexedConnection;
 use regex::Regex;
 use routes::analytics::{advanced_analytics, advertiser_analytics, analytics, publisher_analytics};
+use routes::campaign::{create_campaign, update_campaign};
 use routes::cfg::config;
 use routes::channel::{
     channel_list, channel_validate, create_channel, create_validator_messages, insert_events,
@@ -34,6 +35,7 @@ pub mod routes {
     pub mod analytics;
     pub mod cfg;
     pub mod channel;
+    pub mod campaign;
     pub mod event_aggregate;
     pub mod validator_message;
 }
@@ -150,6 +152,28 @@ impl<A: Adapter + 'static> Application<A> {
 
                 publisher_analytics(req, &self).await
             }
+            // For creating campaigns
+            ("/campaign", &Method::POST) => {
+                let req = match AuthRequired.call(req, &self).await {
+                    Ok(req) => req,
+                    Err(error) => {
+                        return map_response_error(error);
+                    }
+                };
+
+                create_campaign(req, &self).await
+            }
+            // For editing campaigns
+            ("/campaign/:id", &Method::POST) => {
+                let req = match AuthRequired.call(req, &self).await {
+                    Ok(req) => req,
+                    Err(error) => {
+                        return map_response_error(error);
+                    }
+                };
+
+                update_campaign(req, &self).await
+            }
             (route, _) if route.starts_with("/analytics") => analytics_router(req, &self).await,
             // This is important becuase it prevents us from doing
             // expensive regex matching for routes without /channel