Merge pull request #70 from vohoanglong0107/refactor-global-event

refactor: split global event from per repo ones

Author: Kobzol

src/bin/bors.rs

@@ -5,8 +5,8 @@ use std::time::Duration;
 
 use anyhow::Context;
 use bors::{
-    create_app, create_bors_process, BorsContext, BorsEvent, CommandParser, GithubAppState,
-    SeaORMClient, ServerState, WebhookSecret,
+    create_app, create_bors_process, BorsContext, BorsEvent, BorsGlobalEvent, CommandParser,
+    GithubAppState, SeaORMClient, ServerState, WebhookSecret,
 };
 use clap::Parser;
 use sea_orm::Database;
@@ -85,7 +85,9 @@ fn try_main(opts: Opts) -> anyhow::Result<()> {
     let refresh_process = async move {
         loop {
             tokio::time::sleep(PERIODIC_REFRESH).await;
-            refresh_tx.send(BorsEvent::Refresh).await?;
+            refresh_tx
+                .send(BorsEvent::Global(BorsGlobalEvent::Refresh))
+                .await?;
         }
     };
 

src/bors/event.rs

@@ -3,7 +3,7 @@ use crate::github::{CommitSha, GithubRepoName, GithubUser, PullRequestNumber};
 use octocrab::models::RunId;
 
 #[derive(Debug)]
-pub enum BorsEvent {
+pub enum BorsRepositoryEvent {
     /// A comment was posted on a pull request.
     Comment(PullRequestComment),
     /// A workflow run on Github Actions or a check run from external CI system has been started.
@@ -13,12 +13,35 @@ pub enum BorsEvent {
     /// A check suite has been completed, either as a workflow run on Github Actions, or as a
     /// workflow from some external CI system.
     CheckSuiteCompleted(CheckSuiteCompleted),
+}
+
+impl BorsRepositoryEvent {
+    pub fn repository(&self) -> &GithubRepoName {
+        match self {
+            BorsRepositoryEvent::Comment(comment) => &comment.repository,
+            BorsRepositoryEvent::WorkflowStarted(workflow) => &workflow.repository,
+            BorsRepositoryEvent::WorkflowCompleted(workflow) => &workflow.repository,
+            BorsRepositoryEvent::CheckSuiteCompleted(payload) => &payload.repository,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum BorsGlobalEvent {
     /// The configuration of some repository has been changed for the bot's Github App.
     InstallationsChanged,
     /// Periodic event that serves for checking e.g. timeouts.
     Refresh,
 }
 
+#[derive(Debug)]
+pub enum BorsEvent {
+    /// An event that happen per repository basis.
+    Repository(BorsRepositoryEvent),
+    /// An event that happen with bors globally.
+    Global(BorsGlobalEvent),
+}
+
 #[derive(Debug)]
 pub struct PullRequestComment {
     pub repository: GithubRepoName,

src/bors/handlers/mod.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use tracing::Instrument;
 
 use crate::bors::command::{BorsCommand, CommandParseError};
-use crate::bors::event::{BorsEvent, PullRequestComment};
+use crate::bors::event::{BorsRepositoryEvent, PullRequestComment};
 use crate::bors::handlers::help::command_help;
 use crate::bors::handlers::ping::command_ping;
 use crate::bors::handlers::refresh::refresh_repository;
@@ -16,6 +16,8 @@ use crate::database::DbClient;
 use crate::github::GithubRepoName;
 use crate::utils::logging::LogError;
 
+use super::event::{BorsEvent, BorsGlobalEvent};
+
 mod help;
 mod labels;
 mod ping;
@@ -28,93 +30,118 @@ pub async fn handle_bors_event<Client: RepositoryClient>(
     event: BorsEvent,
     state: Arc<dyn BorsState<Client>>,
     ctx: Arc<BorsContext>,
+) -> anyhow::Result<()> {
+    match event {
+        BorsEvent::Repository(event) => {
+            handle_bors_repository_event(event, state, ctx).await?;
+        }
+        BorsEvent::Global(event) => {
+            handle_bors_global_event(event, state, ctx).await?;
+        }
+    }
+    Ok(())
+}
+
+pub async fn handle_bors_repository_event<Client: RepositoryClient>(
+    event: BorsRepositoryEvent,
+    state: Arc<dyn BorsState<Client>>,
+    ctx: Arc<BorsContext>,
 ) -> anyhow::Result<()> {
     let db = Arc::clone(&ctx.db);
+    let Some(repo) = get_repo_state(state, event.repository()) else {
+        return Err(anyhow::anyhow!(
+            "Repository {} not found in the bot state",
+            event.repository()
+        ));
+    };
+
     match event {
-        BorsEvent::Comment(comment) => {
+        BorsRepositoryEvent::Comment(comment) => {
             // We want to ignore comments made by this bot
-            if let Some(repo) = get_repo_state(state, &comment.repository) {
-                if repo.client.is_comment_internal(&comment).await? {
-                    tracing::trace!(
-                        "Ignoring comment {comment:?} because it was authored by this bot"
-                    );
-                    return Ok(());
-                }
+            if repo.client.is_comment_internal(&comment).await? {
+                tracing::trace!("Ignoring comment {comment:?} because it was authored by this bot");
+                return Ok(());
+            }
 
-                let span = tracing::info_span!(
-                    "Comment",
-                    pr = format!("{}#{}", comment.repository, comment.pr_number),
-                    author = comment.author.username
-                );
-                let pr_number = comment.pr_number;
-                if let Err(error) = handle_comment(Arc::clone(&repo), db, ctx, comment)
-                    .instrument(span.clone())
+            let span = tracing::info_span!(
+                "Comment",
+                pr = format!("{}#{}", comment.repository, comment.pr_number),
+                author = comment.author.username
+            );
+            let pr_number = comment.pr_number;
+            if let Err(error) = handle_comment(Arc::clone(&repo), db, ctx, comment)
+                .instrument(span.clone())
+                .await
+            {
+                span.log_error(error);
+                repo.client
+                    .post_comment(
+                        pr_number,
+                        Comment::new(
+                            ":x: Encountered an error while executing command".to_string(),
+                        ),
+                    )
                     .await
-                {
-                    span.log_error(error);
-                    repo.client
-                        .post_comment(
-                            pr_number,
-                            Comment::new(
-                                ":x: Encountered an error while executing command".to_string(),
-                            ),
-                        )
-                        .await
-                        .context("Cannot send comment reacting to an error")?;
-                }
+                    .context("Cannot send comment reacting to an error")?;
             }
         }
-        BorsEvent::InstallationsChanged => {
-            let span = tracing::info_span!("Repository reload");
-            if let Err(error) = state.reload_repositories().instrument(span.clone()).await {
+
+        BorsRepositoryEvent::WorkflowStarted(payload) => {
+            let span = tracing::info_span!(
+                "Workflow started",
+                repo = payload.repository.to_string(),
+                id = payload.run_id.into_inner()
+            );
+            if let Err(error) = handle_workflow_started(db, payload)
+                .instrument(span.clone())
+                .await
+            {
                 span.log_error(error);
             }
         }
-        BorsEvent::WorkflowStarted(payload) => {
-            if let Some(_) = get_repo_state(state, &payload.repository) {
-                let span = tracing::info_span!(
-                    "Workflow started",
-                    repo = payload.repository.to_string(),
-                    id = payload.run_id.into_inner()
-                );
-                if let Err(error) = handle_workflow_started(db, payload)
-                    .instrument(span.clone())
-                    .await
-                {
-                    span.log_error(error);
-                }
+        BorsRepositoryEvent::WorkflowCompleted(payload) => {
+            let span = tracing::info_span!(
+                "Workflow completed",
+                repo = payload.repository.to_string(),
+                id = payload.run_id.into_inner()
+            );
+            if let Err(error) = handle_workflow_completed(repo, db, payload)
+                .instrument(span.clone())
+                .await
+            {
+                span.log_error(error);
             }
         }
-        BorsEvent::WorkflowCompleted(payload) => {
-            if let Some(repo) = get_repo_state(state, &payload.repository) {
-                let span = tracing::info_span!(
-                    "Workflow completed",
-                    repo = payload.repository.to_string(),
-                    id = payload.run_id.into_inner()
-                );
-                if let Err(error) = handle_workflow_completed(repo, db, payload)
-                    .instrument(span.clone())
-                    .await
-                {
-                    span.log_error(error);
-                }
+        BorsRepositoryEvent::CheckSuiteCompleted(payload) => {
+            let span = tracing::info_span!(
+                "Check suite completed",
+                repo = payload.repository.to_string(),
+            );
+            if let Err(error) = handle_check_suite_completed(repo, db, payload)
+                .instrument(span.clone())
+                .await
+            {
+                span.log_error(error);
             }
         }
-        BorsEvent::CheckSuiteCompleted(payload) => {
-            if let Some(repo) = get_repo_state(state, &payload.repository) {
-                let span = tracing::info_span!(
-                    "Check suite completed",
-                    repo = payload.repository.to_string(),
-                );
-                if let Err(error) = handle_check_suite_completed(repo, db, payload)
-                    .instrument(span.clone())
-                    .await
-                {
-                    span.log_error(error);
-                }
+    }
+    Ok(())
+}
+
+pub async fn handle_bors_global_event<Client: RepositoryClient>(
+    event: BorsGlobalEvent,
+    state: Arc<dyn BorsState<Client>>,
+    ctx: Arc<BorsContext>,
+) -> anyhow::Result<()> {
+    let db = Arc::clone(&ctx.db);
+    match event {
+        BorsGlobalEvent::InstallationsChanged => {
+            let span = tracing::info_span!("Repository reload");
+            if let Err(error) = state.reload_repositories().instrument(span.clone()).await {
+                span.log_error(error);
             }
         }
-        BorsEvent::Refresh => {
+        BorsGlobalEvent::Refresh => {
             let span = tracing::info_span!("Refresh");
             let repos = state.get_all_repos();
             futures::future::join_all(repos.into_iter().map(|repo| {