Merge pull request 'Make it webhook based instead of polling API every 2.5 seconds' (#1) from webhook-based into master

Reviewed-on: #1

.env.example

@@ -1,5 +1,5 @@
 TG_KEY=
-GITEA_KEY=
-GITEA_URL=https://git.blek.codes
 CHAT_ID=
-POLL_TIME=2000
+
+WH_SECRET=
+WH_URL=0.0.0.0:80

Cargo.toml

@@ -6,10 +6,13 @@ edition = "2021"
 [dependencies]
 dotenvy_macro = "0.15.7"
 femme = "2.2.1"
+lazy_static = "1.4.0"
 log = "0.4.19"
 reqwest = "0.11.18"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0.104"
 teloxide = "0.12.2"
 teloxide-core = "0.9.1"
+tide = "0.16.0"
 tokio = { version = "1.29.1", features = [ "full" ] }
+async-std = { version = "1", features = ["attributes", "tokio1"] }

src/api.rs

@@ -1,11 +1,5 @@
 
-use crate::{GITEA_KEY, gitea::{User, Repo, Pull}};
+use crate::gitea::{User, Repo, Pull};
-
-macro_rules! gitea_api {
-    ($path: expr) => {
-        format!("{}{}?token={}", dotenvy_macro::dotenv!("GITEA_URL", "Gitea url is not set!"), $path, GITEA_KEY)
-    };
-}
 
 pub async fn get_user() -> User {
     let user = reqwest::get(gitea_api!("/api/v1/user")).await.unwrap();

src/gitea.rs

@@ -40,4 +40,13 @@ pub struct Pull {
     pub head: Branch,
     pub merge_base: String,
     pub user: User
+}
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct PullWh {
+    pub action: String,
+    pub number: i64,
+    pub sender: User,
+    pub repository: Repo,
+    pub pull_request: Pull
 }

src/main.rs

@@ -1,20 +1,116 @@
 #![forbid(unsafe_code)]
-
+#![feature(new_uninit)]
-use std::{path::Path, thread, time::Duration};
 
 use log;
-use teloxide_core::{prelude::*, types::Recipient};
+use teloxide_core::{prelude::*, types::{Recipient, Chat}};
 use teloxide::Bot;
-use tokio::fs;
+use tide::{Server, Request, Response};
 
 mod gitea;
-mod api;
 
-const POLL_TIME: &str = dotenvy_macro::dotenv!("POLL_TIME", "Poll time is not set!");
 const CHAT_ID: &str = dotenvy_macro::dotenv!("CHAT_ID", "Chat ID is not set!");
 const TG_KEY: &str = dotenvy_macro::dotenv!("TG_KEY", "Telegram key is not set!");
-const GITEA_KEY: &str = dotenvy_macro::dotenv!("GITEA_KEY", "Gitea key is not set!");
+
-const GITEA_URL: &str = dotenvy_macro::dotenv!("GITEA_URL", "Gitea url is not set!");
+const WH_SECRET: &str = dotenvy_macro::dotenv!("WH_SECRET", "Webhook secret is not set!");
+const WH_LISTEN_URL: &str = dotenvy_macro::dotenv!("WH_URL", "Webhook listen URL is not set!");
+
+#[derive(Debug, Clone)]
+struct SharedState {
+    bot: Bot,
+    chat: Chat
+}
+
+async fn start_tg() -> (Bot, Chat) {
+    log::info!("Starting up telegram bot...");
+    let bot = Bot::new(TG_KEY);
+    let me = bot.get_me().await.unwrap();
+    log::info!("Logged in to telegram as \"{}\"", me.first_name);
+    
+    let chat = bot.get_chat(Recipient::Id(ChatId(CHAT_ID.parse::<i64>().unwrap()))).await.unwrap();
+    log::info!("Using chat {}", {
+        if chat.is_group() {
+            chat.title().unwrap()
+        } else {
+            chat.first_name().unwrap()
+        }
+    });
+
+    (bot, chat)
+}
+
+async fn start_http(state: SharedState) -> Server<SharedState> {
+    let mut app = tide::with_state(state);
+
+    app.at("/pull_wh").post(webhook);
+
+    app
+}
+
+async fn webhook(mut req: Request<SharedState>) -> tide::Result {
+
+    let body = req.body_string().await.unwrap();
+    let pr = serde_json::from_str::<gitea::PullWh>(body.as_str());
+
+    let event_type = req.header("X-Gitea-Event-Type");
+    if event_type.is_none() {
+        return Ok(
+            Response::builder(400)
+                .body("{\"error\":\"no event type (X-Gitea-Event-Type)\"}")
+                .content_type("application/json")
+                .build()
+        )
+    }
+
+    let event_type = event_type.unwrap().get(0).unwrap().to_string();
+    if event_type != "pull_request" {
+        return Ok(
+            Response::builder(200)
+                .body("{\"status\":\"ignoring non-pr event\"}")
+                .content_type("application/json")
+                .build()
+        )
+    }
+
+    if pr.is_err() {
+        return Ok(format!("Bad serialization: {}", pr.unwrap_err().to_string()).into());
+    }
+
+    let pr = pr.unwrap();
+
+    let secret = req.header("Authorization");
+    if secret.is_none() {
+        return Ok(
+            Response::builder(401)
+            //         { error: "bad auth" }
+                .body("{\"error\":\"bad auth\"}")
+                .content_type("application/json")
+                .build()
+        )
+    }
+    let secret = secret.unwrap().get(0).unwrap().to_string();
+
+    if secret != "Bearer ".to_string() + WH_SECRET {
+        println!("{} != {}", secret, "Bearer ".to_string() + WH_SECRET);
+        return Ok(
+            Response::builder(401)
+                .body("{\"error\":\"bad auth\"}")
+                .content_type("application/json")
+                .build()
+        )
+    }
+
+    let state = req.state().clone();
+
+    state.bot.send_message(state.chat.id, format!("New PR\n{} ({}#{}) by {}\n{}", pr.pull_request.title, pr.pull_request.head.repo.name, pr.pull_request.number, pr.pull_request.user.login, pr.pull_request.url)).await.unwrap();
+    state.bot.send_message(state.chat.id, "апрувните пж @bleki42 @balistiktw @x3paerz").await.unwrap();
+
+    Ok(
+        Response::builder(200)
+                .body("{ \"status\": \"sent\" }")
+                .content_type("application/json")
+                .build()
+    )
+}
 
 #[tokio::main]
 async fn main() {
@@ -27,54 +123,11 @@ async fn main() {
         log::info!("Running in production");
     }
 
-    log::info!("Gitea URL: {}", GITEA_URL);
+    let (bot, chat) = start_tg().await;
+    let http = start_http(SharedState { bot, chat }).await;
 
-    let user = api::get_user().await;
+    log::info!("Listening for webhooks on {}", WH_LISTEN_URL);
-    log::info!("Logged into gitea as user \"{}\"", user.full_name);
 
-    let repos = api::get_org_repos("Tochka".into()).await;
+    http.listen(WH_LISTEN_URL.clone()).await.unwrap();
-    log::info!("Found {} repositories", repos.len());
-
-    let repos = api::filter_repos(repos);
-    log::info!("Only  {} repositories are useful", repos.len());
-
-    if !Path::new(".pr-cache").exists() {
-        let prs = api::get_all_prs(repos.clone()).await;
-        let data = serde_json::to_string_pretty(&prs).unwrap();
-        fs::write(".pr-cache", data).await.unwrap();
-    }
-
-    let mut prs = serde_json::from_str::<Vec<gitea::Pull>>(String::from_utf8(fs::read(".pr-cache").await.unwrap()).unwrap().as_str()).unwrap();
-    log::info!("Got {} PRs", prs.len());
-
-    log::info!("Starting up telegram bot...");
-    let bot = Bot::new(TG_KEY);
-    let me = bot.get_me().await.unwrap();
-    log::info!("Logged in to telegram as \"{}\"", me.first_name);
-    
-    let chat = bot.get_chat(Recipient::Id(ChatId(CHAT_ID.parse::<i64>().unwrap()))).await.unwrap();
-    log::info!("Using chat {}", chat.title().unwrap());
-
-    let time = POLL_TIME.parse::<u64>().unwrap();
-
-    log::info!("Starting polling SCM for updates ({})", time);
-    
-    loop {
-        let (new_prs, new_data) = api::get_new_prs(prs.clone(), repos.clone()).await;
-        log::debug!("Found {} new PRs", new_prs.len());
-        thread::sleep(Duration::from_millis(time));
-
-        if new_prs.len() != 0 {
-            prs = new_data;
-            let data = serde_json::to_string_pretty(&prs).unwrap();
-            fs::write(".pr-cache", data).await.unwrap();
-
-            for pr in new_prs.iter() {
-                log::info!("Sending message about {} ({})", pr.title, pr.url);
-                bot.send_message(chat.id, format!("New PR\n{} ({}#{}) by {}\n{}", pr.title, pr.head.repo.name, pr.number, pr.user.login, pr.url)).await.unwrap();
-                bot.send_message(chat.id, "апрувните пж @bleki42 @balistiktw @x3paerz").await.unwrap();
-            }
-        }
-    }
 
 }