From 0472a3ab4b491f712fd5b2217c1e7282d939fd10 Mon Sep 17 00:00:00 2001 From: Nilirad Date: Tue, 8 Sep 2026 13:55:35 +0200 Subject: [PATCH] Split `SqliteRepository` impls into submodules --- src/repository/sqlite/branch.rs | 37 +++ src/repository/sqlite/mod.rs | 62 +++++ .../{sqlite.rs => sqlite/subscription.rs} | 217 +----------------- src/repository/sqlite/trigger.rs | 130 +++++++++++ 4 files changed, 234 insertions(+), 212 deletions(-) create mode 100644 src/repository/sqlite/branch.rs create mode 100644 src/repository/sqlite/mod.rs rename src/repository/{sqlite.rs => sqlite/subscription.rs} (56%) create mode 100644 src/repository/sqlite/trigger.rs diff --git a/src/repository/sqlite/branch.rs b/src/repository/sqlite/branch.rs new file mode 100644 index 0000000..05e9379 --- /dev/null +++ b/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, 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(()) + } +} diff --git a/src/repository/sqlite/mod.rs b/src/repository/sqlite/mod.rs new file mode 100644 index 0000000..a137de7 --- /dev/null +++ b/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 { + 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 + where + F: for<'b> FnOnce(&'b mut SqliteConnection) -> BoxFuture<'b, Result> + Send + 'a, + E: From + Send + 'a, + T: Send + 'a, + { + let mut tx = self.pool.begin().await?; + let result = f(&mut tx).await?; + tx.commit().await?; + Ok(result) + } +} diff --git a/src/repository/sqlite.rs b/src/repository/sqlite/subscription.rs similarity index 56% rename from src/repository/sqlite.rs rename to src/repository/sqlite/subscription.rs index 2096d0b..a85a74c 100644 --- a/src/repository/sqlite.rs +++ b/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 { - 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 - where - F: for<'b> FnOnce(&'b mut SqliteConnection) -> BoxFuture<'b, Result> + Send + 'a, - E: From + 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, 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, 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(()) - } -} diff --git a/src/repository/sqlite/trigger.rs b/src/repository/sqlite/trigger.rs new file mode 100644 index 0000000..b7c8844 --- /dev/null +++ b/src/repository/sqlite/trigger.rs @@ -0,0 +1,130 @@ +//! `TriggerRepository` implementation for SQLite. + +use async_trait::async_trait; + +use super::SqliteRepository; +use crate::model::TriggerQueueItem; +use crate::repository::{ + RepositoryError, + trigger::{TriggerRepository, UpdateRetryStatus}, +}; + +#[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, 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(()) + } +}