Merge pull request #45 from w3f/updates

lamafab · web-flow · commit 6c7f415fe7f8 · 2023-03-21T16:09:29.000+01:00
Update dependencies and additional changes
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -14,21 +14,21 @@ name = "matrixbot"
 path = "src/main.rs"
 
 [dependencies]
-log = "0.4.14"
-env_logger = "0.9.0"
+log = "0.4.17"
+env_logger = "0.10.0"
+tokio = "1.26.0"
 anyhow = "1.0.43"
-serde = "1.0.129"
-serde_json = "1.0.66"
-serde_yaml = "0.8.20"
+serde = "1.0.158"
+serde_json = "1.0.94"
+serde_yaml = "0.9.19"
 matrix-sdk = "0.3.0"
 ruma = "0.2.0"
-actix = "0.12.0"
-actix-web = "4.0.0-beta.9"
+actix = "0.13.0"
+actix-web = "4.3.1"
 url = "2.2.2"
-async-trait = "0.1.51"
-futures = "0.3.17"
-structopt = "0.3.23"
+async-trait = "0.1.67"
+futures = "0.3.27"
+structopt = "0.3.26"
 md5 = "0.7.0"
-mongodb = { version = "2.0.0-beta", features = ["bson-u2i"] }
-bson = "2.0.0-beta"
-tokio = "1.14.0"
+mongodb =  "2.4.0"
+bson = "2.6.1"
diff --git a/Dockerfile b/Dockerfile
@@ -1,7 +1,7 @@
 # https://github.com/LukeMathWalker/cargo-chef
 # Leveraging the pre-built Docker images with 
 # cargo-chef and the Rust toolchain
-FROM lukemathwalker/cargo-chef:latest-rust-1.55.0 AS chef
+FROM lukemathwalker/cargo-chef:latest-rust-1.68.0 AS chef
 WORKDIR app
 
 FROM chef AS planner
diff --git a/charts/matrixbot-ack/Chart.yaml b/charts/matrixbot-ack/Chart.yaml
@@ -1,4 +1,4 @@
 description: Matrixbot Ack
 name: matrixbot-ack
-version: v0.2.7
+version: v0.2.10
 apiVersion: v2
diff --git a/charts/matrixbot-ack/values.yaml b/charts/matrixbot-ack/values.yaml
@@ -2,7 +2,7 @@ environment: production
 
 image:
   repository: web3f/matrixbot-ack
-  tag: v0.2.7
+  tag: v0.2.10
   pullPolicy: IfNotPresent
 
 config:
diff --git a/src/database.rs b/src/database.rs
@@ -4,15 +4,16 @@ use crate::{unix_time, AlertId, Result};
 // TODO: Can this be avoided somehow?
 use bson::{doc, to_bson};
 use futures::stream::StreamExt;
+use mongodb::IndexModel;
 use mongodb::{
     options::{FindOneAndUpdateOptions, ReplaceOptions, ReturnDocument},
     Client, Database as MongoDb,
 };
 use std::collections::HashMap;
 
