crates: production models and repo layer

Signed-off-by: NotAShelf <raf@notashelf.dev>
Change-Id: Iceb76724c09eaca7ca5d823010db76776a6a6964

crates/common/src/repo/api_keys.rs

@@ -0,0 +1,57 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::ApiKey;
+
+pub async fn create(pool: &PgPool, name: &str, key_hash: &str, role: &str) -> Result<ApiKey> {
+    sqlx::query_as::<_, ApiKey>(
+        "INSERT INTO api_keys (name, key_hash, role) VALUES ($1, $2, $3) RETURNING *",
+    )
+    .bind(name)
+    .bind(key_hash)
+    .bind(role)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!("API key with this hash already exists"))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn get_by_hash(pool: &PgPool, key_hash: &str) -> Result<Option<ApiKey>> {
+    sqlx::query_as::<_, ApiKey>("SELECT * FROM api_keys WHERE key_hash = $1")
+        .bind(key_hash)
+        .fetch_optional(pool)
+        .await
+        .map_err(CiError::Database)
+}
+
+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)
+        .await
+        .map_err(CiError::Database)
+}
+
+pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
+    let result = sqlx::query("DELETE FROM api_keys WHERE id = $1")
+        .bind(id)
+        .execute(pool)
+        .await?;
+    if result.rows_affected() == 0 {
+        return Err(CiError::NotFound(format!("API key {id} not found")));
+    }
+    Ok(())
+}
+
+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)
+        .execute(pool)
+        .await
+        .map_err(CiError::Database)?;
+    Ok(())
+}

crates/common/src/repo/build_dependencies.rs

@@ -0,0 +1,79 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::BuildDependency;
+
+pub async fn create(
+    pool: &PgPool,
+    build_id: Uuid,
+    dependency_build_id: Uuid,
+) -> Result<BuildDependency> {
+    sqlx::query_as::<_, BuildDependency>(
+        "INSERT INTO build_dependencies (build_id, dependency_build_id) VALUES ($1, $2) RETURNING *",
+    )
+    .bind(build_id)
+    .bind(dependency_build_id)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!(
+                "Dependency from {build_id} to {dependency_build_id} already exists"
+            ))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn list_for_build(pool: &PgPool, build_id: Uuid) -> Result<Vec<BuildDependency>> {
+    sqlx::query_as::<_, BuildDependency>("SELECT * FROM build_dependencies WHERE build_id = $1")
+        .bind(build_id)
+        .fetch_all(pool)
+        .await
+        .map_err(CiError::Database)
+}
+
+/// 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.
+pub async fn check_deps_for_builds(
+    pool: &PgPool,
+    build_ids: &[Uuid],
+) -> Result<std::collections::HashMap<Uuid, bool>> {
+    if build_ids.is_empty() {
+        return Ok(std::collections::HashMap::new());
+    }
+
+    // Find build_ids that have incomplete deps
+    let rows: Vec<(Uuid,)> = sqlx::query_as(
+        "SELECT DISTINCT bd.build_id FROM build_dependencies bd \
+         JOIN builds b ON bd.dependency_build_id = b.id \
+         WHERE bd.build_id = ANY($1) AND b.status != 'completed'",
+    )
+    .bind(build_ids)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)?;
+
+    let incomplete: std::collections::HashSet<Uuid> = rows.into_iter().map(|(id,)| id).collect();
+
+    Ok(build_ids
+        .iter()
+        .map(|id| (*id, !incomplete.contains(id)))
+        .collect())
+}
+
+/// Check if all dependency builds for a given build are completed.
+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 bd.dependency_build_id = b.id \
+         WHERE bd.build_id = $1 AND b.status != 'completed'",
+    )
+    .bind(build_id)
+    .fetch_one(pool)
+    .await
+    .map_err(CiError::Database)?;
+
+    Ok(row.0 == 0)
+}

crates/common/src/repo/build_products.rs

