commit_bridge/polling/
mod.rs

1//! Asynchronous task to periodically check for updated remote branches.
2
3use async_trait::async_trait;
4use futures::{StreamExt, future::BoxFuture, stream};
5use tracing::{info, warn};
6
7use crate::{
8    context::SharedContext,
9    engine::AsyncEngine,
10    error::CommitHashError,
11    polling::{
12        branch::BranchInfo,
13        error::{PollingError, handle_polling_error},
14    },
15    repository::{RepositoryError, branch::BranchRepository, trigger::TriggerRepository},
16};
17
18mod branch;
19mod error;
20pub mod git;
21
22/// Runs an asynchronous task
23/// that periodically polls git branches in remote repositories.
24pub struct PollingEngine {
25    /// Shared data for all async engines.
26    pub ctx: SharedContext,
27}
28
29#[async_trait]
30impl AsyncEngine for PollingEngine {
31    async fn run(&self) {
32        polling_loop(self.ctx.clone()).await;
33    }
34}
35
36/// Controls whether to shut down the polling engine or run a polling cycle.
37async fn polling_loop(ctx: SharedContext) {
38    loop {
39        tokio::select! {
40            res = poll_branches(&ctx) => {followup_poll(res, &ctx).await}
41            _ = ctx.token.cancelled() => break,
42        }
43    }
44    info!("Gracefully shutting down polling engine");
45}
46
47/// Polls branches for updates,
48/// updates them in the `branches` table,
49/// and queues the updates for the [`TriggerEngine`].
50///
51/// <!-- LINKS -->
52/// [`TriggerEngine`]: crate::trigger::TriggerEngine
53async fn poll_branches(ctx: &SharedContext) -> Result<(), PollingError> {
54    let updated_branches = gather_updated_branches(ctx).await?;
55    if updated_branches.is_empty() {
56        return Ok(());
57    }
58
59    let repository = ctx.repository.clone();
60    let shared_branches = std::sync::Arc::new(updated_branches);
61    ctx.repository
62        .run_in_transaction(|tx| {
63            execute_branch_updates(repository.clone(), shared_branches.clone(), tx)
64        })
65        .await?;
66
67    Ok(())
68}
69
70/// Gathers stored branches that need to be updated.
71async fn gather_updated_branches(ctx: &SharedContext) -> Result<Vec<BranchInfo>, sqlx::Error> {
72    let branches = BranchRepository::get_all(ctx.repository.as_ref())
73        .await
74        .map_err(|e| match e {
75            crate::repository::RepositoryError::Database(e) => e,
76            _ => sqlx::Error::RowNotFound,
77        })?;
78
79    let branch_results = stream::iter(branches)
80        .map(|b| BranchInfo::new(b, ctx.git_fetcher.as_ref()))
81        .buffer_unordered(ctx.config.database.polling_db_buffer_size)
82        .collect::<Vec<Result<BranchInfo, CommitHashError>>>()
83        .await;
84
85    let errs = branch_results.iter().filter_map(|res| res.as_ref().err());
86    for e in errs {
87        warn!("{e}");
88    }
89
90    let updated_branches = branch_results
91        .into_iter()
92        .filter_map(|res| res.ok())
93        .filter(BranchInfo::has_updated)
94        .collect();
95    Ok(updated_branches)
96}
97
98/// Helper function to pin the branch update process.
99fn execute_branch_updates<'a>(
100    repository: std::sync::Arc<crate::repository::SqliteRepository>,
101    shared_branches: std::sync::Arc<Vec<branch::BranchInfo>>,
102    tx: &'a mut sqlx::SqliteConnection,
103) -> BoxFuture<'a, Result<(), RepositoryError>> {
104    Box::pin(process_branches(repository, shared_branches, tx))
105}
106
107/// Processes branch updates within a transaction.
108async fn process_branches(
109    repo: std::sync::Arc<crate::repository::SqliteRepository>,
110    shared_branches: std::sync::Arc<Vec<branch::BranchInfo>>,
111    tx: &mut sqlx::SqliteConnection,
112) -> Result<(), RepositoryError> {
113    let branches = std::sync::Arc::clone(&shared_branches);
114    for branch_info in branches.iter() {
115        repo.update_last_commit_hash_in_tx(branch_info.branch.id, &branch_info.latest_hash, tx)
116            .await?;
117
118        info!(
119            "New commit detected for branch {}. Hash: {}",
120            branch_info.branch.name, branch_info.latest_hash
121        );
122
123        repo.queue_triggers_for_branch(branch_info.branch.id, &branch_info.latest_hash, tx)
124            .await?;
125    }
126    Ok(())
127}
128
129/// Handles polling results and puts the task to sleep.
130async fn followup_poll(res: Result<(), PollingError>, ctx: &SharedContext) {
131    match res {
132        Ok(_) => tokio::select! {
133            _ = tokio::time::sleep(ctx.config.engine.polling_sleep) => {}
134            _ = ctx.token.cancelled() => {}
135        },
136        Err(e) => handle_polling_error(e, ctx).await,
137    }
138}
139
140#[cfg(test)]
141#[allow(
142    clippy::panic,
143    clippy::expect_used,
144    clippy::todo,
145    clippy::unimplemented,
146    clippy::indexing_slicing
147)]
148mod tests {
149    use crate::context::SharedContext;
150    use crate::domain::CommitHash;
151    use crate::polling::poll_branches;
152    use crate::test_utils::MockGitFetcher;
153    use std::sync::Arc;
154    use tokio_util::sync::CancellationToken;
155
156    #[tokio::test]
157    async fn test_poll_branches_updates_db_and_queues_trigger() {
158        let pool = crate::test_utils::create_test_db().await;
159
160        // Insert a branch
161        let hash = "a".repeat(40);
162        sqlx::query!(
163            "INSERT INTO branches (repo_url, name, last_commit_hash) VALUES (?, ?, ?)",
164            "https://github.com/owner/repo",
165            "main",
166            hash
167        )
168        .execute(&pool)
169        .await
170        .unwrap();
171        // Insert a subscription
172        sqlx::query!("INSERT INTO subscriptions (branch_id, target_repo, event_type, gh_app_installation_id) VALUES (?, ?, ?, ?)",
173            1,
174            "org/target",
175            "dispatch",
176            1
177        )
178            .execute(&pool)
179            .await
180            .unwrap();
181
182        let mock_fetcher = Arc::new(MockGitFetcher {
183            hash: CommitHash::new("b".repeat(40)).unwrap(),
184        });
185
186        let ctx = SharedContext {
187            config: crate::test_utils::create_test_config(),
188            repository: std::sync::Arc::new(crate::repository::SqliteRepository::new(pool.clone())),
189            db_pool: pool.clone(),
190            git_fetcher: mock_fetcher,
191            token: CancellationToken::new(),
192        };
193
194        poll_branches(&ctx).await.unwrap();
195
196        // Verify DB update
197        let branch = sqlx::query!("SELECT last_commit_hash FROM branches WHERE name = 'main'")
198            .fetch_one(&pool)
199            .await
200            .unwrap();
201        assert_eq!(branch.last_commit_hash, Some("b".repeat(40)));
202
203        // Verify trigger queued
204        let queued_event = sqlx::query!("SELECT branch_id, new_hash FROM trigger_queue")
205            .fetch_one(&pool)
206            .await
207            .unwrap();
208
209        assert_eq!(queued_event.branch_id, Some(1));
210        assert_eq!(queued_event.new_hash, Some("b".repeat(40)));
211    }
212
213    #[tokio::test]
214    async fn test_coalescing_of_trigger_events() {
215        let pool = crate::test_utils::create_test_db().await;
216
217        // Insert a branch
218        let hash = "a".repeat(40);
219        sqlx::query!(
220            "INSERT INTO branches (repo_url, name, last_commit_hash) VALUES (?, ?, ?)",
221            "https://github.com/owner/repo",
222            "main",
223            hash
224        )
225        .execute(&pool)
226        .await
227        .unwrap();
228        // Insert a subscription
229        sqlx::query!("INSERT INTO subscriptions (branch_id, target_repo, event_type, gh_app_installation_id) VALUES (?, ?, ?, ?)",
230            1,
231            "org/target",
232            "dispatch",
233            1
234        )
235            .execute(&pool)
236            .await
237            .unwrap();
238
239        let ctx = SharedContext {
240            config: crate::test_utils::create_test_config(),
241            repository: std::sync::Arc::new(crate::repository::SqliteRepository::new(pool.clone())),
242            db_pool: pool.clone(),
243            git_fetcher: Arc::new(crate::test_utils::MockGitFetcher {
244                hash: CommitHash::new("b".repeat(40)).unwrap(),
245            }),
246            token: CancellationToken::new(),
247        };
248
249        // First update
250        poll_branches(&ctx).await.unwrap();
251
252        // Second update (coalescing)
253        // Manually update the mock fetcher to a new hash
254        let mock_fetcher = Arc::new(crate::test_utils::MockGitFetcher {
255            hash: CommitHash::new("c".repeat(40)).unwrap(),
256        });
257        let ctx = SharedContext {
258            config: ctx.config,
259            repository: std::sync::Arc::new(crate::repository::SqliteRepository::new(pool.clone())),
260            db_pool: pool.clone(),
261            git_fetcher: mock_fetcher,
262            token: ctx.token,
263        };
264        poll_branches(&ctx).await.unwrap();
265
266        // Verify only one entry in queue
267        let queued_events = sqlx::query!("SELECT branch_id, new_hash FROM trigger_queue")
268            .fetch_all(&pool)
269            .await
270            .unwrap();
271
272        assert_eq!(queued_events.len(), 1);
273        assert_eq!(queued_events[0].branch_id, Some(1));
274        assert_eq!(queued_events[0].new_hash, Some("c".repeat(40)));
275    }
276}
commit_bridge/polling/mod.rs

commit_bridge/polling/
mod.rs