-const PENDING: &'static str = "pending";
-const HISTORY: &'static str = "history";
-const ID_CURSOR: &'static str = "id_cursor";
+const PENDING: &str = "pending";
+const HISTORY: &str = "history";
+const ID_CURSOR: &str = "id_cursor";
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DatabaseConfig {
@@ -41,11 +42,31 @@ struct PendingAlertsEntry(HashMap<AlertId, Alert>);
 
 impl Database {
     pub async fn new(config: DatabaseConfig) -> Result<Self> {
-        Ok(Database {
-            db: Client::with_uri_str(config.uri)
-                .await?
-                .database(&config.name),
-        })
+        let db = Client::with_uri_str(config.uri)
+            .await?
+            .database(&config.name);
+
+        // Create index for fields `id` and `last_notified`.
+        let index_model = IndexModel::builder()
+            .keys(doc! {
+                "id": 1,
+                "last_notified": 1,
+            })
+            .build();
+
+        db.collection::<AlertContext>(PENDING)
+            .create_index(index_model, None)
+            .await?;
+
+        Ok(Database { db })
+    }
+    /// Simply checks if a connection could be established to the database.
+    pub async fn connectivity_check(&self) -> Result<()> {
+        self.db
+            .list_collection_names(None)
+            .await
+            .map_err(|err| anyhow!("Failed to connect to database: {:?}", err))
+            .map(|_| ())
     }
     pub async fn insert_alerts(&self, alerts: &[AlertContext]) -> Result<()> {
         if alerts.is_empty() {
@@ -122,8 +143,8 @@ impl Database {
                 history
                     .insert_one(
                         AlertAcknowledged {
-                            alert: alert,
-                            acked_by: acked_by,
+                            alert,
+                            acked_by,
                             acked_timestamp: unix_time(),
                         },
                         None,
@@ -154,7 +175,7 @@ impl Database {
             let now = unix_time();
             doc! {
                 "last_notified": {
-                    "$lt": now - escalation_window,
+                    "$lt": (now - escalation_window) as i64,
                 }
             }
         } else {
diff --git a/src/lib.rs b/src/lib.rs
@@ -7,10 +7,9 @@ extern crate serde;
 #[macro_use]
 extern crate async_trait;
 
-use actix::clock::sleep;
 use actix::{prelude::*, SystemRegistry};
-use std::time::Duration;
 use structopt::StructOpt;
+use tokio::sync::mpsc::unbounded_channel;
 
 mod database;
 mod matrix;
@@ -87,7 +86,7 @@ pub async fn run() -> Result<()> {
         "Opening config at {}",
         std::fs::canonicalize(&cli.config)?
             .to_str()
-            .ok_or(anyhow!("Path to config is not valid unicode"))?
+            .ok_or_else(|| anyhow!("Path to config is not valid unicode"))?
     );
 
     let content = std::fs::read_to_string(&cli.config)?;
@@ -120,14 +119,21 @@ pub async fn run() -> Result<()> {
     let opt_db = if let Some(db_conf) = config.database {
         info!("Setting up database {:?}", db_conf);
         let db = database::Database::new(db_conf).await?;
+        db.connectivity_check().await?;
+
         Some(db)
     } else {
         warn!("Skipping database setup");
         None
     };
 
+    // Setup channels for shutdown signals. The Processor and the API server
+    // task (below) hold the _sender_. Any message sent to it indicates a full shutdown
+    // of the service, which is handled at the end of this function.
+    let (tx, mut recv) = unbounded_channel();
+
     info!("Adding message processor to system registry");
-    let proc = processor::Processor::new(opt_db, escalation_window, should_escalate);
+    let proc = processor::Processor::new(opt_db, escalation_window, should_escalate, tx.clone());
     SystemRegistry::set(proc.start());
 
     info!("Initializing Matrix client");
@@ -137,9 +143,22 @@ pub async fn run() -> Result<()> {
     SystemRegistry::set(matrix.start());
 
     info!("Starting API server");
-    webhook::run_api_server(&config.listener).await?;
-
-    loop {
-        sleep(Duration::from_secs(u64::MAX)).await;
+    let tx_api = tx.clone();
+    let server = webhook::run_api_server(&config.listener).await?;
+
+    // Run server in seperate task, send a shutdown signal in case of an error.
+    tokio::spawn(async move {
+        if let Err(err) = server.await {
+            error!("Failed to run API server: {:?}", err);
+            tx_api.send(()).unwrap();
+        }
+    });
+
+    // On shutdown signal, shutdown service.
+    while let Some(_) = recv.recv().await {
+        warn!("Shutting down service...");
+        return Ok(());
     }
+
+    Ok(())
 }
diff --git a/src/matrix.rs b/src/matrix.rs
@@ -44,7 +44,7 @@ impl MatrixClient {
         let url = Url::parse(&config.homeserver)?;
         let client = Client::new_with_config(url, client_config)?;
 
-        info!("Login with credentials");
+        info!("Logging in with credentials...");
         client
             .login(
                 &config.username,
@@ -79,7 +79,7 @@ impl MatrixClient {
             client
                 .sync_token()
                 .await
-                .ok_or(anyhow!("Failed to acquire sync token"))?,
+                .ok_or_else(|| anyhow!("Failed to acquire sync token"))?,
         );
 
         // Sync in background.
@@ -137,7 +137,7 @@ impl Handler<NotifyAlert> for MatrixClient {
                 return Ok(());
             }
 
-            let current_room_id = rooms.get(0).unwrap_or(rooms.last().unwrap());
+            let current_room_id = rooms.get(0).unwrap_or_else(|| rooms.last().unwrap());
 
             let mut msg = String::from("⚠️ Alert occurred!\n\n");
 
@@ -180,11 +180,11 @@ impl Handler<Escalation> for MatrixClient {
             // Determine which rooms to send the alerts to.
             let current_room_id = rooms
                 .get(notify.escalation_idx.saturating_sub(1))
-                .unwrap_or(rooms.last().unwrap());
+                .unwrap_or_else(|| rooms.last().unwrap());
 
             let next_room_id = rooms
                 .get(notify.escalation_idx)
-                .unwrap_or(rooms.last().unwrap());
+                .unwrap_or_else(|| rooms.last().unwrap());
 
             let is_last = current_room_id == next_room_id;
 
@@ -286,9 +286,11 @@ impl EventHandler for Listener {
                 let cmd = match msg_body.trim() {
                     "pending" => Command::Pending,
                     "help" => Command::Help,
-                    txt @ _ => {
-                        if txt.to_lowercase().starts_with("ack") || txt.to_lowercase().starts_with("acknowledge") {
-                            let parts: Vec<&str> = txt.split(" ").collect();
+                    txt => {
+                        if txt.to_lowercase().starts_with("ack")
+                            || txt.to_lowercase().starts_with("acknowledge")
+                        {
+                            let parts: Vec<&str> = txt.split(' ').collect();
                             if parts.len() == 2 {
                                 if let Ok(id) = AlertId::from_str(parts[1]) {
                                     Command::Ack(id, event.sender.to_string())
@@ -316,7 +318,7 @@ impl EventHandler for Listener {
 
                 // Prepare action type.
                 let action = UserAction {
-                    escalation_idx: escalation_idx,
+                    escalation_idx,
                     command: cmd,
                 };
 
@@ -341,7 +343,7 @@ impl EventHandler for Listener {
 
             match res(room, event.clone()).await {
                 Ok(_) => {}
-                Err(err) => error!("{:?}", err),
+                Err(err) => error!("Error when trying to process Matrix message {:?}", err),
             }
         }
     }
diff --git a/src/processor.rs b/src/processor.rs
diff --git a/src/webhook.rs b/src/webhook.rs
