Split SqliteRepository impls into submodules

src/repository/sqlite/branch.rs

@@ -0,0 +1,37 @@
+//! `BranchRepository` implementation for SQLite.
+
+use async_trait::async_trait;
+use sqlx::SqliteConnection;
+
+use super::SqliteRepository;
+use crate::model::Branch;
+use crate::repository::{RepositoryError, branch::BranchRepository};
+
+#[async_trait]
+impl BranchRepository for SqliteRepository {
+    #[tracing::instrument(skip_all, fields(otel.kind = "client"))]
+    async fn branches_get_all(&self) -> Result<Vec<Branch>, RepositoryError> {
+        sqlx::query_as::<_, Branch>("SELECT * FROM branches")
+            .fetch_all(&self.pool)
+            .await
+            .map_err(RepositoryError::Database)
+    }
+
+    #[tracing::instrument(skip_all, fields(otel.kind = "client", id = %id))]
+    async fn branches_update_last_commit_hash(
+        &self,
+        id: i64,
+        hash: &crate::domain::CommitHash,
+        tx: &mut SqliteConnection,
+    ) -> Result<(), RepositoryError> {
+        sqlx::query!(
+            "UPDATE branches SET last_commit_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
+            hash,
+            id
+        )
+        .execute(tx)
+        .await
+        .map_err(RepositoryError::Database)?;
+        Ok(())
+    }
+}

src/repository/sqlite/mod.rs

@@ -0,0 +1,62 @@
+//! SQLite implementation of the repository.
+//!
+//! This module hosts the [`SqliteRepository`] type and its connection
+//! plumbing; each repository trait is implemented in its own submodule
+//! (`branch`, `subscription`, `trigger`).
+
+mod branch;
+mod subscription;
+mod trigger;
+
+use std::str::FromStr;
+
+use crate::config::DatabaseConfig;
+use crate::error::FatalError;
+use futures::future::BoxFuture;
+use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
+use sqlx::{SqliteConnection, SqlitePool};
+
+#[derive(Debug)]
+/// Access point of the repository using a SQLite connection pool.
+pub struct SqliteRepository {
+    /// The SQLite connection pool to the database.
+    pool: SqlitePool,
+}
+
+impl SqliteRepository {
+    /// Connects to the database described by `config`.
+    pub async fn connect(config: &DatabaseConfig) -> Result<Self, FatalError> {
+        let options = SqliteConnectOptions::from_str(config.url.as_str())?
+            .foreign_keys(true)
+            .journal_mode(SqliteJournalMode::Wal);
+
+        let pool = SqlitePoolOptions::new()
+            .acquire_timeout(config.timeout)
+            .connect_with(options)
+            .await?;
+
+        // Ensures database schema is up to date in all environments.
+        sqlx::migrate!().run(&pool).await?;
+
+        Ok(Self { pool })
+    }
+
+    /// Creates a new [`SqliteRepository`] from a [`SqlitePool`].
+    pub fn new(pool: SqlitePool) -> Self {
+        Self { pool }
+    }
+
+    /// Runs a closure within a transaction.
+    #[tracing::instrument(skip_all, fields(otel.kind = "internal"))]
+    pub async fn run_in_transaction<'a, F, T, E>(&self, f: F) -> Result<T, E>
+    where
+        F: for<'b> FnOnce(&'b mut SqliteConnection) -> BoxFuture<'b, Result<T, E>> + Send + 'a,
+        E: From<sqlx::Error> + Send + 'a,
+        T: Send + 'a,
+    {
+        let mut tx = self.pool.begin().await?;
+        let result = f(&mut tx).await?;
+        tx.commit().await?;
+        Ok(result)
+    }
+}

src/repository/sqlite.rs → src/repository/sqlite/subscription.rs

@@ -1,99 +1,12 @@
-//! SQLite implementation of the repository.
+//! `SubscriptionRepository` implementation for SQLite.
 
-use std::str::FromStr;
-
-use crate::config::DatabaseConfig;
-use crate::domain::{BranchName, EventType, RepoUrl, TargetRepo};
-use crate::error::FatalError;
-use crate::model::{
-    Branch, CreateSubscription, Subscription, SubscriptionWithBranch, TriggerQueueItem,
-    UpdateSubscription,
-};
-use crate::repository::{
-    RepositoryError,
-    branch::BranchRepository,
-    subscription::SubscriptionRepository,
-    trigger::{TriggerRepository, UpdateRetryStatus},
-};
 use async_trait::async_trait;
 use chrono::NaiveDateTime;
