feat: add ensure webhook

Signed-off-by: kjuulh <contact@kjuulh.io>

Cargo.lock

@@ -332,6 +332,7 @@ dependencies = [
  "regex",
  "reqwest",
  "serde",
+ "serde_json",
  "sqlx",
  "tokio",
  "tower-http",

crates/contractor/Cargo.toml

@@ -20,3 +20,4 @@ futures = "0.3.30"
 reqwest = {version = "0.12.3", default-features = false, features = ["json", "rustls-tls"]}
 itertools = "0.12.1"
 regex = "1.10.4"
+serde_json = "1.0.115"

crates/contractor/src/main.rs

@@ -26,6 +26,9 @@ enum Commands {
 
         #[arg(long, env = "CONTRACTOR_FILTER")]
         filter: Option<String>,
+
+        #[arg(long = "force-refresh", env = "CONTRACTOR_FORCE_REFRESH")]
+        force_refresh: bool,
     },
 }
 
@@ -64,12 +67,20 @@ async fn main() -> anyhow::Result<()> {
                 result??
             }
         }
-        Some(Commands::Reconcile { user, org, filter }) => {
+        Some(Commands::Reconcile {
+            user,
+            org,
+            filter,
+            force_refresh,
+        }) => {
             tracing::info!("running reconcile");
 
             let state = SharedState::from(Arc::new(State::new().await?));
 
-            state.reconciler().reconcile(user, org, filter).await?;
+            state
+                .reconciler()
+                .reconcile(user, org, filter, force_refresh)
+                .await?;
         }
         None => {}
     }

crates/contractor/src/services/gitea.rs

@@ -76,6 +76,42 @@ impl Default for DefaultGiteaClient {
     }
 }
 
+#[derive(Clone, Debug, Deserialize)]
+pub struct GiteaWebhook {
+    id: isize,
+    #[serde(rename = "type")]
+    r#type: GiteaWebhookType,
+    config: GiteaWebhookConfig,
+}
+#[derive(Clone, Debug, Deserialize)]
+pub struct GiteaWebhookConfig {
+    url: String,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
+pub enum GiteaWebhookType {
+    #[serde(rename = "gitea")]
+    Gitea,
+    Other(String),
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct CreateGiteaWebhook {
+    active: bool,
+    authorization_header: Option<String>,
+    branch_filter: Option<String>,
+    config: CreateGiteaWebhookConfig,
+    events: Vec<String>,
+    #[serde(rename = "type")]
+    r#type: GiteaWebhookType,
+}
+
+#[derive(Clone, Debug, Serialize)]
+pub struct CreateGiteaWebhookConfig {
+    content_type: String,
+    url: String,
+}
+
 impl DefaultGiteaClient {
     pub async fn fetch_user_repos(&self) -> anyhow::Result<Vec<Repository>> {
         //FIXME: We should collect the pages for these queries
@@ -147,6 +183,126 @@ impl DefaultGiteaClient {
             },
         }
     }
+
+    async fn get_webhook(&self, repo: &Repository) -> anyhow::Result<Option<GiteaWebhook>> {
+        let client = reqwest::Client::new();
+
+        let url = format!(
+            "{}/api/v1/repos/{}/{}/hooks",
+            self.url, &repo.owner, &repo.name
+        );
+
+        tracing::trace!("calling url: {}", &url);
+
+        let response = client
+            .get(&url)
+            .header("Content-Type", "application/json")
+            .header("Authorization", format!("token {}", self.token))
+            .send()
+            .await?;
+
+        let webhooks = response.json::<Vec<GiteaWebhook>>().await?;
+
+        let valid_webhooks = webhooks
+            .into_iter()
+            .filter(|w| w.r#type == GiteaWebhookType::Gitea)
+            .filter(|w| w.config.url.contains("contractor"))
+            .collect::<Vec<_>>();
+
+        Ok(valid_webhooks.first().map(|f| f.to_owned()))
+    }
+
+    async fn add_webhook(&self, repo: &Repository) -> anyhow::Result<()> {
+        let client = reqwest::Client::new();
+
+        let url = format!(
+            "{}/api/v1/repos/{}/{}/hooks",
+            self.url, &repo.owner, &repo.name
+        );
+
+        let val = CreateGiteaWebhook {
+            active: true,
+            authorization_header: Some("something".into()),
+            branch_filter: Some("*".into()),
+            config: CreateGiteaWebhookConfig {
+                content_type: "json".into(),
+                url: "https://url?type=contractor".into(),
+            },
+            events: vec!["pull_request_review_comment".into()],
+            r#type: GiteaWebhookType::Gitea,
+        };
+
+        tracing::trace!(
+            "calling url: {} with body {}",
+            &url,
+            serde_json::to_string(&val)?
+        );
+
+        let response = client
+            .post(&url)
+            .header("Content-Type", "application/json")
+            .header("Accept", "application/json")
+            .header("Authorization", format!("token {}", self.token))
+            .json(&val)
+            .send()
+            .await?;
+
+        if let Err(e) = response.error_for_status_ref() {
+            if let Ok(ok) = response.text().await {
+                anyhow::bail!("failed to create webhook: {}, body: {}", e, ok);
+            }
+
+            anyhow::bail!("failed to create webhook: {}", e)
+        }
+
+        Ok(())
+    }
+
+    async fn update_webhook(&self, repo: &Repository, webhook: GiteaWebhook) -> anyhow::Result<()> {
+        let client = reqwest::Client::new();
+
+        let url = format!(
+            "{}/api/v1/repos/{}/{}/hooks/{}",
+            self.url, &repo.owner, &repo.name, &webhook.id,
+        );
+
+        let val = CreateGiteaWebhook {
+            active: true,
+            authorization_header: Some("something".into()),
+            branch_filter: Some("*".into()),
+            config: CreateGiteaWebhookConfig {
+                content_type: "json".into(),
+                url: "https://url?type=contractor".into(),
+            },
+            events: vec!["pull_request_review_comment".into()],
+            r#type: GiteaWebhookType::Gitea,
+        };
+
+        tracing::trace!(
+            "calling url: {} with body {}",
+            &url,
+            serde_json::to_string(&val)?
+        );
+
+        let response = client
+            .patch(&url)
+            .header("Content-Type", "application/json")
+            .header("Accept", "application/json")
+            .header("Authorization", format!("token {}", self.token))
+            .json(&val)
+            .send()
+            .await?;
+
+        if let Err(e) = response.error_for_status_ref() {
+            if let Ok(ok) = response.text().await {
+                anyhow::bail!("failed to create webhook: {}, body: {}", e, ok);
+            }
+
+            anyhow::bail!("failed to create webhook: {}", e)
+        }
+
+        Ok(())
+    }
 }
 
 impl traits::GiteaClient for DefaultGiteaClient {
@@ -178,6 +334,32 @@ impl traits::GiteaClient for DefaultGiteaClient {
 
         Box::pin(async { self.fetch_renovate(repo).await.map(|s| s.is_some()) })
     }
+
+    fn ensure_webhook<'a>(
+        &'a self,
+        repo: &'a Repository,
+        force_refresh: bool,
+    ) -> Pin<Box<dyn futures::prelude::Future<Output = anyhow::Result<()>> + Send + 'a>> {
+        tracing::trace!("ensuring webhook exists for repo: {}", repo);
+
+        Box::pin(async move {
+            match (self.get_webhook(repo).await?, force_refresh) {
+                (Some(_), false) => {
+                    tracing::trace!("webhook already found for {} skipping...", repo);
+                }
+                (Some(webhook), true) => {
+                    tracing::trace!("webhook already found for {} refreshing it", repo);
+                    self.update_webhook(repo, webhook).await?;
+                }
+                (None, _) => {
+                    tracing::trace!("webhook was not found for {} adding", repo);
+                    self.add_webhook(repo).await?;
+                }
+            }
+
+            Ok(())
+        })
+    }
 }
 
 mod extensions;
