NotAShelf/circus
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

treewide: address all clippy lints

Browse source 
Signed-off-by: NotAShelf <raf@notashelf.dev>
Change-Id: I5cf55cc4cb558c3f9f764c71224e87176a6a6964
This commit is contained in:
raf 2026-02-27 21:50:35 +03:00
commit a127f3f62c
Signed by: NotAShelf
GPG key ID: 29D95B64378DB4BF
63 changed files with 1790 additions and 1089 deletions
Download patch file Download diff file Expand all files Collapse all files

30
crates/common/src/repo/api_keys.rs
View file

@ -6,6 +6,11 @@ use crate::{
models::ApiKey,
};
/// Create a new API key.
///
/// # Errors
///
/// Returns error if database insert fails or key already exists.
pub async fn create(
pool: &PgPool,
name: &str,
@ -31,6 +36,11 @@ pub async fn create(
})
}
/// Insert or update an API key by hash.
///
/// # Errors
///
/// Returns error if database operation fails.
pub async fn upsert(
pool: &PgPool,
name: &str,
@ -50,6 +60,11 @@ pub async fn upsert(
.map_err(CiError::Database)
}
/// Find an API key by its hash.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_by_hash(
pool: &PgPool,
key_hash: &str,
@ -61,6 +76,11 @@ pub async fn get_by_hash(
.map_err(CiError::Database)
}
/// List all API keys.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list(pool: &PgPool) -> Result<Vec<ApiKey>> {
sqlx::query_as::<_, ApiKey>("SELECT * FROM api_keys ORDER BY created_at DESC")
.fetch_all(pool)
@ -68,6 +88,11 @@ pub async fn list(pool: &PgPool) -> Result<Vec<ApiKey>> {
.map_err(CiError::Database)
}
/// Delete an API key by ID.
///
/// # Errors
///
/// Returns error if database delete fails or key not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM api_keys WHERE id = $1")
.bind(id)
@ -79,6 +104,11 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
Ok(())
}
/// Update the `last_used_at` timestamp for an API key.
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn touch_last_used(pool: &PgPool, id: Uuid) -> Result<()> {
sqlx::query("UPDATE api_keys SET last_used_at = NOW() WHERE id = $1")
.bind(id)

18
crates/common/src/repo/build_dependencies.rs
View file

@ -6,6 +6,11 @@ use crate::{
models::BuildDependency,
};
/// Create a build dependency relationship.
///
/// # Errors
///
/// Returns error if database insert fails or dependency already exists.
pub async fn create(
pool: &PgPool,
build_id: Uuid,
@ -31,6 +36,11 @@ pub async fn create(
})
}
/// List all dependencies for a build.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_build(
pool: &PgPool,
build_id: Uuid,
@ -46,6 +56,10 @@ pub async fn list_for_build(
/// Batch check if all dependency builds are completed for multiple builds at
/// once. Returns a map from `build_id` to whether all deps are completed.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn check_deps_for_builds(
pool: &PgPool,
build_ids: &[Uuid],
@ -77,6 +91,10 @@ pub async fn check_deps_for_builds(
}
/// Check if all dependency builds for a given build are completed.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn all_deps_completed(pool: &PgPool, build_id: Uuid) -> Result<bool> {
let row: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM build_dependencies bd JOIN builds b ON \

57
crates/common/src/repo/build_metrics.rs
View file

@ -7,6 +7,8 @@ use crate::{
models::BuildMetric,
};
type PercentileRow = (DateTime<Utc>, Option<f64>, Option<f64>, Option<f64>);
/// Time-series data point for metrics visualization.
#[derive(Debug, Clone)]
pub struct TimeseriesPoint {
@ -32,6 +34,11 @@ pub struct DurationPercentiles {
pub p99: Option<f64>,
}
/// Insert or update a build metric.
///
/// # Errors
///
/// Returns error if database operation fails.
pub async fn upsert(
pool: &PgPool,
build_id: Uuid,
@ -54,6 +61,11 @@ pub async fn upsert(
.map_err(CiError::Database)
}
/// Calculate build failure rate over a time window.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn calculate_failure_rate(
pool: &PgPool,
project_id: Option<Uuid>,
@ -87,6 +99,10 @@ pub async fn calculate_failure_rate(
/// Get build success/failure counts over time.
/// Buckets builds by time interval for charting.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_build_stats_timeseries(
pool: &PgPool,
project_id: Option<Uuid>,
@ -136,6 +152,10 @@ pub async fn get_build_stats_timeseries(
}
/// Get build duration percentiles over time.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_duration_percentiles_timeseries(
pool: &PgPool,
project_id: Option<Uuid>,
@ -143,18 +163,17 @@ pub async fn get_duration_percentiles_timeseries(
hours: i32,
bucket_minutes: i32,
) -> Result<Vec<DurationPercentiles>> {
let rows: Vec<(DateTime<Utc>, Option<f64>, Option<f64>, Option<f64>)> =
sqlx::query_as(
"SELECT
let rows: Vec<PercentileRow> = sqlx::query_as(
"SELECT
date_trunc('minute', b.completed_at) +
(EXTRACT(MINUTE FROM b.completed_at)::int / $4) * INTERVAL '1 minute' \
* $4 AS bucket_time,
* $4 AS bucket_time,
PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY EXTRACT(EPOCH FROM \
(b.completed_at - b.started_at))) AS p50,
(b.completed_at - b.started_at))) AS p50,
PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY EXTRACT(EPOCH FROM \
(b.completed_at - b.started_at))) AS p95,
(b.completed_at - b.started_at))) AS p95,
PERCENTILE_CONT(0.99) WITHIN GROUP (ORDER BY EXTRACT(EPOCH FROM \
(b.completed_at - b.started_at))) AS p99
(b.completed_at - b.started_at))) AS p99
FROM builds b
JOIN evaluations e ON b.evaluation_id = e.id
JOIN jobsets j ON e.jobset_id = j.id
@ -165,14 +184,14 @@ pub async fn get_duration_percentiles_timeseries(
AND ($3::uuid IS NULL OR j.id = $3)
GROUP BY bucket_time
ORDER BY bucket_time ASC",
)
.bind(hours)
.bind(project_id)
.bind(jobset_id)
.bind(bucket_minutes)
.fetch_all(pool)
.await
.map_err(CiError::Database)?;
)
.bind(hours)
.bind(project_id)
.bind(jobset_id)
.bind(bucket_minutes)
.fetch_all(pool)
.await
.map_err(CiError::Database)?;
Ok(
rows
@ -190,6 +209,10 @@ pub async fn get_duration_percentiles_timeseries(
}
/// Get queue depth over time.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_queue_depth_timeseries(
pool: &PgPool,
hours: i32,
@ -228,6 +251,10 @@ pub async fn get_queue_depth_timeseries(
}
/// Get per-system build distribution.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_system_distribution(
pool: &PgPool,
project_id: Option<Uuid>,

15
crates/common/src/repo/build_products.rs
View file

@ -6,6 +6,11 @@ use crate::{
models::{BuildProduct, CreateBuildProduct},
};
/// Create a build product record.
///
/// # Errors
///
/// Returns error if database insert fails.
pub async fn create(
pool: &PgPool,
input: CreateBuildProduct,
@ -27,6 +32,11 @@ pub async fn create(
.map_err(CiError::Database)
}
/// Get a build product by ID.
///
/// # Errors
///
/// Returns error if database query fails or product not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<BuildProduct> {
sqlx::query_as::<_, BuildProduct>(
"SELECT * FROM build_products WHERE id = $1",
@ -37,6 +47,11 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<BuildProduct> {
.ok_or_else(|| CiError::NotFound(format!("Build product {id} not found")))
}
/// List all build products for a build.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_build(
pool: &PgPool,
build_id: Uuid,

15
crates/common/src/repo/build_steps.rs
View file

@ -6,6 +6,11 @@ use crate::{
models::{BuildStep, CreateBuildStep},
};
/// Create a build step record.
///
/// # Errors
///
/// Returns error if database insert fails or step already exists.
pub async fn create(
pool: &PgPool,
input: CreateBuildStep,
@ -32,6 +37,11 @@ pub async fn create(
})
}
/// Mark a build step as completed.
///
/// # Errors
///
/// Returns error if database update fails or step not found.
pub async fn complete(
pool: &PgPool,
id: Uuid,
@ -52,6 +62,11 @@ pub async fn complete(
.ok_or_else(|| CiError::NotFound(format!("Build step {id} not found")))
}
/// List all build steps for a build.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_build(
pool: &PgPool,
build_id: Uuid,

102
crates/common/src/repo/builds.rs
View file

@ -6,6 +6,11 @@ use crate::{
models::{Build, BuildStats, BuildStatus, CreateBuild},
};
/// Create a new build record in pending state.
///
/// # Errors
///
/// Returns error if database insert fails or job already exists.
pub async fn create(pool: &PgPool, input: CreateBuild) -> Result<Build> {
let is_aggregate = input.is_aggregate.unwrap_or(false);
sqlx::query_as::<_, Build>(
@ -35,6 +40,11 @@ pub async fn create(pool: &PgPool, input: CreateBuild) -> Result<Build> {
})
}
/// Find a succeeded build by derivation path (for build result caching).
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_completed_by_drv_path(
pool: &PgPool,
drv_path: &str,
@ -48,6 +58,11 @@ pub async fn get_completed_by_drv_path(
.map_err(CiError::Database)
}
/// Get a build by ID.
///
/// # Errors
///
/// Returns error if database query fails or build not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<Build> {
sqlx::query_as::<_, Build>("SELECT * FROM builds WHERE id = $1")
.bind(id)
@ -56,6 +71,11 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<Build> {
.ok_or_else(|| CiError::NotFound(format!("Build {id} not found")))
}
/// List all builds for a given evaluation.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_evaluation(
pool: &PgPool,
evaluation_id: Uuid,
@ -69,6 +89,12 @@ pub async fn list_for_evaluation(
.map_err(CiError::Database)
}
/// List pending builds, prioritizing non-aggregate jobs.
/// Returns up to `limit * worker_count` builds.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_pending(
pool: &PgPool,
limit: i64,
@ -99,6 +125,10 @@ pub async fn list_pending(
/// Atomically claim a pending build by setting it to running.
/// Returns `None` if the build was already claimed by another worker.
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn start(pool: &PgPool, id: Uuid) -> Result<Option<Build>> {
sqlx::query_as::<_, Build>(
"UPDATE builds SET status = 'running', started_at = NOW() WHERE id = $1 \
@ -110,6 +140,11 @@ pub async fn start(pool: &PgPool, id: Uuid) -> Result<Option<Build>> {
.map_err(CiError::Database)
}
/// Mark a build as completed with final status and outputs.
///
/// # Errors
///
/// Returns error if database update fails or build not found.
pub async fn complete(
pool: &PgPool,
id: Uuid,
@ -132,6 +167,11 @@ pub async fn complete(
.ok_or_else(|| CiError::NotFound(format!("Build {id} not found")))
}
/// List recent builds ordered by creation time.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_recent(pool: &PgPool, limit: i64) -> Result<Vec<Build>> {
sqlx::query_as::<_, Build>(
"SELECT * FROM builds ORDER BY created_at DESC LIMIT $1",
@ -142,6 +182,11 @@ pub async fn list_recent(pool: &PgPool, limit: i64) -> Result<Vec<Build>> {
.map_err(CiError::Database)
}
/// List all builds for a project.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_project(
pool: &PgPool,
project_id: Uuid,
@ -157,6 +202,11 @@ pub async fn list_for_project(
.map_err(CiError::Database)
}
/// Get aggregate build statistics.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_stats(pool: &PgPool) -> Result<BuildStats> {
match sqlx::query_as::<_, BuildStats>("SELECT * FROM build_stats")
.fetch_optional(pool)
@ -178,6 +228,10 @@ pub async fn get_stats(pool: &PgPool) -> Result<BuildStats> {
/// Reset builds that were left in 'running' state (orphaned by a crashed
/// runner). Limited to 50 builds per call to prevent thundering herd.
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn reset_orphaned(
pool: &PgPool,
older_than_secs: i64,
@ -197,6 +251,10 @@ pub async fn reset_orphaned(
/// List builds with optional `evaluation_id`, status, system, and `job_name`
/// filters, with pagination.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_filtered(
pool: &PgPool,
evaluation_id: Option<Uuid>,
@ -223,6 +281,11 @@ pub async fn list_filtered(
.map_err(CiError::Database)
}
/// Count builds matching filter criteria.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn count_filtered(
pool: &PgPool,
evaluation_id: Option<Uuid>,
@ -247,6 +310,10 @@ pub async fn count_filtered(
/// Return the subset of the given build IDs whose status is 'cancelled'.
/// Used by the cancel-checker loop to detect builds cancelled while running.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_cancelled_among(
pool: &PgPool,
build_ids: &[Uuid],
@ -265,6 +332,11 @@ pub async fn get_cancelled_among(
Ok(rows.into_iter().map(|(id,)| id).collect())
}
/// Cancel a build.
///
/// # Errors
///
/// Returns error if database update fails or build not in cancellable state.
pub async fn cancel(pool: &PgPool, id: Uuid) -> Result<Build> {
sqlx::query_as::<_, Build>(
"UPDATE builds SET status = 'cancelled', completed_at = NOW() WHERE id = \
@ -281,6 +353,10 @@ pub async fn cancel(pool: &PgPool, id: Uuid) -> Result<Build> {
}
/// Cancel a build and all its transitive dependents.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn cancel_cascade(pool: &PgPool, id: Uuid) -> Result<Vec<Build>> {
let mut cancelled = Vec::new();
@ -312,7 +388,11 @@ pub async fn cancel_cascade(pool: &PgPool, id: Uuid) -> Result<Vec<Build>> {
}
/// Restart a build by resetting it to pending state.
/// Only works for failed, succeeded, cancelled, or cached_failure builds.
/// Only works for failed, succeeded, cancelled, or `cached_failure` builds.
///
/// # Errors
///
/// Returns error if database update fails or build not in restartable state.
pub async fn restart(pool: &PgPool, id: Uuid) -> Result<Build> {
let build = sqlx::query_as::<_, Build>(
"UPDATE builds SET status = 'pending', started_at = NULL, completed_at = \
@ -339,6 +419,10 @@ pub async fn restart(pool: &PgPool, id: Uuid) -> Result<Build> {
}
/// Mark a build's outputs as signed.
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn mark_signed(pool: &PgPool, id: Uuid) -> Result<()> {
sqlx::query("UPDATE builds SET signed = true WHERE id = $1")
.bind(id)
@ -350,6 +434,10 @@ pub async fn mark_signed(pool: &PgPool, id: Uuid) -> Result<()> {
/// Batch-fetch completed builds by derivation paths.
/// Returns a map from `drv_path` to Build for deduplication.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_completed_by_drv_paths(
pool: &PgPool,
drv_paths: &[String],
@ -375,6 +463,10 @@ pub async fn get_completed_by_drv_paths(
}
/// Return the set of build IDs that have `keep = true` (GC-pinned).
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_pinned_ids(
pool: &PgPool,
) -> Result<std::collections::HashSet<Uuid>> {
@ -387,6 +479,10 @@ pub async fn list_pinned_ids(
}
/// Set the `keep` (GC pin) flag on a build.
///
/// # Errors
///
/// Returns error if database update fails or build not found.
pub async fn set_keep(pool: &PgPool, id: Uuid, keep: bool) -> Result<Build> {
sqlx::query_as::<_, Build>(
"UPDATE builds SET keep = $1 WHERE id = $2 RETURNING *",
@ -399,6 +495,10 @@ pub async fn set_keep(pool: &PgPool, id: Uuid, keep: bool) -> Result<Build> {
}
/// Set the `builder_id` for a build.
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn set_builder(
pool: &PgPool,
id: Uuid,

43
crates/common/src/repo/channels.rs
View file

@ -7,6 +7,11 @@ use crate::{
models::{Channel, CreateChannel},
};
/// Create a release channel.
///
/// # Errors
///
/// Returns error if database insert fails or channel already exists.
pub async fn create(pool: &PgPool, input: CreateChannel) -> Result<Channel> {
sqlx::query_as::<_, Channel>(
"INSERT INTO channels (project_id, name, jobset_id) VALUES ($1, $2, $3) \
@ -30,6 +35,11 @@ pub async fn create(pool: &PgPool, input: CreateChannel) -> Result<Channel> {
})
}
/// Get a channel by ID.
///
/// # Errors
///
/// Returns error if database query fails or channel not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<Channel> {
sqlx::query_as::<_, Channel>("SELECT * FROM channels WHERE id = $1")
.bind(id)
@ -38,6 +48,11 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<Channel> {
.ok_or_else(|| CiError::NotFound(format!("Channel {id} not found")))
}
/// List all channels for a project.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_project(
pool: &PgPool,
project_id: Uuid,
@ -51,6 +66,11 @@ pub async fn list_for_project(
.map_err(CiError::Database)
}
/// List all channels.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_all(pool: &PgPool) -> Result<Vec<Channel>> {
sqlx::query_as::<_, Channel>("SELECT * FROM channels ORDER BY name")
.fetch_all(pool)
@ -59,6 +79,10 @@ pub async fn list_all(pool: &PgPool) -> Result<Vec<Channel>> {
}
/// Promote an evaluation to a channel (set it as the current evaluation).
///
/// # Errors
///
/// Returns error if database update fails or channel not found.
pub async fn promote(
pool: &PgPool,
channel_id: Uuid,
@ -75,6 +99,11 @@ pub async fn promote(
.ok_or_else(|| CiError::NotFound(format!("Channel {channel_id} not found")))
}
/// Delete a channel.
///
/// # Errors
///
/// Returns error if database delete fails or channel not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM channels WHERE id = $1")
.bind(id)
@ -88,6 +117,10 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
}
/// Upsert a channel (insert or update on conflict).
///
/// # Errors
///
/// Returns error if database operation fails.
pub async fn upsert(
pool: &PgPool,
project_id: Uuid,
@ -109,6 +142,10 @@ pub async fn upsert(
/// Sync channels from declarative config.
/// Deletes channels not in the declarative list and upserts those that are.
///
/// # Errors
///
/// Returns error if database operations fail.
pub async fn sync_for_project(
pool: &PgPool,
project_id: Uuid,
@ -146,6 +183,10 @@ pub async fn sync_for_project(
/// Find the channel for a jobset and auto-promote if all builds in the
/// evaluation succeeded.
///
/// # Errors
///
/// Returns error if database operations fail.
pub async fn auto_promote_if_complete(
pool: &PgPool,
jobset_id: Uuid,
@ -166,7 +207,7 @@ pub async fn auto_promote_if_complete(
return Ok(());
}
// All builds completed — promote to any channels tracking this jobset
// All builds completed, promote to any channels tracking this jobset
let channels =
sqlx::query_as::<_, Channel>("SELECT * FROM channels WHERE jobset_id = $1")
.bind(jobset_id)

53
crates/common/src/repo/evaluations.rs
View file

@ -6,6 +6,11 @@ use crate::{
models::{CreateEvaluation, Evaluation, EvaluationStatus},
};
/// Create a new evaluation in pending state.
///
/// # Errors
///
/// Returns error if database insert fails or evaluation already exists.
pub async fn create(
pool: &PgPool,
input: CreateEvaluation,
@ -36,6 +41,11 @@ pub async fn create(
})
}
/// Get an evaluation by ID.
///
/// # Errors
///
/// Returns error if database query fails or evaluation not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<Evaluation> {
sqlx::query_as::<_, Evaluation>("SELECT * FROM evaluations WHERE id = $1")
.bind(id)
@ -44,6 +54,11 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<Evaluation> {
.ok_or_else(|| CiError::NotFound(format!("Evaluation {id} not found")))
}
/// List all evaluations for a jobset.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_jobset(
pool: &PgPool,
jobset_id: Uuid,
@ -60,6 +75,10 @@ pub async fn list_for_jobset(
/// List evaluations with optional `jobset_id` and status filters, with
/// pagination.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_filtered(
pool: &PgPool,
jobset_id: Option<Uuid>,
@ -81,6 +100,11 @@ pub async fn list_filtered(
.map_err(CiError::Database)
}
/// Count evaluations matching filter criteria.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn count_filtered(
pool: &PgPool,
jobset_id: Option<Uuid>,
@ -98,6 +122,11 @@ pub async fn count_filtered(
Ok(row.0)
}
/// Update evaluation status and optional error message.
///
/// # Errors
///
/// Returns error if database update fails or evaluation not found.
pub async fn update_status(
pool: &PgPool,
id: Uuid,
@ -116,6 +145,11 @@ pub async fn update_status(
.ok_or_else(|| CiError::NotFound(format!("Evaluation {id} not found")))
}
/// Get the latest evaluation for a jobset.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_latest(
pool: &PgPool,
jobset_id: Uuid,
@ -131,6 +165,10 @@ pub async fn get_latest(
}
/// Set the inputs hash for an evaluation (used for eval caching).
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn set_inputs_hash(
pool: &PgPool,
id: Uuid,
@ -147,6 +185,10 @@ pub async fn set_inputs_hash(
/// Check if an evaluation with the same `inputs_hash` already exists for this
/// jobset.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_by_inputs_hash(
pool: &PgPool,
jobset_id: Uuid,
@ -163,6 +205,11 @@ pub async fn get_by_inputs_hash(
.map_err(CiError::Database)
}
/// Count total evaluations.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn count(pool: &PgPool) -> Result<i64> {
let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM evaluations")
.fetch_one(pool)
@ -171,7 +218,11 @@ pub async fn count(pool: &PgPool) -> Result<i64> {
Ok(row.0)
}
/// Get an evaluation by jobset_id and commit_hash.
/// Get an evaluation by `jobset_id` and `commit_hash`.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_by_jobset_and_commit(
pool: &PgPool,
jobset_id: Uuid,

20
crates/common/src/repo/failed_paths_cache.rs
View file

@ -6,6 +6,11 @@ use crate::{
models::BuildStatus,
};
/// Check if a derivation path is in the failed paths cache.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn is_cached_failure(pool: &PgPool, drv_path: &str) -> Result<bool> {
let row: Option<(bool,)> =
sqlx::query_as("SELECT true FROM failed_paths_cache WHERE drv_path = $1")
@ -17,6 +22,11 @@ pub async fn is_cached_failure(pool: &PgPool, drv_path: &str) -> Result<bool> {
Ok(row.is_some())
}
/// Insert a failed derivation path into the cache.
///
/// # Errors
///
/// Returns error if database insert fails.
pub async fn insert(
pool: &PgPool,
drv_path: &str,
@ -40,6 +50,11 @@ pub async fn insert(
Ok(())
}
/// Remove a derivation path from the failed paths cache.
///
/// # Errors
///
/// Returns error if database delete fails.
pub async fn invalidate(pool: &PgPool, drv_path: &str) -> Result<()> {
sqlx::query("DELETE FROM failed_paths_cache WHERE drv_path = $1")
.bind(drv_path)
@ -50,6 +65,11 @@ pub async fn invalidate(pool: &PgPool, drv_path: &str) -> Result<()> {
Ok(())
}
/// Remove expired entries from the failed paths cache.
///
/// # Errors
///
/// Returns error if database delete fails.
pub async fn cleanup_expired(pool: &PgPool, ttl_seconds: u64) -> Result<u64> {
let result = sqlx::query(
"DELETE FROM failed_paths_cache WHERE failed_at < NOW() - \

23
crates/common/src/repo/jobset_inputs.rs
View file

@ -7,6 +7,11 @@ use crate::{
models::JobsetInput,
};
/// Create a new jobset input.
///
/// # Errors
///
/// Returns error if database insert fails or input already exists.
pub async fn create(
pool: &PgPool,
jobset_id: Uuid,
@ -38,6 +43,11 @@ pub async fn create(
})
}
/// List all inputs for a jobset.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_jobset(
pool: &PgPool,
jobset_id: Uuid,
@ -51,6 +61,11 @@ pub async fn list_for_jobset(
.map_err(CiError::Database)
}
/// Delete a jobset input.
///
/// # Errors
///
/// Returns error if database delete fails or input not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM jobset_inputs WHERE id = $1")
.bind(id)
@ -63,6 +78,10 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
}
/// Upsert a jobset input (insert or update on conflict).
///
/// # Errors
///
/// Returns error if database operation fails.
pub async fn upsert(
pool: &PgPool,
jobset_id: Uuid,
@ -89,6 +108,10 @@ pub async fn upsert(
/// Sync jobset inputs from declarative config.
/// Deletes inputs not in the config and upserts those that are.
///
/// # Errors
///
/// Returns error if database operations fail.
pub async fn sync_for_jobset(
pool: &PgPool,
jobset_id: Uuid,

56
crates/common/src/repo/jobsets.rs
View file

@ -6,6 +6,11 @@ use crate::{
models::{ActiveJobset, CreateJobset, Jobset, JobsetState, UpdateJobset},
};
/// Create a new jobset with defaults applied.
///
/// # Errors
///
/// Returns error if database insert fails or jobset already exists.
pub async fn create(pool: &PgPool, input: CreateJobset) -> Result<Jobset> {
let state = input.state.unwrap_or(JobsetState::Enabled);
// Sync enabled with state if state was explicitly set, otherwise use
@ -50,6 +55,11 @@ pub async fn create(pool: &PgPool, input: CreateJobset) -> Result<Jobset> {
})
}
/// Get a jobset by ID.
///
/// # Errors
///
/// Returns error if database query fails or jobset not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<Jobset> {
sqlx::query_as::<_, Jobset>("SELECT * FROM jobsets WHERE id = $1")
.bind(id)
@ -58,6 +68,11 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<Jobset> {
.ok_or_else(|| CiError::NotFound(format!("Jobset {id} not found")))
}
/// List all jobsets for a project.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_project(
pool: &PgPool,
project_id: Uuid,
@ -76,6 +91,11 @@ pub async fn list_for_project(
.map_err(CiError::Database)
}
/// Count jobsets for a project.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn count_for_project(pool: &PgPool, project_id: Uuid) -> Result<i64> {
let row: (i64,) =
sqlx::query_as("SELECT COUNT(*) FROM jobsets WHERE project_id = $1")
@ -86,6 +106,11 @@ pub async fn count_for_project(pool: &PgPool, project_id: Uuid) -> Result<i64> {
Ok(row.0)
}
/// Update a jobset with partial fields.
///
/// # Errors
///
/// Returns error if database update fails or jobset not found.
pub async fn update(
pool: &PgPool,
id: Uuid,
@ -139,6 +164,11 @@ pub async fn update(
})
}
/// Delete a jobset.
///
/// # Errors
///
/// Returns error if database delete fails or jobset not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM jobsets WHERE id = $1")
.bind(id)
@ -152,6 +182,11 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
Ok(())
}
/// Insert or update a jobset by name.
///
/// # Errors
///
/// Returns error if database operation fails.
pub async fn upsert(pool: &PgPool, input: CreateJobset) -> Result<Jobset> {
let state = input.state.unwrap_or(JobsetState::Enabled);
// Sync enabled with state if state was explicitly set, otherwise use
@ -191,6 +226,11 @@ pub async fn upsert(pool: &PgPool, input: CreateJobset) -> Result<Jobset> {
.map_err(CiError::Database)
}
/// List all active jobsets with project info.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_active(pool: &PgPool) -> Result<Vec<ActiveJobset>> {
sqlx::query_as::<_, ActiveJobset>("SELECT * FROM active_jobsets")
.fetch_all(pool)
@ -199,6 +239,10 @@ pub async fn list_active(pool: &PgPool) -> Result<Vec<ActiveJobset>> {
}
/// Mark a one-shot jobset as complete (set state to disabled).
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn mark_one_shot_complete(pool: &PgPool, id: Uuid) -> Result<()> {
sqlx::query(
"UPDATE jobsets SET state = 'disabled', enabled = false WHERE id = $1 AND \
@ -212,6 +256,10 @@ pub async fn mark_one_shot_complete(pool: &PgPool, id: Uuid) -> Result<()> {
}
/// Update the `last_checked_at` timestamp for a jobset.
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn update_last_checked(pool: &PgPool, id: Uuid) -> Result<()> {
sqlx::query("UPDATE jobsets SET last_checked_at = NOW() WHERE id = $1")
.bind(id)
@ -222,6 +270,10 @@ pub async fn update_last_checked(pool: &PgPool, id: Uuid) -> Result<()> {
}
/// Check if a jobset has any running builds.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn has_running_builds(
pool: &PgPool,
jobset_id: Uuid,
@ -240,6 +292,10 @@ pub async fn has_running_builds(
/// List jobsets that are due for evaluation based on their `check_interval`.
/// Returns jobsets where `last_checked_at` is NULL or older than
/// `check_interval` seconds.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_due_for_eval(
pool: &PgPool,
limit: i64,

23
crates/common/src/repo/notification_configs.rs
View file

@ -7,6 +7,11 @@ use crate::{
models::{CreateNotificationConfig, NotificationConfig},
};
/// Create a new notification config.
///
/// # Errors
///
/// Returns error if database insert fails or config already exists.
pub async fn create(
pool: &PgPool,
input: CreateNotificationConfig,
@ -33,6 +38,11 @@ pub async fn create(
})
}
/// List all enabled notification configs for a project.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_project(
pool: &PgPool,
project_id: Uuid,
@ -47,6 +57,11 @@ pub async fn list_for_project(
.map_err(CiError::Database)
}
/// Delete a notification config.
///
/// # Errors
///
/// Returns error if database delete fails or config not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM notification_configs WHERE id = $1")
.bind(id)
@ -61,6 +76,10 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
}
/// Upsert a notification config (insert or update on conflict).
///
/// # Errors
///
/// Returns error if database operation fails.
pub async fn upsert(
pool: &PgPool,
project_id: Uuid,
@ -85,6 +104,10 @@ pub async fn upsert(
/// Sync notification configs from declarative config.
/// Deletes configs not in the declarative list and upserts those that are.
///
/// # Errors
///
/// Returns error if database operations fail.
pub async fn sync_for_project(
pool: &PgPool,
project_id: Uuid,

74
crates/common/src/repo/notification_tasks.rs
View file

@ -6,6 +6,10 @@ use uuid::Uuid;
use crate::{error::Result, models::NotificationTask};
/// Create a new notification task for later delivery
///
/// # Errors
///
/// Returns error if database insert fails.
pub async fn create(
pool: &PgPool,
notification_type: &str,
@ -13,11 +17,11 @@ pub async fn create(
max_attempts: i32,
) -> Result<NotificationTask> {
let task = sqlx::query_as::<_, NotificationTask>(
r#"
r"
INSERT INTO notification_tasks (notification_type, payload, max_attempts)
VALUES ($1, $2, $3)
RETURNING *
"#,
",
)
.bind(notification_type)
.bind(payload)
@ -29,19 +33,23 @@ pub async fn create(
}
/// Fetch pending tasks that are ready for retry
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_pending(
pool: &PgPool,
limit: i32,
) -> Result<Vec<NotificationTask>> {
let tasks = sqlx::query_as::<_, NotificationTask>(
r#"
r"
SELECT *
FROM notification_tasks
WHERE status = 'pending'
AND next_retry_at <= NOW()
ORDER BY next_retry_at ASC
LIMIT $1
"#,
",
)
.bind(limit)
.fetch_all(pool)
@ -51,14 +59,18 @@ pub async fn list_pending(
}
/// Mark a task as running (claimed by worker)
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn mark_running(pool: &PgPool, task_id: Uuid) -> Result<()> {
sqlx::query(
r#"
r"
UPDATE notification_tasks
SET status = 'running',
attempts = attempts + 1
WHERE id = $1
"#,
",
)
.bind(task_id)
.execute(pool)
@ -68,14 +80,18 @@ pub async fn mark_running(pool: &PgPool, task_id: Uuid) -> Result<()> {
}
/// Mark a task as completed successfully
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn mark_completed(pool: &PgPool, task_id: Uuid) -> Result<()> {
sqlx::query(
r#"
r"
UPDATE notification_tasks
SET status = 'completed',
completed_at = NOW()
WHERE id = $1
"#,
",
)
.bind(task_id)
.execute(pool)
@ -86,13 +102,17 @@ pub async fn mark_completed(pool: &PgPool, task_id: Uuid) -> Result<()> {
/// Mark a task as failed and schedule retry with exponential backoff
/// Backoff formula: 1s, 2s, 4s, 8s, 16s...
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn mark_failed_and_retry(
pool: &PgPool,
task_id: Uuid,
error: &str,
) -> Result<()> {
sqlx::query(
r#"
r"
UPDATE notification_tasks
SET status = CASE
WHEN attempts >= max_attempts THEN 'failed'::varchar
@ -101,14 +121,14 @@ pub async fn mark_failed_and_retry(
last_error = $2,
next_retry_at = CASE
WHEN attempts >= max_attempts THEN NOW()
ELSE NOW() + (POWER(2, attempts) || ' seconds')::interval
ELSE NOW() + (POWER(2, attempts - 1) || ' seconds')::interval
END,
completed_at = CASE
WHEN attempts >= max_attempts THEN NOW()
ELSE NULL
END
WHERE id = $1
"#,
",
)
.bind(task_id)
.bind(error)
@ -119,11 +139,15 @@ pub async fn mark_failed_and_retry(
}
/// Get task by ID
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get(pool: &PgPool, task_id: Uuid) -> Result<NotificationTask> {
let task = sqlx::query_as::<_, NotificationTask>(
r#"
r"
SELECT * FROM notification_tasks WHERE id = $1
"#,
",
)
.bind(task_id)
.fetch_one(pool)
@ -133,17 +157,21 @@ pub async fn get(pool: &PgPool, task_id: Uuid) -> Result<NotificationTask> {
}
/// Clean up old completed/failed tasks (older than retention days)
///
/// # Errors
///
/// Returns error if database delete fails.
pub async fn cleanup_old_tasks(
pool: &PgPool,
retention_days: i64,
) -> Result<u64> {
let result = sqlx::query(
r#"
r"
DELETE FROM notification_tasks
WHERE status IN ('completed', 'failed')
AND (completed_at < NOW() - ($1 || ' days')::interval
OR created_at < NOW() - ($1 || ' days')::interval)
"#,
",
)
.bind(retention_days)
.execute(pool)
@ -153,11 +181,15 @@ pub async fn cleanup_old_tasks(
}
/// Count pending tasks (for monitoring)
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn count_pending(pool: &PgPool) -> Result<i64> {
let count: (i64,) = sqlx::query_as(
r#"
r"
SELECT COUNT(*) FROM notification_tasks WHERE status = 'pending'
"#,
",
)
.fetch_one(pool)
.await?;
@ -166,11 +198,15 @@ pub async fn count_pending(pool: &PgPool) -> Result<i64> {
}
/// Count failed tasks (for monitoring)
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn count_failed(pool: &PgPool) -> Result<i64> {
let count: (i64,) = sqlx::query_as(
r#"
r"
SELECT COUNT(*) FROM notification_tasks WHERE status = 'failed'
"#,
",
)
.fetch_one(pool)
.await?;

44
crates/common/src/repo/project_members.rs
View file

@ -12,6 +12,10 @@ use crate::{
};
/// Add a member to a project with role validation
///
/// # Errors
///
/// Returns error if validation fails or database insert fails.
pub async fn create(
pool: &PgPool,
project_id: Uuid,
@ -43,6 +47,10 @@ pub async fn create(
}
/// Get a project member by ID
///
/// # Errors
///
/// Returns error if database query fails or member not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<ProjectMember> {
sqlx::query_as::<_, ProjectMember>(
"SELECT * FROM project_members WHERE id = $1",
@ -61,6 +69,10 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<ProjectMember> {
}
/// Get a project member by project and user
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_by_project_and_user(
pool: &PgPool,
project_id: Uuid,
@ -77,6 +89,10 @@ pub async fn get_by_project_and_user(
}
/// List all members of a project
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_project(
pool: &PgPool,
project_id: Uuid,
@ -91,6 +107,10 @@ pub async fn list_for_project(
}
/// List all projects a user is a member of
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_user(
pool: &PgPool,
user_id: Uuid,
@ -105,6 +125,10 @@ pub async fn list_for_user(
}
/// Update a project member's role with validation
///
/// # Errors
///
/// Returns error if validation fails or database update fails.
pub async fn update(
pool: &PgPool,
id: Uuid,
@ -135,6 +159,10 @@ pub async fn update(
}
/// Remove a member from a project
///
/// # Errors
///
/// Returns error if database delete fails or member not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM project_members WHERE id = $1")
.bind(id)
@ -147,6 +175,10 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
}
/// Remove a specific user from a project
///
/// # Errors
///
/// Returns error if database delete fails or user not found.
pub async fn delete_by_project_and_user(
pool: &PgPool,
project_id: Uuid,
@ -168,6 +200,10 @@ pub async fn delete_by_project_and_user(
}
/// Check if a user has a specific role or higher in a project
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn check_permission(
pool: &PgPool,
project_id: Uuid,
@ -186,6 +222,10 @@ pub async fn check_permission(
}
/// Upsert a project member (insert or update on conflict).
///
/// # Errors
///
/// Returns error if validation fails or database operation fails.
pub async fn upsert(
pool: &PgPool,
project_id: Uuid,
@ -211,6 +251,10 @@ pub async fn upsert(
/// Sync project members from declarative config.
/// Deletes members not in the declarative list and upserts those that are.
///
/// # Errors
///
/// Returns error if database operations fail.
pub async fn sync_for_project(
pool: &PgPool,
project_id: Uuid,

42
crates/common/src/repo/projects.rs
View file

@ -6,6 +6,11 @@ use crate::{
models::{CreateProject, Project, UpdateProject},
};
/// Create a new project.
///
/// # Errors
///
/// Returns error if database insert fails or project name already exists.
pub async fn create(pool: &PgPool, input: CreateProject) -> Result<Project> {
sqlx::query_as::<_, Project>(
"INSERT INTO projects (name, description, repository_url) VALUES ($1, $2, \
@ -26,6 +31,11 @@ pub async fn create(pool: &PgPool, input: CreateProject) -> Result<Project> {
})
}
/// Get a project by ID.
///
/// # Errors
///
/// Returns error if database query fails or project not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<Project> {
sqlx::query_as::<_, Project>("SELECT * FROM projects WHERE id = $1")
.bind(id)
@ -34,6 +44,11 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<Project> {
.ok_or_else(|| CiError::NotFound(format!("Project {id} not found")))
}
/// Get a project by name.
///
/// # Errors
///
/// Returns error if database query fails or project not found.
pub async fn get_by_name(pool: &PgPool, name: &str) -> Result<Project> {
sqlx::query_as::<_, Project>("SELECT * FROM projects WHERE name = $1")
.bind(name)
@ -42,6 +57,11 @@ pub async fn get_by_name(pool: &PgPool, name: &str) -> Result<Project> {
.ok_or_else(|| CiError::NotFound(format!("Project '{name}' not found")))
}
/// List projects with pagination.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list(
pool: &PgPool,
limit: i64,
@ -57,6 +77,11 @@ pub async fn list(
.map_err(CiError::Database)
}
/// Count total number of projects.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn count(pool: &PgPool) -> Result<i64> {
let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM projects")
.fetch_one(pool)
@ -65,12 +90,17 @@ pub async fn count(pool: &PgPool) -> Result<i64> {
Ok(row.0)
}
/// Update a project with partial fields.
///
/// # Errors
///
/// Returns error if database update fails or project not found.
pub async fn update(
pool: &PgPool,
id: Uuid,
input: UpdateProject,
) -> Result<Project> {
// Build dynamic update — only set provided fields
// Dynamic update - only set provided fields
let existing = get(pool, id).await?;
let name = input.name.unwrap_or(existing.name);
@ -97,6 +127,11 @@ pub async fn update(
})
}
/// Insert or update a project by name.
///
/// # Errors
///
/// Returns error if database operation fails.
pub async fn upsert(pool: &PgPool, input: CreateProject) -> Result<Project> {
sqlx::query_as::<_, Project>(
"INSERT INTO projects (name, description, repository_url) VALUES ($1, $2, \
@ -111,6 +146,11 @@ pub async fn upsert(pool: &PgPool, input: CreateProject) -> Result<Project> {
.map_err(CiError::Database)
}
/// Delete a project by ID.
///
/// # Errors
///
/// Returns error if database delete fails or project not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM projects WHERE id = $1")
.bind(id)

122
crates/common/src/repo/remote_builders.rs
View file

@ -7,6 +7,11 @@ use crate::{
models::{CreateRemoteBuilder, RemoteBuilder},
};
/// Create a new remote builder.
///
/// # Errors
///
/// Returns error if database insert fails or builder already exists.
pub async fn create(
pool: &PgPool,
input: CreateRemoteBuilder,
@ -40,6 +45,11 @@ pub async fn create(
})
}
/// Get a remote builder by ID.
///
/// # Errors
///
/// Returns error if database query fails or builder not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<RemoteBuilder> {
sqlx::query_as::<_, RemoteBuilder>(
"SELECT * FROM remote_builders WHERE id = $1",
@ -50,6 +60,11 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<RemoteBuilder> {
.ok_or_else(|| CiError::NotFound(format!("Remote builder {id} not found")))
}
/// List all remote builders.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list(pool: &PgPool) -> Result<Vec<RemoteBuilder>> {
sqlx::query_as::<_, RemoteBuilder>(
"SELECT * FROM remote_builders ORDER BY speed_factor DESC, name",
@ -59,6 +74,11 @@ pub async fn list(pool: &PgPool) -> Result<Vec<RemoteBuilder>> {
.map_err(CiError::Database)
}
/// List all enabled remote builders.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_enabled(pool: &PgPool) -> Result<Vec<RemoteBuilder>> {
sqlx::query_as::<_, RemoteBuilder>(
"SELECT * FROM remote_builders WHERE enabled = true ORDER BY speed_factor \
@ -71,6 +91,10 @@ pub async fn list_enabled(pool: &PgPool) -> Result<Vec<RemoteBuilder>> {
/// Find a suitable builder for the given system.
/// Excludes builders that are temporarily disabled due to consecutive failures.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn find_for_system(
pool: &PgPool,
system: &str,
@ -87,9 +111,14 @@ pub async fn find_for_system(
}
/// Record a build failure for a remote builder.
/// Increments consecutive_failures (capped at 4), sets last_failure,
/// and computes disabled_until with exponential backoff.
///
/// Increments `consecutive_failures` (capped at 4), sets `last_failure`,
/// and computes `disabled_until` with exponential backoff.
/// Backoff formula (from Hydra): delta = 60 * 3^(min(failures, 4) - 1) seconds.
///
/// # Errors
///
/// Returns error if database update fails or builder not found.
pub async fn record_failure(pool: &PgPool, id: Uuid) -> Result<RemoteBuilder> {
sqlx::query_as::<_, RemoteBuilder>(
"UPDATE remote_builders SET consecutive_failures = \
@ -105,7 +134,11 @@ pub async fn record_failure(pool: &PgPool, id: Uuid) -> Result<RemoteBuilder> {
}
/// Record a build success for a remote builder.
/// Resets consecutive_failures and clears disabled_until.
/// Resets `consecutive_failures` and clears `disabled_until`.
///
/// # Errors
///
/// Returns error if database update fails or builder not found.
pub async fn record_success(pool: &PgPool, id: Uuid) -> Result<RemoteBuilder> {
sqlx::query_as::<_, RemoteBuilder>(
"UPDATE remote_builders SET consecutive_failures = 0, disabled_until = \
@ -117,12 +150,17 @@ pub async fn record_success(pool: &PgPool, id: Uuid) -> Result<RemoteBuilder> {
.ok_or_else(|| CiError::NotFound(format!("Remote builder {id} not found")))
}
/// Update a remote builder with partial fields.
///
/// # Errors
///
/// Returns error if database update fails or builder not found.
pub async fn update(
pool: &PgPool,
id: Uuid,
input: crate::models::UpdateRemoteBuilder,
) -> Result<RemoteBuilder> {
// Build dynamic update — use COALESCE pattern
// Dynamic update using COALESCE pattern
sqlx::query_as::<_, RemoteBuilder>(
"UPDATE remote_builders SET name = COALESCE($1, name), ssh_uri = \
COALESCE($2, ssh_uri), systems = COALESCE($3, systems), max_jobs = \
@ -148,6 +186,11 @@ pub async fn update(
.ok_or_else(|| CiError::NotFound(format!("Remote builder {id} not found")))
}
/// Delete a remote builder.
///
/// # Errors
///
/// Returns error if database delete fails or builder not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM remote_builders WHERE id = $1")
.bind(id)
@ -160,6 +203,11 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
Ok(())
}
/// Count total remote builders.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn count(pool: &PgPool) -> Result<i64> {
let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM remote_builders")
.fetch_one(pool)
@ -169,18 +217,13 @@ pub async fn count(pool: &PgPool) -> Result<i64> {
}
/// Upsert a remote builder (insert or update on conflict by name).
///
/// # Errors
///
/// Returns error if database operation fails.
pub async fn upsert(
pool: &PgPool,
name: &str,
ssh_uri: &str,
systems: &[String],
max_jobs: i32,
speed_factor: i32,
supported_features: &[String],
mandatory_features: &[String],
enabled: bool,
public_host_key: Option<&str>,
ssh_key_file: Option<&str>,
params: &crate::models::RemoteBuilderParams<'_>,
) -> Result<RemoteBuilder> {
sqlx::query_as::<_, RemoteBuilder>(
"INSERT INTO remote_builders (name, ssh_uri, systems, max_jobs, \
@ -194,16 +237,16 @@ pub async fn upsert(
remote_builders.public_host_key), ssh_key_file = \
COALESCE(EXCLUDED.ssh_key_file, remote_builders.ssh_key_file) RETURNING *",
)
.bind(name)
.bind(ssh_uri)
.bind(systems)
.bind(max_jobs)
.bind(speed_factor)
.bind(supported_features)
.bind(mandatory_features)
.bind(enabled)
.bind(public_host_key)
.bind(ssh_key_file)
.bind(params.name)
.bind(params.ssh_uri)
.bind(params.systems)
.bind(params.max_jobs)
.bind(params.speed_factor)
.bind(params.supported_features)
.bind(params.mandatory_features)
.bind(params.enabled)
.bind(params.public_host_key)
.bind(params.ssh_key_file)
.fetch_one(pool)
.await
.map_err(CiError::Database)
@ -211,6 +254,10 @@ pub async fn upsert(
/// Sync remote builders from declarative config.
/// Deletes builders not in the declarative list and upserts those that are.
///
/// # Errors
///
/// Returns error if database operations fail.
pub async fn sync_all(
pool: &PgPool,
builders: &[DeclarativeRemoteBuilder],
@ -227,20 +274,19 @@ pub async fn sync_all(
// Upsert each builder
for builder in builders {
upsert(
pool,
&builder.name,
&builder.ssh_uri,
&builder.systems,
builder.max_jobs,
builder.speed_factor,
&builder.supported_features,
&builder.mandatory_features,
builder.enabled,
builder.public_host_key.as_deref(),
builder.ssh_key_file.as_deref(),
)
.await?;
let params = crate::models::RemoteBuilderParams {
name: &builder.name,
ssh_uri: &builder.ssh_uri,
systems: &builder.systems,
max_jobs: builder.max_jobs,
speed_factor: builder.speed_factor,
supported_features: &builder.supported_features,
mandatory_features: &builder.mandatory_features,
enabled: builder.enabled,
public_host_key: builder.public_host_key.as_deref(),
ssh_key_file: builder.ssh_key_file.as_deref(),
};
upsert(pool, &params).await?;
}
Ok(())

8
crates/common/src/repo/search.rs
View file

@ -146,6 +146,10 @@ pub struct SearchResults {
}
/// Execute a comprehensive search across all entities
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn search(
pool: &PgPool,
params: &SearchParams,
@ -511,6 +515,10 @@ async fn search_builds(
}
/// Quick search - simple text search across entities
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn quick_search(
pool: &PgPool,
query: &str,

32
crates/common/src/repo/starred_jobs.rs
View file

@ -9,6 +9,10 @@ use crate::{
};
/// Create a new starred job
///
/// # Errors
///
/// Returns error if database insert fails or job already starred.
pub async fn create(
pool: &PgPool,
user_id: Uuid,
@ -35,6 +39,10 @@ pub async fn create(
}
/// Get a starred job by ID
///
/// # Errors
///
/// Returns error if database query fails or starred job not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<StarredJob> {
sqlx::query_as::<_, StarredJob>("SELECT * FROM starred_jobs WHERE id = $1")
.bind(id)
@ -51,6 +59,10 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<StarredJob> {
}
/// List starred jobs for a user with pagination
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_user(
pool: &PgPool,
user_id: Uuid,
@ -70,6 +82,10 @@ pub async fn list_for_user(
}
/// Count starred jobs for a user
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn count_for_user(pool: &PgPool, user_id: Uuid) -> Result<i64> {
let (count,): (i64,) =
sqlx::query_as("SELECT COUNT(*) FROM starred_jobs WHERE user_id = $1")
@ -80,6 +96,10 @@ pub async fn count_for_user(pool: &PgPool, user_id: Uuid) -> Result<i64> {
}
/// Check if a user has starred a specific job
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn is_starred(
pool: &PgPool,
user_id: Uuid,
@ -101,6 +121,10 @@ pub async fn is_starred(
}
/// Delete a starred job
///
/// # Errors
///
/// Returns error if database delete fails or starred job not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM starred_jobs WHERE id = $1")
.bind(id)
@ -113,6 +137,10 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
}
/// Delete a starred job by user and job details
///
/// # Errors
///
/// Returns error if database delete fails or starred job not found.
pub async fn delete_by_job(
pool: &PgPool,
user_id: Uuid,
@ -137,6 +165,10 @@ pub async fn delete_by_job(
}
/// Delete all starred jobs for a user (when user is deleted)
///
/// # Errors
///
/// Returns error if database delete fails.
pub async fn delete_all_for_user(pool: &PgPool, user_id: Uuid) -> Result<()> {
sqlx::query("DELETE FROM starred_jobs WHERE user_id = $1")
.bind(user_id)

89
crates/common/src/repo/users.rs
View file

@ -17,6 +17,10 @@ use crate::{
};
/// Hash a password using argon2id
///
/// # Errors
///
/// Returns error if password hashing fails.
pub fn hash_password(password: &str) -> Result<String> {
use argon2::{
Argon2,
@ -33,6 +37,10 @@ pub fn hash_password(password: &str) -> Result<String> {
}
/// Verify a password against a hash
///
/// # Errors
///
/// Returns error if password hash parsing fails.
pub fn verify_password(password: &str, hash: &str) -> Result<bool> {
use argon2::{Argon2, PasswordHash, PasswordVerifier};
@ -47,6 +55,10 @@ pub fn verify_password(password: &str, hash: &str) -> Result<bool> {
}
/// Create a new user with validation
///
/// # Errors
///
/// Returns error if validation fails or database insert fails.
pub async fn create(pool: &PgPool, data: &CreateUser) -> Result<User> {
// Validate username
validate_username(&data.username)
@ -94,6 +106,10 @@ pub async fn create(pool: &PgPool, data: &CreateUser) -> Result<User> {
}
/// Authenticate a user with username and password
///
/// # Errors
///
/// Returns error if credentials are invalid or database query fails.
pub async fn authenticate(
pool: &PgPool,
creds: &LoginCredentials,
@ -129,6 +145,10 @@ pub async fn authenticate(
}
/// Get a user by ID
///
/// # Errors
///
/// Returns error if database query fails or user not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<User> {
sqlx::query_as::<_, User>("SELECT * FROM users WHERE id = $1")
.bind(id)
@ -145,6 +165,10 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<User> {
}
/// Get a user by username
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_by_username(
pool: &PgPool,
username: &str,
@ -157,6 +181,10 @@ pub async fn get_by_username(
}
/// Get a user by email
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_by_email(pool: &PgPool, email: &str) -> Result<Option<User>> {
sqlx::query_as::<_, User>("SELECT * FROM users WHERE email = $1")
.bind(email)
@ -166,6 +194,10 @@ pub async fn get_by_email(pool: &PgPool, email: &str) -> Result<Option<User>> {
}
/// List all users with pagination
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list(pool: &PgPool, limit: i64, offset: i64) -> Result<Vec<User>> {
sqlx::query_as::<_, User>(
"SELECT * FROM users ORDER BY created_at DESC LIMIT $1 OFFSET $2",
@ -178,6 +210,10 @@ pub async fn list(pool: &PgPool, limit: i64, offset: i64) -> Result<Vec<User>> {
}
/// Count total users
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn count(pool: &PgPool) -> Result<i64> {
let (count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM users")
.fetch_one(pool)
@ -186,6 +222,10 @@ pub async fn count(pool: &PgPool) -> Result<i64> {
}
/// Update a user with the provided data
///
/// # Errors
///
/// Returns error if validation fails or database update fails.
pub async fn update(
pool: &PgPool,
id: Uuid,
@ -220,6 +260,10 @@ pub async fn update(
}
/// Update user email with validation
///
/// # Errors
///
/// Returns error if validation fails or database update fails.
pub async fn update_email(
pool: &PgPool,
id: Uuid,
@ -245,6 +289,10 @@ pub async fn update_email(
}
/// Update user full name with validation
///
/// # Errors
///
/// Returns error if validation fails or database update fails.
pub async fn update_full_name(
pool: &PgPool,
id: Uuid,
@ -263,6 +311,10 @@ pub async fn update_full_name(
}
/// Update user password with validation
///
/// # Errors
///
/// Returns error if validation fails or database update fails.
pub async fn update_password(
pool: &PgPool,
id: Uuid,
@ -281,6 +333,10 @@ pub async fn update_password(
}
/// Update user role with validation
///
/// # Errors
///
/// Returns error if validation fails or database update fails.
pub async fn update_role(pool: &PgPool, id: Uuid, role: &str) -> Result<()> {
validate_role(role, VALID_ROLES)
.map_err(|e| CiError::Validation(e.to_string()))?;
@ -294,6 +350,10 @@ pub async fn update_role(pool: &PgPool, id: Uuid, role: &str) -> Result<()> {
}
/// Enable/disable user
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn set_enabled(pool: &PgPool, id: Uuid, enabled: bool) -> Result<()> {
sqlx::query("UPDATE users SET enabled = $1 WHERE id = $2")
.bind(enabled)
@ -304,6 +364,10 @@ pub async fn set_enabled(pool: &PgPool, id: Uuid, enabled: bool) -> Result<()> {
}
/// Set public dashboard preference
///
/// # Errors
///
/// Returns error if database update fails.
pub async fn set_public_dashboard(
pool: &PgPool,
id: Uuid,
@ -318,6 +382,10 @@ pub async fn set_public_dashboard(
}
/// Delete a user
///
/// # Errors
///
/// Returns error if database delete fails or user not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM users WHERE id = $1")
.bind(id)
@ -330,6 +398,10 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
}
/// Create or update OAuth user
///
/// # Errors
///
/// Returns error if validation fails or database operation fails.
pub async fn upsert_oauth_user(
pool: &PgPool,
username: &str,
@ -399,6 +471,10 @@ pub async fn upsert_oauth_user(
}
/// Create a new session for a user. Returns (`session_token`, `session_id`).
///
/// # Errors
///
/// Returns error if database insert fails.
pub async fn create_session(
pool: &PgPool,
user_id: Uuid,
@ -427,6 +503,10 @@ pub async fn create_session(
}
/// Validate a session token and return the associated user if valid.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn validate_session(
pool: &PgPool,
token: &str,
@ -444,17 +524,16 @@ pub async fn validate_session(
.await?;
// Update last_used_at
if result.is_some() {
if let Err(e) = sqlx::query(
if result.is_some()
&& let Err(e) = sqlx::query(
"UPDATE user_sessions SET last_used_at = NOW() WHERE session_token_hash \
= $1",
)
.bind(&token_hash)
.execute(pool)
.await
{
tracing::warn!(token_hash = %token_hash, "Failed to update session last_used_at: {e}");
}
{
tracing::warn!(token_hash = %token_hash, "Failed to update session last_used_at: {e}");
}
Ok(result)

33
crates/common/src/repo/webhook_configs.rs
View file

@ -7,6 +7,11 @@ use crate::{
models::{CreateWebhookConfig, WebhookConfig},
};
/// Create a new webhook config.
///
/// # Errors
///
/// Returns error if database insert fails or config already exists.
pub async fn create(
pool: &PgPool,
input: CreateWebhookConfig,
@ -34,6 +39,11 @@ pub async fn create(
})
}
/// Get a webhook config by ID.
///
/// # Errors
///
/// Returns error if database query fails or config not found.
pub async fn get(pool: &PgPool, id: Uuid) -> Result<WebhookConfig> {
sqlx::query_as::<_, WebhookConfig>(
"SELECT * FROM webhook_configs WHERE id = $1",
@ -44,6 +54,11 @@ pub async fn get(pool: &PgPool, id: Uuid) -> Result<WebhookConfig> {
.ok_or_else(|| CiError::NotFound(format!("Webhook config {id} not found")))
}
/// List all webhook configs for a project.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn list_for_project(
pool: &PgPool,
project_id: Uuid,
@ -58,6 +73,11 @@ pub async fn list_for_project(
.map_err(CiError::Database)
}
/// Get a webhook config by project and forge type.
///
/// # Errors
///
/// Returns error if database query fails.
pub async fn get_by_project_and_forge(
pool: &PgPool,
project_id: Uuid,
@ -74,6 +94,11 @@ pub async fn get_by_project_and_forge(
.map_err(CiError::Database)
}
/// Delete a webhook config.
///
/// # Errors
///
/// Returns error if database delete fails or config not found.
pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
let result = sqlx::query("DELETE FROM webhook_configs WHERE id = $1")
.bind(id)
@ -86,6 +111,10 @@ pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
}
/// Upsert a webhook config (insert or update on conflict).
///
/// # Errors
///
/// Returns error if database operation fails.
pub async fn upsert(
pool: &PgPool,
project_id: Uuid,
@ -110,6 +139,10 @@ pub async fn upsert(
/// Sync webhook configs from declarative config.
/// Deletes configs not in the declarative list and upserts those that are.
///
/// # Errors
///
/// Returns error if database operations fail.
pub async fn sync_for_project(
pool: &PgPool,
project_id: Uuid,