-use futures::future::BoxFuture;
-use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
-use sqlx::{SqliteConnection, SqlitePool};
-
-#[derive(Debug)]
-/// Access point of the repository using a SQLite connection pool.
-pub struct SqliteRepository {
-    /// The SQLite connection pool to the database.
-    pool: SqlitePool,
-}
-
-impl SqliteRepository {
-    /// Connects to the database described by `config`.
-    pub async fn connect(config: &DatabaseConfig) -> Result<Self, FatalError> {
-        let options = SqliteConnectOptions::from_str(config.url.as_str())?
-            .foreign_keys(true)
-            .journal_mode(SqliteJournalMode::Wal);
-
-        let pool = SqlitePoolOptions::new()
-            .acquire_timeout(config.timeout)
-            .connect_with(options)
-            .await?;
-
-        // Ensures database schema is up to date in all environments.
-        sqlx::migrate!().run(&pool).await?;
-
-        Ok(Self { pool })
-    }
-
-    /// Creates a new [`SqliteRepository`] from a [`SqlitePool`].
-    pub fn new(pool: SqlitePool) -> Self {
-        Self { pool }
-    }
-
-    /// Runs a closure within a transaction.
-    #[tracing::instrument(skip_all, fields(otel.kind = "internal"))]
-    pub async fn run_in_transaction<'a, F, T, E>(&self, f: F) -> Result<T, E>
-    where
-        F: for<'b> FnOnce(&'b mut SqliteConnection) -> BoxFuture<'b, Result<T, E>> + Send + 'a,
-        E: From<sqlx::Error> + Send + 'a,
-        T: Send + 'a,
-    {
-        let mut tx = self.pool.begin().await?;
-        let result = f(&mut tx).await?;
-        tx.commit().await?;
-        Ok(result)
-    }
-}
-
-#[async_trait]
-impl BranchRepository for SqliteRepository {
-    #[tracing::instrument(skip_all, fields(otel.kind = "client"))]
-    async fn branches_get_all(&self) -> Result<Vec<Branch>, RepositoryError> {
-        sqlx::query_as::<_, Branch>("SELECT * FROM branches")
-            .fetch_all(&self.pool)
-            .await
-            .map_err(RepositoryError::Database)
-    }
 
-    #[tracing::instrument(skip_all, fields(otel.kind = "client", id = %id))]
-    async fn branches_update_last_commit_hash(
-        &self,
-        id: i64,
-        hash: &crate::domain::CommitHash,
-        tx: &mut sqlx::SqliteConnection,
-    ) -> Result<(), RepositoryError> {
-        sqlx::query!(
-            "UPDATE branches SET last_commit_hash = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
-            hash,
-            id
-        )
-        .execute(tx)
-        .await
-        .map_err(RepositoryError::Database)?;
-        Ok(())
-    }
-}
+use super::SqliteRepository;
+use crate::domain::{BranchName, EventType, RepoUrl, TargetRepo};
+use crate::model::{CreateSubscription, Subscription, SubscriptionWithBranch, UpdateSubscription};
+use crate::repository::{RepositoryError, subscription::SubscriptionRepository};
 
 /// A row of the `subscriptions` table joined with its `branches` row.
 ///
@@ -351,123 +264,3 @@ impl SubscriptionRepository for SqliteRepository {
         .await
     }
 }