@@ -186,4 +368,4 @@ pub mod traits;
 use anyhow::Context;
 pub use extensions::*;
 use reqwest::StatusCode;
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};

crates/contractor/src/services/gitea/traits.rs

@@ -19,4 +19,10 @@ pub trait GiteaClient {
         &'a self,
         repo: &'a Repository,
     ) -> Pin<Box<dyn Future<Output = anyhow::Result<bool>> + Send + 'a>>;
+
+    fn ensure_webhook<'a>(
+        &'a self,
+        repo: &'a Repository,
+        force_refresh: bool,
+    ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send + 'a>>;
 }

crates/contractor/src/services/reconciler.rs

@@ -20,6 +20,7 @@ impl Reconciler {
         user: Option<String>,
         orgs: Option<Vec<String>>,
         filter: Option<String>,
+        force_refresh: bool,
     ) -> anyhow::Result<()> {
         let repos = self.get_repos(user, orgs).await?;
         tracing::debug!("found repositories: {}", repos.len());
@@ -56,6 +57,9 @@ impl Reconciler {
             renovate_enabled.len()
         );
 
+        self.ensure_webhook(&renovate_enabled, force_refresh)
+            .await?;
+
         Ok(())
     }
 
@@ -109,6 +113,32 @@ impl Reconciler {
 
         Ok(enabled)
     }
+
+    async fn ensure_webhook(
+        &self,
+        repos: &[Repository],
+        force_refresh: bool,
+    ) -> anyhow::Result<()> {
+        tracing::debug!("ensuring webhooks are setup for repos");
+
+        let mut tasks = FuturesUnordered::new();
+
+        for repo in repos {
+            tasks.push(async move {
+                self.gitea_client
+                    .ensure_webhook(repo, force_refresh)
+                    .await?;
+
+                Ok::<(), anyhow::Error>(())
+            })
+        }
+
+        while let Some(res) = tasks.next().await {
+            res?;
+        }
+
+        Ok(())
+    }
 }
 
 pub trait ReconcilerState {