add: exhausted impl sentry

Author: samparsky

adapter/src/ethereum.rs

@@ -666,6 +666,7 @@ mod test {
                 ad_units: vec![],
                 pricing_bounds: None,
             },
+            exhausted: Default::default(),
         };
 
         // convert to eth channel

primitives/src/channel.rs

@@ -232,6 +232,16 @@ impl SpecValidators {
         }
     }
 
+    pub fn find_index(&self, validator_id: &ValidatorId) -> i32 {
+        if &self.leader().id == validator_id {
+            0
+        } else if &self.follower().id == validator_id {
+            1
+        } else {
+            -1
+        }
+    }
+
     pub fn iter(&self) -> Iter<'_> {
         Iter::new(&self)
     }

sentry/src/db/channel.rs

@@ -96,6 +96,30 @@ pub async fn insert_validator_messages(
         .await
 }
 
+pub async fn update_exhausted_channel(
+    pool: &DbPool,
+    channel: &Channel,
+    index: i32,
+) -> Result<bool, RunError<bb8_postgres::tokio_postgres::Error>> {
+    pool
+        .run(move | connection| {
+            async move {
+                match connection.prepare("UPDATE channels SET exhausted[$1] = true WHERE id = $2").await {
+                    Ok(stmt) => match connection.execute(&stmt, &[&index, &channel.id]).await {
+                        Ok(row) => {
+                            let updated = row == 1;
+                            Ok((updated, connection))
+                        },
+                        Err(e) => Err((e, connection)),
+                    },
+                    Err(e) => Err((e, connection)),
+                }
+            }
+        })
+        .await
+
+}
+
 mod list_channels {
     use crate::db::DbPool;
     use bb8::RunError;

sentry/src/routes/channel.rs

@@ -1,5 +1,5 @@
 use crate::db::event_aggregate::{latest_approve_state, latest_heartbeats, latest_new_state};
-use crate::db::{get_channel_by_id, insert_channel, insert_validator_messages, list_channels};
+use crate::db::{get_channel_by_id, insert_channel, insert_validator_messages, list_channels, update_exhausted_channel};
 use crate::{success_response, Application, Auth, ResponseError, RouteParams, Session};
 use bb8::RunError;
 use bb8_postgres::tokio_postgres::error;
@@ -238,6 +238,13 @@ pub async fn create_validator_messages<A: Adapter + 'static>(
         .get("messages")
         .ok_or_else(|| ResponseError::BadRequest("missing messages body".to_string()))?;
 
+    let channel_is_exhausted = messages.iter().any(|message| match message {
+        MessageTypes::ApproveState(approve) => approve.exhausted,
+        MessageTypes::NewState(new_state) => new_state.exhausted,
+        _ => false,
+    });
+
+
     match channel.spec.validators.find(&session.uid) {
         None => Err(ResponseError::Unauthorized),
         _ => {
@@ -246,6 +253,17 @@ pub async fn create_validator_messages<A: Adapter + 'static>(
             }))
             .await?;
 
+            if channel_is_exhausted {
+                // can never be -1
+                let validator_index = channel.spec.validators.find_index(&session.uid);                
+                update_exhausted_channel(
+                    &app.pool,
+                    &channel,
+                    validator_index
+                ).await?;
+            }
+
+
             Ok(success_response(serde_json::to_string(&SuccessResponse {
                 success: true,
             })?))