-
-#[async_trait]
-impl TriggerRepository for SqliteRepository {
-    #[tracing::instrument(skip_all, fields(otel.kind = "client", id = %id))]
-    async fn trigger_queue_delete(&self, id: i64) -> Result<(), RepositoryError> {
-        sqlx::query!("DELETE FROM trigger_queue WHERE id = ?", id)
-            .execute(&self.pool)
-            .await
-            .map_err(RepositoryError::Database)?;
-        Ok(())
-    }
-
-    #[tracing::instrument(skip_all, fields(otel.kind = "client"))]
-    async fn trigger_queue_process_oldest_pending(
-        &self,
-    ) -> Result<Option<TriggerQueueItem>, RepositoryError> {
-        let trigger = sqlx::query_as::<_, TriggerQueueItem>(
-            "UPDATE trigger_queue
-             SET status = 'PROCESSING', status_updated_at = CURRENT_TIMESTAMP
-             WHERE id = (
-                 SELECT id FROM trigger_queue
-                 WHERE status IN ('PENDING') AND next_retry_at <= CURRENT_TIMESTAMP
-                 ORDER BY next_retry_at ASC LIMIT 1
-             )
-             RETURNING id, branch_id, new_hash, retry_count, target_repo, event_type, gh_app_installation_id, span_context",
-        )
-        .fetch_optional(&self.pool)
-        .await
-        .map_err(RepositoryError::Database)?;
-
-        Ok(trigger)
-    }
-
-    #[tracing::instrument(
-        skip_all,
-        fields(otel.kind = "client", id = %params.id, retry_count = %params.retry_count)
-    )]
-    async fn trigger_queue_update_retry_status(
-        &self,
-        params: UpdateRetryStatus,
-    ) -> Result<(), RepositoryError> {
-        let next_retry_count = params.retry_count + 1;
-
-        if next_retry_count as u32 >= params.max_attempts {
-            sqlx::query!(
-                "UPDATE trigger_queue SET status = 'FAILED', retry_count = ? WHERE id = ?",
-                next_retry_count,
-                params.id
-            )
-            .execute(&self.pool)
-            .await
-            .map_err(RepositoryError::Database)?;
-        } else {
-            let backoff_secs = (params.backoff_base_secs * (1 << (next_retry_count - 1))) as i64;
-            sqlx::query!(
-                "UPDATE trigger_queue SET status = 'PENDING', retry_count = ?, next_retry_at = datetime('now', ? || ' seconds') WHERE id = ?",
-                next_retry_count,
-                backoff_secs,
-                params.id
-            )
-            .execute(&self.pool)
-            .await
-            .map_err(RepositoryError::Database)?;
-        }
-        Ok(())
-    }
-
-    #[tracing::instrument(
-        skip_all,
-        fields(otel.kind = "client", threshold_seconds = %threshold_seconds)
-    )]
-    async fn trigger_queue_recover_stuck_tasks(
-        &self,
-        threshold_seconds: u64,
-    ) -> Result<(), RepositoryError> {
-        let threshold_str = format!("-{} seconds", threshold_seconds);
-
-        sqlx::query!(
-            "UPDATE trigger_queue
-             SET status = 'PENDING', status_updated_at = CURRENT_TIMESTAMP
-             WHERE status = 'PROCESSING'
-               AND status_updated_at < DATETIME('now', ?)",
-            threshold_str
-        )
-        .execute(&self.pool)
-        .await
-        .map_err(RepositoryError::Database)?;
-        Ok(())
-    }
-
-    #[tracing::instrument(
-        skip_all,
-        fields(otel.kind = "client", branch_id = %params.branch_id)
-    )]
-    async fn trigger_queue_upsert(
-        &self,
-        params: crate::repository::trigger::TriggerQueueUpsertParams<'_>,
-        executor: &mut sqlx::SqliteConnection,
-    ) -> Result<(), RepositoryError> {
-        let branch_id = params.branch_id;
-        let new_hash = params.new_hash;
-        let span_context = params.span_context;
-        sqlx::query!(
-            "INSERT INTO trigger_queue (branch_id, new_hash, target_repo, event_type, gh_app_installation_id, span_context)
-             SELECT ?, ?, s.target_repo, s.event_type, s.gh_app_installation_id, ?
-             FROM subscriptions s
-             WHERE s.branch_id = ?
-             ON CONFLICT(target_repo, event_type) WHERE status = 'PENDING'
-             DO UPDATE SET branch_id = excluded.branch_id, new_hash = excluded.new_hash, span_context = excluded.span_context, status_updated_at = CURRENT_TIMESTAMP",
-            branch_id,
-            new_hash,
-            span_context,
-            branch_id
-        )
-        .execute(executor)
-        .await
-        .map_err(RepositoryError::Database)?;
-        Ok(())
-    }
-}