@@ -0,0 +1,40 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::{BuildProduct, CreateBuildProduct};
+
+pub async fn create(pool: &PgPool, input: CreateBuildProduct) -> Result<BuildProduct> {
+    sqlx::query_as::<_, BuildProduct>(
+        "INSERT INTO build_products (build_id, name, path, sha256_hash, file_size, content_type, is_directory) \
+         VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING *",
+    )
+    .bind(input.build_id)
+    .bind(&input.name)
+    .bind(&input.path)
+    .bind(&input.sha256_hash)
+    .bind(input.file_size)
+    .bind(&input.content_type)
+    .bind(input.is_directory)
+    .fetch_one(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn get(pool: &PgPool, id: Uuid) -> Result<BuildProduct> {
+    sqlx::query_as::<_, BuildProduct>("SELECT * FROM build_products WHERE id = $1")
+        .bind(id)
+        .fetch_optional(pool)
+        .await?
+        .ok_or_else(|| CiError::NotFound(format!("Build product {id} not found")))
+}
+
+pub async fn list_for_build(pool: &PgPool, build_id: Uuid) -> Result<Vec<BuildProduct>> {
+    sqlx::query_as::<_, BuildProduct>(
+        "SELECT * FROM build_products WHERE build_id = $1 ORDER BY created_at ASC",
+    )
+    .bind(build_id)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}

crates/common/src/repo/build_steps.rs

@@ -0,0 +1,54 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::{BuildStep, CreateBuildStep};
+
+pub async fn create(pool: &PgPool, input: CreateBuildStep) -> Result<BuildStep> {
+    sqlx::query_as::<_, BuildStep>(
+        "INSERT INTO build_steps (build_id, step_number, command) VALUES ($1, $2, $3) RETURNING *",
+    )
+    .bind(input.build_id)
+    .bind(input.step_number)
+    .bind(&input.command)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!(
+                "Build step {} already exists for this build",
+                input.step_number
+            ))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn complete(
+    pool: &PgPool,
+    id: Uuid,
+    exit_code: i32,
+    output: Option<&str>,
+    error_output: Option<&str>,
+) -> Result<BuildStep> {
+    sqlx::query_as::<_, BuildStep>(
+        "UPDATE build_steps SET completed_at = NOW(), exit_code = $1, output = $2, error_output = $3 WHERE id = $4 RETURNING *",
+    )
+    .bind(exit_code)
+    .bind(output)
+    .bind(error_output)
+    .bind(id)
+    .fetch_optional(pool)
+    .await?
+    .ok_or_else(|| CiError::NotFound(format!("Build step {id} not found")))
+}
+
+pub async fn list_for_build(pool: &PgPool, build_id: Uuid) -> Result<Vec<BuildStep>> {
+    sqlx::query_as::<_, BuildStep>(
+        "SELECT * FROM build_steps WHERE build_id = $1 ORDER BY step_number ASC",
+    )
+    .bind(build_id)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}

crates/common/src/repo/builds.rs

@@ -0,0 +1,292 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::{Build, BuildStats, BuildStatus, CreateBuild};
+
+pub async fn create(pool: &PgPool, input: CreateBuild) -> Result<Build> {
+    let is_aggregate = input.is_aggregate.unwrap_or(false);
+    sqlx::query_as::<_, Build>(
+        "INSERT INTO builds (evaluation_id, job_name, drv_path, status, system, outputs, is_aggregate, constituents) \
+         VALUES ($1, $2, $3, 'pending', $4, $5, $6, $7) RETURNING *",
+    )
+    .bind(input.evaluation_id)
+    .bind(&input.job_name)
+    .bind(&input.drv_path)
+    .bind(&input.system)
+    .bind(&input.outputs)
+    .bind(is_aggregate)
+    .bind(&input.constituents)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!(
+                "Build for job '{}' already exists in this evaluation",
+                input.job_name
+            ))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn get_completed_by_drv_path(pool: &PgPool, drv_path: &str) -> Result<Option<Build>> {
+    sqlx::query_as::<_, Build>(
+        "SELECT * FROM builds WHERE drv_path = $1 AND status = 'completed' LIMIT 1",
+    )
+    .bind(drv_path)
+    .fetch_optional(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn get(pool: &PgPool, id: Uuid) -> Result<Build> {
+    sqlx::query_as::<_, Build>("SELECT * FROM builds WHERE id = $1")
+        .bind(id)
+        .fetch_optional(pool)
+        .await?
+        .ok_or_else(|| CiError::NotFound(format!("Build {id} not found")))
+}
+
+pub async fn list_for_evaluation(pool: &PgPool, evaluation_id: Uuid) -> Result<Vec<Build>> {
+    sqlx::query_as::<_, Build>(
+        "SELECT * FROM builds WHERE evaluation_id = $1 ORDER BY created_at DESC",
+    )
+    .bind(evaluation_id)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn list_pending(pool: &PgPool, limit: i64) -> Result<Vec<Build>> {
+    sqlx::query_as::<_, Build>(
+        "SELECT * FROM builds WHERE status = 'pending' ORDER BY priority DESC, created_at ASC LIMIT $1",
+    )
+    .bind(limit)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+/// Atomically claim a pending build by setting it to running.
+/// Returns `None` if the build was already claimed by another worker.
+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 AND status = 'pending' RETURNING *",
+    )
+    .bind(id)
+    .fetch_optional(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn complete(
+    pool: &PgPool,
+    id: Uuid,
+    status: BuildStatus,
+    log_path: Option<&str>,
+    build_output_path: Option<&str>,
+    error_message: Option<&str>,
+) -> Result<Build> {
+    sqlx::query_as::<_, Build>(
+        "UPDATE builds SET status = $1, completed_at = NOW(), log_path = $2, build_output_path = $3, error_message = $4 WHERE id = $5 RETURNING *",
+    )
+    .bind(status)
+    .bind(log_path)
+    .bind(build_output_path)
+    .bind(error_message)
+    .bind(id)
+    .fetch_optional(pool)
+    .await?
+    .ok_or_else(|| CiError::NotFound(format!("Build {id} not found")))
+}
+
+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")
+        .bind(limit)
+        .fetch_all(pool)
+        .await
+        .map_err(CiError::Database)
+}
+
+pub async fn list_for_project(pool: &PgPool, project_id: Uuid) -> Result<Vec<Build>> {
+    sqlx::query_as::<_, Build>(
+        "SELECT b.* FROM builds b \
+         JOIN evaluations e ON b.evaluation_id = e.id \
+         JOIN jobsets j ON e.jobset_id = j.id \
+         WHERE j.project_id = $1 \
+         ORDER BY b.created_at DESC",
+    )
+    .bind(project_id)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn get_stats(pool: &PgPool) -> Result<BuildStats> {
+    sqlx::query_as::<_, BuildStats>("SELECT * FROM build_stats")
+        .fetch_optional(pool)
+        .await
+        .map_err(CiError::Database)
+        .map(|opt| opt.unwrap_or_default())
+}
+
+/// Reset builds that were left in 'running' state (orphaned by a crashed runner).
+/// Limited to 50 builds per call to prevent thundering herd.
+pub async fn reset_orphaned(pool: &PgPool, older_than_secs: i64) -> Result<u64> {
+    let result = sqlx::query(
+        "UPDATE builds SET status = 'pending', started_at = NULL \
+         WHERE id IN (SELECT id FROM builds WHERE status = 'running' \
+         AND started_at < NOW() - make_interval(secs => $1) LIMIT 50)",
+    )
+    .bind(older_than_secs)
+    .execute(pool)
+    .await
+    .map_err(CiError::Database)?;
+
+    Ok(result.rows_affected())
+}
+
+/// List builds with optional evaluation_id, status, system, and job_name filters, with pagination.
+pub async fn list_filtered(
+    pool: &PgPool,
+    evaluation_id: Option<Uuid>,
+    status: Option<&str>,
+    system: Option<&str>,
+    job_name: Option<&str>,
+    limit: i64,
+    offset: i64,
+) -> Result<Vec<Build>> {
+    sqlx::query_as::<_, Build>(
+        "SELECT * FROM builds \
+         WHERE ($1::uuid IS NULL OR evaluation_id = $1) \
+         AND ($2::text IS NULL OR status = $2) \
+         AND ($3::text IS NULL OR system = $3) \
+         AND ($4::text IS NULL OR job_name ILIKE '%' || $4 || '%') \
+         ORDER BY created_at DESC LIMIT $5 OFFSET $6",
+    )
+    .bind(evaluation_id)
+    .bind(status)
+    .bind(system)
+    .bind(job_name)
+    .bind(limit)
+    .bind(offset)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn count_filtered(
+    pool: &PgPool,
+    evaluation_id: Option<Uuid>,
+    status: Option<&str>,
+    system: Option<&str>,
+    job_name: Option<&str>,
+) -> Result<i64> {
+    let row: (i64,) = sqlx::query_as(
+        "SELECT COUNT(*) FROM builds \
+         WHERE ($1::uuid IS NULL OR evaluation_id = $1) \
+         AND ($2::text IS NULL OR status = $2) \
+         AND ($3::text IS NULL OR system = $3) \
+         AND ($4::text IS NULL OR job_name ILIKE '%' || $4 || '%')",
+    )
+    .bind(evaluation_id)
+    .bind(status)
+    .bind(system)
+    .bind(job_name)
+    .fetch_one(pool)
+    .await
+    .map_err(CiError::Database)?;
+    Ok(row.0)
+}
+
+pub async fn cancel(pool: &PgPool, id: Uuid) -> Result<Build> {
+    sqlx::query_as::<_, Build>(
+        "UPDATE builds SET status = 'cancelled', completed_at = NOW() WHERE id = $1 AND status IN ('pending', 'running') RETURNING *",
+    )
+    .bind(id)
+    .fetch_optional(pool)
+    .await?
+    .ok_or_else(|| {
+        CiError::NotFound(format!(
+            "Build {id} not found or not in a cancellable state"
+        ))
+    })
+}
+
+/// Cancel a build and all its transitive dependents.
+pub async fn cancel_cascade(pool: &PgPool, id: Uuid) -> Result<Vec<Build>> {
+    let mut cancelled = Vec::new();
+
+    // Cancel the target build
+    if let Ok(build) = cancel(pool, id).await {
+        cancelled.push(build);
+    }
+
+    // Find and cancel all dependents recursively
+    let mut to_cancel: Vec<Uuid> = vec![id];
+    while let Some(build_id) = to_cancel.pop() {
+        let dependents: Vec<(Uuid,)> = sqlx::query_as(
+            "SELECT build_id FROM build_dependencies WHERE dependency_build_id = $1",
+        )
+        .bind(build_id)
+        .fetch_all(pool)
+        .await
+        .map_err(CiError::Database)?;
+
+        for (dep_id,) in dependents {
+            if let Ok(build) = cancel(pool, dep_id).await {
+                to_cancel.push(dep_id);
+                cancelled.push(build);
+            }
+        }
+    }
+
+    Ok(cancelled)
+}
+
+/// Mark a build's outputs as signed.
+pub async fn mark_signed(pool: &PgPool, id: Uuid) -> Result<()> {
+    sqlx::query("UPDATE builds SET signed = true WHERE id = $1")
+        .bind(id)
+        .execute(pool)
+        .await
+        .map_err(CiError::Database)?;
+    Ok(())
+}
+
+/// Batch-fetch completed builds by derivation paths.
+/// Returns a map from drv_path to Build for deduplication.
+pub async fn get_completed_by_drv_paths(
+    pool: &PgPool,
+    drv_paths: &[String],
+) -> Result<std::collections::HashMap<String, Build>> {
+    if drv_paths.is_empty() {
+        return Ok(std::collections::HashMap::new());
+    }
+    let builds = sqlx::query_as::<_, Build>(
+        "SELECT DISTINCT ON (drv_path) * FROM builds \
+         WHERE drv_path = ANY($1) AND status = 'completed' \
+         ORDER BY drv_path, completed_at DESC",
+    )
+    .bind(drv_paths)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)?;
+
+    Ok(builds
+        .into_iter()
+        .map(|b| (b.drv_path.clone(), b))
+        .collect())
+}
+
+/// Set the builder_id for a build.
+pub async fn set_builder(pool: &PgPool, id: Uuid, builder_id: Uuid) -> Result<()> {
+    sqlx::query("UPDATE builds SET builder_id = $1 WHERE id = $2")
+        .bind(builder_id)
+        .bind(id)
+        .execute(pool)
+        .await
+        .map_err(CiError::Database)?;
+    Ok(())
+}

crates/common/src/repo/channels.rs

@@ -0,0 +1,111 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::{Channel, CreateChannel};
+
+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) RETURNING *",
+    )
+    .bind(input.project_id)
+    .bind(&input.name)
+    .bind(input.jobset_id)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => CiError::Conflict(
+            format!("Channel '{}' already exists for this project", input.name),
+        ),
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn get(pool: &PgPool, id: Uuid) -> Result<Channel> {
+    sqlx::query_as::<_, Channel>("SELECT * FROM channels WHERE id = $1")
+        .bind(id)
+        .fetch_optional(pool)
+        .await?
+        .ok_or_else(|| CiError::NotFound(format!("Channel {id} not found")))
+}
+
+pub async fn list_for_project(pool: &PgPool, project_id: Uuid) -> Result<Vec<Channel>> {
+    sqlx::query_as::<_, Channel>("SELECT * FROM channels WHERE project_id = $1 ORDER BY name")
+        .bind(project_id)
+        .fetch_all(pool)
+        .await
+        .map_err(CiError::Database)
+}
+
+pub async fn list_all(pool: &PgPool) -> Result<Vec<Channel>> {
+    sqlx::query_as::<_, Channel>("SELECT * FROM channels ORDER BY name")
+        .fetch_all(pool)
+        .await
+        .map_err(CiError::Database)
+}
+
+/// Promote an evaluation to a channel (set it as the current evaluation).
+pub async fn promote(pool: &PgPool, channel_id: Uuid, evaluation_id: Uuid) -> Result<Channel> {
+    sqlx::query_as::<_, Channel>(
+        "UPDATE channels SET current_evaluation_id = $1, updated_at = NOW() \
+         WHERE id = $2 RETURNING *",
+    )
+    .bind(evaluation_id)
+    .bind(channel_id)
+    .fetch_optional(pool)
+    .await?
+    .ok_or_else(|| CiError::NotFound(format!("Channel {channel_id} not found")))
+}
+
+pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
+    let result = sqlx::query("DELETE FROM channels WHERE id = $1")
+        .bind(id)
+        .execute(pool)
+        .await
+        .map_err(CiError::Database)?;
+    if result.rows_affected() == 0 {
+        return Err(CiError::NotFound(format!("Channel {id} not found")));
+    }
+    Ok(())
+}
+
+/// Find the channel for a jobset and auto-promote if all builds in the evaluation succeeded.
+pub async fn auto_promote_if_complete(
+    pool: &PgPool,
+    jobset_id: Uuid,
+    evaluation_id: Uuid,
+) -> Result<()> {
+    // Check if all builds for this evaluation are completed
+    let row: (i64, i64) = sqlx::query_as(
+        "SELECT COUNT(*), COUNT(*) FILTER (WHERE status = 'completed') \
+         FROM builds WHERE evaluation_id = $1",
+    )
+    .bind(evaluation_id)
+    .fetch_one(pool)
+    .await
+    .map_err(CiError::Database)?;
+
+    let (total, completed) = row;
+    if total == 0 || total != completed {
+        return Ok(());
+    }
+
+    // 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)
+        .fetch_all(pool)
+        .await
+        .map_err(CiError::Database)?;
+
+    for channel in channels {
+        let _ = promote(pool, channel.id, evaluation_id).await;
+        tracing::info!(
+            channel = %channel.name,
+            evaluation_id = %evaluation_id,
+            "Auto-promoted evaluation to channel"
+        );
+    }
+
+    Ok(())
+}

crates/common/src/repo/evaluations.rs

@@ -0,0 +1,146 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::{CreateEvaluation, Evaluation, EvaluationStatus};
+
+pub async fn create(pool: &PgPool, input: CreateEvaluation) -> Result<Evaluation> {
+    sqlx::query_as::<_, Evaluation>(
+        "INSERT INTO evaluations (jobset_id, commit_hash, status) VALUES ($1, $2, 'pending') RETURNING *",
+    )
+    .bind(input.jobset_id)
+    .bind(&input.commit_hash)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!(
+                "Evaluation for commit '{}' already exists in this jobset",
+                input.commit_hash
+            ))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn get(pool: &PgPool, id: Uuid) -> Result<Evaluation> {
+    sqlx::query_as::<_, Evaluation>("SELECT * FROM evaluations WHERE id = $1")
+        .bind(id)
+        .fetch_optional(pool)
+        .await?
+        .ok_or_else(|| CiError::NotFound(format!("Evaluation {id} not found")))
+}
+
+pub async fn list_for_jobset(pool: &PgPool, jobset_id: Uuid) -> Result<Vec<Evaluation>> {
+    sqlx::query_as::<_, Evaluation>(
+        "SELECT * FROM evaluations WHERE jobset_id = $1 ORDER BY evaluation_time DESC",
+    )
+    .bind(jobset_id)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+/// List evaluations with optional jobset_id and status filters, with pagination.
+pub async fn list_filtered(
+    pool: &PgPool,
+    jobset_id: Option<Uuid>,
+    status: Option<&str>,
+    limit: i64,
+    offset: i64,
+) -> Result<Vec<Evaluation>> {
+    sqlx::query_as::<_, Evaluation>(
+        "SELECT * FROM evaluations \
+         WHERE ($1::uuid IS NULL OR jobset_id = $1) \
+         AND ($2::text IS NULL OR status = $2) \
+         ORDER BY evaluation_time DESC LIMIT $3 OFFSET $4",
+    )
+    .bind(jobset_id)
+    .bind(status)
+    .bind(limit)
+    .bind(offset)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn count_filtered(
+    pool: &PgPool,
+    jobset_id: Option<Uuid>,
+    status: Option<&str>,
+) -> Result<i64> {
+    let row: (i64,) = sqlx::query_as(
+        "SELECT COUNT(*) FROM evaluations \
+         WHERE ($1::uuid IS NULL OR jobset_id = $1) \
+         AND ($2::text IS NULL OR status = $2)",
+    )
+    .bind(jobset_id)
+    .bind(status)
+    .fetch_one(pool)
+    .await
+    .map_err(CiError::Database)?;
+    Ok(row.0)
+}
+
+pub async fn update_status(
+    pool: &PgPool,
+    id: Uuid,
+    status: EvaluationStatus,
+    error_message: Option<&str>,
+) -> Result<Evaluation> {
+    sqlx::query_as::<_, Evaluation>(
+        "UPDATE evaluations SET status = $1, error_message = $2 WHERE id = $3 RETURNING *",
+    )
+    .bind(status)
+    .bind(error_message)
+    .bind(id)
+    .fetch_optional(pool)
+    .await?
+    .ok_or_else(|| CiError::NotFound(format!("Evaluation {id} not found")))
+}
+
+pub async fn get_latest(pool: &PgPool, jobset_id: Uuid) -> Result<Option<Evaluation>> {
+    sqlx::query_as::<_, Evaluation>(
+        "SELECT * FROM evaluations WHERE jobset_id = $1 ORDER BY evaluation_time DESC LIMIT 1",
+    )
+    .bind(jobset_id)
+    .fetch_optional(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+/// Set the inputs hash for an evaluation (used for eval caching).
+pub async fn set_inputs_hash(pool: &PgPool, id: Uuid, hash: &str) -> Result<()> {
+    sqlx::query("UPDATE evaluations SET inputs_hash = $1 WHERE id = $2")
+        .bind(hash)
+        .bind(id)
+        .execute(pool)
+        .await
+        .map_err(CiError::Database)?;
+    Ok(())
+}
+
+/// Check if an evaluation with the same inputs_hash already exists for this jobset.
+pub async fn get_by_inputs_hash(
+    pool: &PgPool,
+    jobset_id: Uuid,
+    inputs_hash: &str,
+) -> Result<Option<Evaluation>> {
+    sqlx::query_as::<_, Evaluation>(
+        "SELECT * FROM evaluations WHERE jobset_id = $1 AND inputs_hash = $2 \
+         AND status = 'completed' ORDER BY evaluation_time DESC LIMIT 1",
+    )
+    .bind(jobset_id)
+    .bind(inputs_hash)
+    .fetch_optional(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn count(pool: &PgPool) -> Result<i64> {
+    let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM evaluations")
+        .fetch_one(pool)
+        .await
+        .map_err(CiError::Database)?;
+    Ok(row.0)
+}

crates/common/src/repo/jobset_inputs.rs

@@ -0,0 +1,52 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::JobsetInput;
+
+pub async fn create(
+    pool: &PgPool,
+    jobset_id: Uuid,
+    name: &str,
+    input_type: &str,
+    value: &str,
+    revision: Option<&str>,
+) -> Result<JobsetInput> {
+    sqlx::query_as::<_, JobsetInput>(
+        "INSERT INTO jobset_inputs (jobset_id, name, input_type, value, revision) VALUES ($1, $2, $3, $4, $5) RETURNING *",
+    )
+    .bind(jobset_id)
+    .bind(name)
+    .bind(input_type)
+    .bind(value)
+    .bind(revision)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!("Input '{name}' already exists in this jobset"))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn list_for_jobset(pool: &PgPool, jobset_id: Uuid) -> Result<Vec<JobsetInput>> {
+    sqlx::query_as::<_, JobsetInput>(
+        "SELECT * FROM jobset_inputs WHERE jobset_id = $1 ORDER BY name ASC",
+    )
+    .bind(jobset_id)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
+    let result = sqlx::query("DELETE FROM jobset_inputs WHERE id = $1")
+        .bind(id)
+        .execute(pool)
+        .await?;
+    if result.rows_affected() == 0 {
+        return Err(CiError::NotFound(format!("Jobset input {id} not found")));
+    }
+    Ok(())
+}

crates/common/src/repo/jobsets.rs

@@ -0,0 +1,137 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::{ActiveJobset, CreateJobset, Jobset, UpdateJobset};
+
+pub async fn create(pool: &PgPool, input: CreateJobset) -> Result<Jobset> {
+    let enabled = input.enabled.unwrap_or(true);
+    let flake_mode = input.flake_mode.unwrap_or(true);
+    let check_interval = input.check_interval.unwrap_or(60);
+
+    sqlx::query_as::<_, Jobset>(
+        "INSERT INTO jobsets (project_id, name, nix_expression, enabled, flake_mode, check_interval) VALUES ($1, $2, $3, $4, $5, $6) RETURNING *",
+    )
+    .bind(input.project_id)
+    .bind(&input.name)
+    .bind(&input.nix_expression)
+    .bind(enabled)
+    .bind(flake_mode)
+    .bind(check_interval)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!("Jobset '{}' already exists in this project", input.name))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn get(pool: &PgPool, id: Uuid) -> Result<Jobset> {
+    sqlx::query_as::<_, Jobset>("SELECT * FROM jobsets WHERE id = $1")
+        .bind(id)
+        .fetch_optional(pool)
+        .await?
+        .ok_or_else(|| CiError::NotFound(format!("Jobset {id} not found")))
+}
+
+pub async fn list_for_project(
+    pool: &PgPool,
+    project_id: Uuid,
+    limit: i64,
+    offset: i64,
+) -> Result<Vec<Jobset>> {
+    sqlx::query_as::<_, Jobset>(
+        "SELECT * FROM jobsets WHERE project_id = $1 ORDER BY created_at DESC LIMIT $2 OFFSET $3",
+    )
+    .bind(project_id)
+    .bind(limit)
+    .bind(offset)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+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")
+        .bind(project_id)
+        .fetch_one(pool)
+        .await
+        .map_err(CiError::Database)?;
+    Ok(row.0)
+}
+
+pub async fn update(pool: &PgPool, id: Uuid, input: UpdateJobset) -> Result<Jobset> {
+    let existing = get(pool, id).await?;
+
+    let name = input.name.unwrap_or(existing.name);
+    let nix_expression = input.nix_expression.unwrap_or(existing.nix_expression);
+    let enabled = input.enabled.unwrap_or(existing.enabled);
+    let flake_mode = input.flake_mode.unwrap_or(existing.flake_mode);
+    let check_interval = input.check_interval.unwrap_or(existing.check_interval);
+
+    sqlx::query_as::<_, Jobset>(
+        "UPDATE jobsets SET name = $1, nix_expression = $2, enabled = $3, flake_mode = $4, check_interval = $5 WHERE id = $6 RETURNING *",
+    )
+    .bind(&name)
+    .bind(&nix_expression)
+    .bind(enabled)
+    .bind(flake_mode)
+    .bind(check_interval)
+    .bind(id)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!("Jobset '{name}' already exists in this project"))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
+    let result = sqlx::query("DELETE FROM jobsets WHERE id = $1")
+        .bind(id)
+        .execute(pool)
+        .await?;
+
+    if result.rows_affected() == 0 {
+        return Err(CiError::NotFound(format!("Jobset {id} not found")));
+    }
+
+    Ok(())
+}
+
+pub async fn upsert(pool: &PgPool, input: CreateJobset) -> Result<Jobset> {
+    let enabled = input.enabled.unwrap_or(true);
+    let flake_mode = input.flake_mode.unwrap_or(true);
+    let check_interval = input.check_interval.unwrap_or(60);
+
+    sqlx::query_as::<_, Jobset>(
+        "INSERT INTO jobsets (project_id, name, nix_expression, enabled, flake_mode, check_interval) \
+         VALUES ($1, $2, $3, $4, $5, $6) \
+         ON CONFLICT (project_id, name) DO UPDATE SET \
+         nix_expression = EXCLUDED.nix_expression, \
+         enabled = EXCLUDED.enabled, \
+         flake_mode = EXCLUDED.flake_mode, \
+         check_interval = EXCLUDED.check_interval \
+         RETURNING *",
+    )
+    .bind(input.project_id)
+    .bind(&input.name)
+    .bind(&input.nix_expression)
+    .bind(enabled)
+    .bind(flake_mode)
+    .bind(check_interval)
+    .fetch_one(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn list_active(pool: &PgPool) -> Result<Vec<ActiveJobset>> {
+    sqlx::query_as::<_, ActiveJobset>("SELECT * FROM active_jobsets")
+        .fetch_all(pool)
+        .await
+        .map_err(CiError::Database)
+}

crates/common/src/repo/mod.rs

@@ -0,0 +1,13 @@
+pub mod api_keys;
+pub mod build_dependencies;
+pub mod build_products;
+pub mod build_steps;
+pub mod builds;
+pub mod channels;
+pub mod evaluations;
+pub mod jobset_inputs;
+pub mod jobsets;
+pub mod notification_configs;
+pub mod projects;
+pub mod remote_builders;
+pub mod webhook_configs;

crates/common/src/repo/notification_configs.rs

@@ -0,0 +1,48 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::{CreateNotificationConfig, NotificationConfig};
+
+pub async fn create(pool: &PgPool, input: CreateNotificationConfig) -> Result<NotificationConfig> {
+    sqlx::query_as::<_, NotificationConfig>(
+        "INSERT INTO notification_configs (project_id, notification_type, config) VALUES ($1, $2, $3) RETURNING *",
+    )
+    .bind(input.project_id)
+    .bind(&input.notification_type)
+    .bind(&input.config)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!(
+                "Notification config '{}' already exists for this project",
+                input.notification_type
+            ))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn list_for_project(pool: &PgPool, project_id: Uuid) -> Result<Vec<NotificationConfig>> {
+    sqlx::query_as::<_, NotificationConfig>(
+        "SELECT * FROM notification_configs WHERE project_id = $1 AND enabled = true ORDER BY created_at DESC",
+    )
+    .bind(project_id)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
+    let result = sqlx::query("DELETE FROM notification_configs WHERE id = $1")
+        .bind(id)
+        .execute(pool)
+        .await?;
+    if result.rows_affected() == 0 {
+        return Err(CiError::NotFound(format!(
+            "Notification config {id} not found"
+        )));
+    }
+    Ok(())
+}

crates/common/src/repo/projects.rs

@@ -0,0 +1,95 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::{CreateProject, Project, UpdateProject};
+
+pub async fn create(pool: &PgPool, input: CreateProject) -> Result<Project> {
+    sqlx::query_as::<_, Project>(
+        "INSERT INTO projects (name, description, repository_url) VALUES ($1, $2, $3) RETURNING *",
+    )
+    .bind(&input.name)
+    .bind(&input.description)
+    .bind(&input.repository_url)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!("Project '{}' already exists", input.name))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn get(pool: &PgPool, id: Uuid) -> Result<Project> {
+    sqlx::query_as::<_, Project>("SELECT * FROM projects WHERE id = $1")
+        .bind(id)
+        .fetch_optional(pool)
+        .await?
+        .ok_or_else(|| CiError::NotFound(format!("Project {id} 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)
+        .fetch_optional(pool)
+        .await?
+        .ok_or_else(|| CiError::NotFound(format!("Project '{name}' not found")))
+}
+
+pub async fn list(pool: &PgPool, limit: i64, offset: i64) -> Result<Vec<Project>> {
+    sqlx::query_as::<_, Project>(
+        "SELECT * FROM projects ORDER BY created_at DESC LIMIT $1 OFFSET $2",
+    )
+    .bind(limit)
+    .bind(offset)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn count(pool: &PgPool) -> Result<i64> {
+    let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM projects")
+        .fetch_one(pool)
+        .await
+        .map_err(CiError::Database)?;
+    Ok(row.0)
+}
+
+pub async fn update(pool: &PgPool, id: Uuid, input: UpdateProject) -> Result<Project> {
+    // Build dynamic update — only set provided fields
+    let existing = get(pool, id).await?;
+
+    let name = input.name.unwrap_or(existing.name);
+    let description = input.description.or(existing.description);
+    let repository_url = input.repository_url.unwrap_or(existing.repository_url);
+
+    sqlx::query_as::<_, Project>(
+        "UPDATE projects SET name = $1, description = $2, repository_url = $3 WHERE id = $4 RETURNING *",
+    )
+    .bind(&name)
+    .bind(&description)
+    .bind(&repository_url)
+    .bind(id)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!("Project '{name}' already exists"))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
+    let result = sqlx::query("DELETE FROM projects WHERE id = $1")
+        .bind(id)
+        .execute(pool)
+        .await?;
+
+    if result.rows_affected() == 0 {
+        return Err(CiError::NotFound(format!("Project {id} not found")));
+    }
+
+    Ok(())
+}

crates/common/src/repo/remote_builders.rs

@@ -0,0 +1,124 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::{CreateRemoteBuilder, RemoteBuilder};
+
+pub async fn create(pool: &PgPool, input: CreateRemoteBuilder) -> Result<RemoteBuilder> {
+    sqlx::query_as::<_, RemoteBuilder>(
+        "INSERT INTO remote_builders (name, ssh_uri, systems, max_jobs, speed_factor, \
+         supported_features, mandatory_features, public_host_key, ssh_key_file) \
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) RETURNING *",
+    )
+    .bind(&input.name)
+    .bind(&input.ssh_uri)
+    .bind(&input.systems)
+    .bind(input.max_jobs.unwrap_or(1))
+    .bind(input.speed_factor.unwrap_or(1))
+    .bind(input.supported_features.as_deref().unwrap_or(&[]))
+    .bind(input.mandatory_features.as_deref().unwrap_or(&[]))
+    .bind(&input.public_host_key)
+    .bind(&input.ssh_key_file)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!("Remote builder '{}' already exists", input.name))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn get(pool: &PgPool, id: Uuid) -> Result<RemoteBuilder> {
+    sqlx::query_as::<_, RemoteBuilder>("SELECT * FROM remote_builders WHERE id = $1")
+        .bind(id)
+        .fetch_optional(pool)
+        .await?
+        .ok_or_else(|| CiError::NotFound(format!("Remote builder {id} not found")))
+}
+
+pub async fn list(pool: &PgPool) -> Result<Vec<RemoteBuilder>> {
+    sqlx::query_as::<_, RemoteBuilder>(
+        "SELECT * FROM remote_builders ORDER BY speed_factor DESC, name",
+    )
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+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 DESC, name",
+    )
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+/// Find a suitable builder for the given system.
+pub async fn find_for_system(pool: &PgPool, system: &str) -> Result<Vec<RemoteBuilder>> {
+    sqlx::query_as::<_, RemoteBuilder>(
+        "SELECT * FROM remote_builders WHERE enabled = true AND $1 = ANY(systems) \
+         ORDER BY speed_factor DESC",
+    )
+    .bind(system)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn update(
+    pool: &PgPool,
+    id: Uuid,
+    input: crate::models::UpdateRemoteBuilder,
+) -> Result<RemoteBuilder> {
+    // Build dynamic update — use 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 = COALESCE($4, max_jobs), \
+         speed_factor = COALESCE($5, speed_factor), \
+         supported_features = COALESCE($6, supported_features), \
+         mandatory_features = COALESCE($7, mandatory_features), \
+         enabled = COALESCE($8, enabled), \
+         public_host_key = COALESCE($9, public_host_key), \
+         ssh_key_file = COALESCE($10, ssh_key_file) \
+         WHERE id = $11 RETURNING *",
+    )
+    .bind(&input.name)
+    .bind(&input.ssh_uri)
+    .bind(&input.systems)
+    .bind(input.max_jobs)
+    .bind(input.speed_factor)
+    .bind(&input.supported_features)
+    .bind(&input.mandatory_features)
+    .bind(input.enabled)
+    .bind(&input.public_host_key)
+    .bind(&input.ssh_key_file)
+    .bind(id)
+    .fetch_optional(pool)
+    .await?
+    .ok_or_else(|| CiError::NotFound(format!("Remote builder {id} not found")))
+}
+
+pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
+    let result = sqlx::query("DELETE FROM remote_builders WHERE id = $1")
+        .bind(id)
+        .execute(pool)
+        .await
+        .map_err(CiError::Database)?;
+    if result.rows_affected() == 0 {
+        return Err(CiError::NotFound(format!("Remote builder {id} not found")));
+    }
+    Ok(())
+}
+
+pub async fn count(pool: &PgPool) -> Result<i64> {
+    let row: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM remote_builders")
+        .fetch_one(pool)
+        .await
+        .map_err(CiError::Database)?;
+    Ok(row.0)
+}

crates/common/src/repo/webhook_configs.rs

@@ -0,0 +1,73 @@
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::error::{CiError, Result};
+use crate::models::{CreateWebhookConfig, WebhookConfig};
+
+pub async fn create(
+    pool: &PgPool,
+    input: CreateWebhookConfig,
+    secret_hash: Option<&str>,
+) -> Result<WebhookConfig> {
+    sqlx::query_as::<_, WebhookConfig>(
+        "INSERT INTO webhook_configs (project_id, forge_type, secret_hash) VALUES ($1, $2, $3) RETURNING *",
+    )
+    .bind(input.project_id)
+    .bind(&input.forge_type)
+    .bind(secret_hash)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| match &e {
+        sqlx::Error::Database(db_err) if db_err.is_unique_violation() => {
+            CiError::Conflict(format!(
+                "Webhook config for forge '{}' already exists for this project",
+                input.forge_type
+            ))
+        }
+        _ => CiError::Database(e),
+    })
+}
+
+pub async fn get(pool: &PgPool, id: Uuid) -> Result<WebhookConfig> {
+    sqlx::query_as::<_, WebhookConfig>("SELECT * FROM webhook_configs WHERE id = $1")
+        .bind(id)
+        .fetch_optional(pool)
+        .await?
+        .ok_or_else(|| CiError::NotFound(format!("Webhook config {id} not found")))
+}
+
+pub async fn list_for_project(pool: &PgPool, project_id: Uuid) -> Result<Vec<WebhookConfig>> {
+    sqlx::query_as::<_, WebhookConfig>(
+        "SELECT * FROM webhook_configs WHERE project_id = $1 ORDER BY created_at DESC",
+    )
+    .bind(project_id)
+    .fetch_all(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn get_by_project_and_forge(
+    pool: &PgPool,
+    project_id: Uuid,
+    forge_type: &str,
+) -> Result<Option<WebhookConfig>> {
+    sqlx::query_as::<_, WebhookConfig>(
+        "SELECT * FROM webhook_configs WHERE project_id = $1 AND forge_type = $2 AND enabled = true",
+    )
+    .bind(project_id)
+    .bind(forge_type)
+    .fetch_optional(pool)
+    .await
+    .map_err(CiError::Database)
+}
+
+pub async fn delete(pool: &PgPool, id: Uuid) -> Result<()> {
+    let result = sqlx::query("DELETE FROM webhook_configs WHERE id = $1")
+        .bind(id)
+        .execute(pool)
+        .await?;
+    if result.rows_affected() == 0 {
+        return Err(CiError::NotFound(format!("Webhook config {id} not found")));
+    }
+    Ok(())
+}