diff --git a/Cargo.lock b/Cargo.lock index 23d81de..e7b6117 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -332,6 +332,7 @@ dependencies = [ "regex", "reqwest", "serde", + "serde_json", "sqlx", "tokio", "tower-http", diff --git a/crates/contractor/Cargo.toml b/crates/contractor/Cargo.toml index d80450f..edde31c 100644 --- a/crates/contractor/Cargo.toml +++ b/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" diff --git a/crates/contractor/src/main.rs b/crates/contractor/src/main.rs index d35415f..e2f513b 100644 --- a/crates/contractor/src/main.rs +++ b/crates/contractor/src/main.rs @@ -26,6 +26,9 @@ enum Commands { #[arg(long, env = "CONTRACTOR_FILTER")] filter: Option, + + #[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 => {} } diff --git a/crates/contractor/src/services/gitea.rs b/crates/contractor/src/services/gitea.rs index 81cb39e..05ad799 100644 --- a/crates/contractor/src/services/gitea.rs +++ b/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, + branch_filter: Option, + config: CreateGiteaWebhookConfig, + events: Vec, + #[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> { //FIXME: We should collect the pages for these queries @@ -147,6 +183,126 @@ impl DefaultGiteaClient { }, } } + + async fn get_webhook(&self, repo: &Repository) -> anyhow::Result> { + 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::>().await?; + + let valid_webhooks = webhooks + .into_iter() + .filter(|w| w.r#type == GiteaWebhookType::Gitea) + .filter(|w| w.config.url.contains("contractor")) + .collect::>(); + + 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> + 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}; diff --git a/crates/contractor/src/services/gitea/traits.rs b/crates/contractor/src/services/gitea/traits.rs index 84efdc8..5394c40 100644 --- a/crates/contractor/src/services/gitea/traits.rs +++ b/crates/contractor/src/services/gitea/traits.rs @@ -19,4 +19,10 @@ pub trait GiteaClient { &'a self, repo: &'a Repository, ) -> Pin> + Send + 'a>>; + + fn ensure_webhook<'a>( + &'a self, + repo: &'a Repository, + force_refresh: bool, + ) -> Pin> + Send + 'a>>; } diff --git a/crates/contractor/src/services/reconciler.rs b/crates/contractor/src/services/reconciler.rs index 5c0d65c..cc1558b 100644 --- a/crates/contractor/src/services/reconciler.rs +++ b/crates/contractor/src/services/reconciler.rs @@ -20,6 +20,7 @@ impl Reconciler { user: Option, orgs: Option>, filter: Option, + 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 {