Menu
AkurAI-Build
publicLatest change 7255093e06e11217a8aa00ce4b6f336d3a5fa654 - Add named service identities and akurai_repo_protect by Ólafur Búi Ólafsson
use std::{
env,
fs::{self, OpenOptions},
io::{Read, Write},
path::{Path, PathBuf},
sync::{Arc, Mutex, MutexGuard},
time::Duration,
};
use crate::config;
use anyhow::{Context, Result, anyhow, bail, ensure};
use rusqlite::{
Connection, OpenFlags, OptionalExtension, TransactionBehavior, config::DbConfig, limits::Limit,
params, params_from_iter, types::Value,
};
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;
use sha2::{Digest, Sha256};
use zeroize::Zeroizing;
const APPLICATION_ID: i64 = 0x414B_5552;
const MIGRATIONS: &[(i64, &str, &str)] = &[
(
1,
"initial CI schema",
include_str!("../migrations/001_init.sql"),
),
(
2,
"repository visibility",
include_str!("../migrations/002_repository_visibility.sql"),
),
(
3,
"build workers",
include_str!("../migrations/003_workers.sql"),
),
(
4,
"delivery provenance",
include_str!("../migrations/004_delivery_provenance.sql"),
),
(
5,
"browser sessions",
include_str!("../migrations/005_browser_sessions.sql"),
),
(
6,
"profile evidence",
include_str!("../migrations/006_profile_evidence.sql"),
),
(
7,
"repository access",
include_str!("../migrations/007_repository_access.sql"),
),
(
8,
"community issues and comments",
include_str!("../migrations/008_community.sql"),
),
(
9,
"worker governance",
include_str!("../migrations/009_worker_governance.sql"),
),
(
10,
"operations audit",
include_str!("../migrations/010_operations_audit.sql"),
),
(
11,
"pull requests and merge queue",
include_str!("../migrations/011_pull_requests.sql"),
),
(
12,
"branch protection",
include_str!("../migrations/012_branch_protection.sql"),
),
(
13,
"merge queue re-enqueue after failure",
include_str!("../migrations/013_merge_queue_reenqueue.sql"),
),
];
const LEDGER: &str = "CREATE TABLE IF NOT EXISTS _migrations (
version INTEGER PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
sql TEXT NOT NULL,
applied_at INTEGER NOT NULL DEFAULT (unixepoch())
) STRICT;";
#[derive(Clone)]
pub struct Database(Arc<Mutex<Connection>>);
/// Server-side leg of an in-flight authorization-code flow.
#[derive(Clone, Debug)]
pub struct LoginState {
pub state: String,
pub nonce: String,
pub code_verifier: String,
pub return_to: String,
}
/// Durable browser session established by AkurAI ID.
#[derive(Clone, Debug)]
pub struct Session {
pub id: String,
pub sub: String,
pub email: String,
pub role: String,
pub organization_id: Option<String>,
pub tenant_id: Option<String>,
pub groups: Vec<String>,
pub id_token: String,
pub created_at: i64,
pub expires_at: i64,
}
impl Session {
/// `owner` and `admin` are platform superusers with full rights everywhere.
pub fn is_superuser(&self) -> bool {
matches!(self.role.as_str(), "owner" | "admin")
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum RepositoryRole {
Read,
Write,
Maintain,
}
impl RepositoryRole {
pub fn allows(self, required: Self) -> bool {
self >= required
}
}
impl TryFrom<&str> for RepositoryRole {
type Error = anyhow::Error;
fn try_from(value: &str) -> Result<Self> {
match value {
"read" => Ok(Self::Read),
"write" => Ok(Self::Write),
"maintain" => Ok(Self::Maintain),
_ => bail!("invalid repository role {value}"),
}
}
}
impl RepositoryRole {
pub fn as_str(self) -> &'static str {
match self {
Self::Read => "read",
Self::Write => "write",
Self::Maintain => "maintain",
}
}
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct RepositoryAccess {
pub subject: String,
pub organization_id: Option<String>,
pub role: RepositoryRole,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Repository {
pub id: i64,
pub name: String,
pub url: String,
pub default_branch: String,
pub visibility: String,
pub owner_sub: Option<String>,
pub owner_organization_id: Option<String>,
pub trusted: bool,
pub protect_default_branch: bool,
pub created_at: i64,
}
#[derive(Clone, Debug, Serialize)]
pub struct PublicRepository {
pub id: i64,
pub name: String,
pub url: String,
pub default_branch: String,
pub created_at: i64,
pub run_count: i64,
pub successful_runs: i64,
pub last_activity_at: i64,
}
#[derive(Clone, Debug)]
pub struct RunQuery {
pub repository: Option<String>,
pub statuses: Vec<String>,
pub git_ref: Option<String>,
pub triggers: Vec<String>,
pub search: Option<String>,
pub before_id: Option<i64>,
pub limit: usize,
pub offset: usize,
pub public_only: bool,
}
impl Default for RunQuery {
fn default() -> Self {
Self {
repository: None,
statuses: Vec::new(),
git_ref: None,
triggers: Vec::new(),
search: None,
before_id: None,
limit: 20,
offset: 0,
public_only: false,
}
}
}
struct ValidatedRunQuery {
predicate: String,
values: Vec<Value>,
}
fn validate_run_query(query: &RunQuery) -> Result<ValidatedRunQuery> {
ensure!(
(1..=200).contains(&query.limit),
"run limit must be 1..=200"
);
ensure!(query.offset <= 10_000, "run offset must be <= 10000");
if let Some(before_id) = query.before_id {
ensure!(before_id > 0, "run cursor must be positive");
}
let repositories = query
.repository
.as_deref()
.map(|value| {
value
.split(',')
.map(str::trim)
.filter(|item| !item.is_empty())
.map(str::to_owned)
.collect::<Vec<_>>()
})
.unwrap_or_default();
ensure!(repositories.len() <= 32, "too many repository filters");
for repository in &repositories {
ensure!(
repository.len() <= 64
&& repository
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || b"._-".contains(&byte)),
"invalid repository filter: {repository}"
);
}
if let Some(git_ref) = &query.git_ref {
crate::config::validate_ref(git_ref)?;
}
for status in &query.statuses {
ensure!(
matches!(
status.as_str(),
"queued"
| "running"
| "waiting"
| "succeeded"
| "failed"
| "canceled"
| "interrupted"
),
"invalid run status: {status}"
);
}
for trigger in &query.triggers {
ensure!(
matches!(trigger.as_str(), "manual" | "webhook" | "retry"),
"invalid run trigger: {trigger}"
);
}
let mut clauses = Vec::new();
let mut values = Vec::new();
if !repositories.is_empty() {
let placeholders = std::iter::repeat_n("?", repositories.len())
.collect::<Vec<_>>()
.join(", ");
clauses.push(format!("p.name IN ({placeholders})"));
values.extend(repositories.into_iter().map(Value::Text));
}
if query.public_only {
clauses.push("p.visibility = 'public'".to_owned());
}
if let Some(before_id) = query.before_id {
clauses.push("r.id <= ?".to_owned());
values.push(Value::Integer(before_id));
}
if let Some(git_ref) = &query.git_ref {
clauses.push("r.git_ref = ?".to_owned());
values.push(Value::Text(git_ref.clone()));
}
if !query.statuses.is_empty() {
let placeholders = std::iter::repeat_n("?", query.statuses.len())
.collect::<Vec<_>>()
.join(", ");
clauses.push(format!("r.status IN ({placeholders})"));
values.extend(query.statuses.iter().cloned().map(Value::Text));
}
if !query.triggers.is_empty() {
let placeholders = std::iter::repeat_n("?", query.triggers.len())
.collect::<Vec<_>>()
.join(", ");
clauses.push(format!("r.trigger IN ({placeholders})"));
values.extend(query.triggers.iter().cloned().map(Value::Text));
}
if let Some(search) = &query.search {
ensure!(
!search.is_empty() && search.len() <= 200,
"search must be 1..=200 bytes"
);
let needle = Value::Text(format!("%{search}%"));
clauses.push(
"(p.name LIKE ? OR r.git_ref LIKE ? OR COALESCE(r.commit_sha, '') LIKE ?
OR COALESCE(r.error, '') LIKE ?)"
.to_owned(),
);
values.extend(std::iter::repeat_n(needle, 4));
}
let predicate = if clauses.is_empty() {
" WHERE 1=1".to_owned()
} else {
format!(" WHERE {}", clauses.join(" AND "))
};
Ok(ValidatedRunQuery { predicate, values })
}
#[derive(Clone, Debug)]
pub struct RepositoryQuery {
pub search: Option<String>,
pub visibility: Option<String>,
pub limit: usize,
pub offset: usize,
}
impl Default for RepositoryQuery {
fn default() -> Self {
Self {
search: None,
visibility: None,
limit: 50,
offset: 0,
}
}
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Worker {
pub id: String,
pub host: String,
pub capabilities: String,
pub status: String,
pub capacity: i64,
pub current_run_id: Option<i64>,
pub started_at: i64,
pub heartbeat_at: i64,
pub completed_runs: i64,
pub draining: bool,
pub lease_expires_at: Option<i64>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct WorkerLease {
pub acquired: bool,
pub previous_run_id: Option<i64>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Run {
pub id: i64,
pub repository_id: i64,
pub repository: String,
pub git_ref: String,
pub commit_sha: Option<String>,
pub trigger: String,
pub status: String,
pub error: Option<String>,
pub created_at: i64,
pub started_at: Option<i64>,
pub finished_at: Option<i64>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Job {
pub id: i64,
pub run_id: i64,
pub base_name: String,
pub name: String,
pub executor: String,
pub image: Option<String>,
pub platform: Option<String>,
pub status: String,
pub needs: Vec<String>,
pub spec_json: String,
pub logs: String,
pub exit_code: Option<i64>,
pub environment: Option<String>,
pub approval_required: bool,
pub approved_at: Option<i64>,
pub approved_by: Option<String>,
pub started_at: Option<i64>,
pub finished_at: Option<i64>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct JobStageMetric {
pub job_id: i64,
pub run_id: i64,
pub name: String,
pub status: String,
pub started_at: Option<i64>,
pub finished_at: Option<i64>,
pub duration_seconds: Option<f64>,
pub cache_paths: Vec<String>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct AuditEvent {
pub id: i64,
pub actor: String,
pub operation: String,
pub repository: Option<String>,
pub environment: Option<String>,
pub target: Option<String>,
pub outcome: String,
pub details_json: String,
pub created_at: i64,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Artifact {
pub id: i64,
pub run_id: i64,
pub job_id: i64,
pub name: String,
pub path: String,
pub sha256: String,
pub bytes: i64,
pub created_at: i64,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Deployment {
pub id: i64,
pub run_id: i64,
pub job_id: i64,
pub environment: String,
pub status: String,
pub artifacts_json: String,
pub created_at: i64,
pub finished_at: Option<i64>,
pub organization_id: Option<String>,
pub workspace_id: Option<String>,
pub installation_id: Option<String>,
pub app_id: Option<String>,
pub installation_revision: Option<i64>,
pub commit_sha: Option<String>,
pub artifact_digest: Option<String>,
pub result: Option<String>,
pub rollback_of_deployment_id: Option<i64>,
pub recovery_of_deployment_id: Option<i64>,
}
pub struct ManagedDeployment<'a> {
pub run_id: i64,
pub job_id: i64,
pub environment: &'a str,
pub artifacts_json: &'a str,
pub installation_id: &'a str,
pub installation_revision: i64,
pub commit_sha: Option<&'a str>,
pub artifact_digest: Option<&'a str>,
pub result: Option<&'a str>,
pub rollback_of_deployment_id: Option<i64>,
pub recovery_of_deployment_id: Option<i64>,
}
const MAX_METRIC_ROWS: usize = 10_000;
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct DeliveryMetrics {
pub repository: Option<String>,
pub environment: Option<String>,
pub window_seconds: i64,
pub since: i64,
pub fresh_at: i64,
pub sample_size: i64,
pub deployment_frequency_per_day: Option<f64>,
pub commit_to_production_seconds: Option<f64>,
pub change_failure_rate: Option<f64>,
pub recovery_time_seconds: Option<f64>,
pub queue_duration_seconds: Option<f64>,
pub job_stages: Vec<JobStageMetric>,
pub deployment_ids: Vec<i64>,
pub run_ids: Vec<i64>,
pub truncated: bool,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct RunDetail {
#[serde(flatten)]
pub run: Run,
pub jobs: Vec<Job>,
pub artifacts: Vec<Artifact>,
pub deployments: Vec<Deployment>,
}
impl Database {
pub fn open(path: &Path, key: &str, create: bool) -> Result<Self> {
validate_secret(key)?;
prepare_database(path, create)?;
let flags = if create {
OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE
} else {
OpenFlags::SQLITE_OPEN_READ_WRITE
} | OpenFlags::SQLITE_OPEN_NO_MUTEX;
let connection = Connection::open_with_flags(path, flags)
.with_context(|| format!("open database {}", path.display()))?;
secure_database_file(path)?;
configure(&connection, key)?;
Ok(Self(Arc::new(Mutex::new(connection))))
}
#[cfg(test)]
pub fn memory(key: &str) -> Result<Self> {
validate_secret(key)?;
let connection = Connection::open_in_memory()?;
configure(&connection, key)?;
let database = Self(Arc::new(Mutex::new(connection)));
database.migrate()?;
Ok(database)
}
pub fn migrate(&self) -> Result<usize> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
transaction.execute_batch(LEDGER)?;
let mut applied = 0;
for &(version, name, sql) in MIGRATIONS {
let existing: Option<(String, String)> = transaction
.query_row(
"SELECT name, sql FROM _migrations WHERE version = ?1",
[version],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.optional()?;
if let Some((existing_name, existing_sql)) = existing {
ensure!(
existing_name == name && existing_sql == sql,
"migration history differs from this binary"
);
continue;
}
transaction.execute_batch(sql)?;
transaction.execute(
"INSERT INTO _migrations(version, name, sql) VALUES (?1, ?2, ?3)",
params![version, name, sql],
)?;
applied += 1;
}
transaction.commit()?;
drop(connection);
self.validate_schema()?;
Ok(applied)
}
pub fn validate_schema(&self) -> Result<()> {
let connection = self.connection()?;
let application_id: i64 =
connection.pragma_query_value(None, "application_id", |row| row.get(0))?;
let version: i64 = connection.pragma_query_value(None, "user_version", |row| row.get(0))?;
ensure!(
application_id == APPLICATION_ID && version == 13,
"database schema is not AkurAI Build v13"
);
for table in [
"repositories",
"runs",
"jobs",
"artifacts",
"deployments",
"workers",
"managed_installations",
"installation_revisions",
"oauth_states",
"profile_baselines",
"repository_collaborators",
"issues",
"issue_comments",
"audit_events",
"pull_requests",
"pull_request_reviews",
"merge_queue_entries",
"_migrations",
] {
let exists: bool = connection.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type='table' AND name=?1)",
[table],
|row| row.get(0),
)?;
ensure!(exists, "database is missing table {table}");
}
Ok(())
}
pub fn check_ready(&self) -> Result<()> {
self.connection()?.query_row("SELECT 1", [], |_| Ok(()))?;
Ok(())
}
/// Persist the single-use server-side leg of an authorization-code flow.
pub fn create_login_state(&self, login: &LoginState, ttl_seconds: i64) -> Result<()> {
let connection = self.connection()?;
connection.execute(
"DELETE FROM oauth_states WHERE expires_at <= unixepoch()",
[],
)?;
connection.execute(
"INSERT INTO oauth_states(state, nonce, code_verifier, return_to, expires_at)
VALUES (?1, ?2, ?3, ?4, unixepoch() + ?5)",
params![
login.state,
login.nonce,
login.code_verifier,
login.return_to,
ttl_seconds
],
)?;
Ok(())
}
/// Consume a login state: returns it only when present and unexpired, and
/// always removes the row so a code may never be replayed.
pub fn take_login_state(&self, state: &str) -> Result<Option<LoginState>> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let login = transaction
.query_row(
"DELETE FROM oauth_states WHERE state = ?1 AND expires_at > unixepoch()
RETURNING state, nonce, code_verifier, return_to",
[state],
|row| {
Ok(LoginState {
state: row.get(0)?,
nonce: row.get(1)?,
code_verifier: row.get(2)?,
return_to: row.get(3)?,
})
},
)
.optional()?;
transaction.execute("DELETE FROM oauth_states WHERE state = ?1", [state])?;
transaction.commit()?;
Ok(login)
}
/// Store a browser session record keyed by the opaque cookie value.
pub fn create_session(&self, session: &Session) -> Result<()> {
let connection = self.connection()?;
connection.execute("DELETE FROM sessions WHERE expires_at <= unixepoch()", [])?;
connection.execute(
"INSERT INTO sessions(
id, sub, email, role, organization_id, tenant_id, groups, id_token,
created_at, expires_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
params![
session.id,
session.sub,
session.email,
session.role,
session.organization_id,
session.tenant_id,
serde_json::to_string(&session.groups)?,
session.id_token,
session.created_at,
session.expires_at,
],
)?;
Ok(())
}
/// Look up a live session. Expired rows are deleted and never returned.
pub fn session(&self, id: &str) -> Result<Option<Session>> {
let connection = self.connection()?;
connection.execute("DELETE FROM sessions WHERE expires_at <= unixepoch()", [])?;
let session = connection
.query_row(
"SELECT id, sub, email, role, organization_id, tenant_id, groups, id_token,
created_at, expires_at
FROM sessions WHERE id = ?1 AND expires_at > unixepoch()",
[id],
|row| {
let groups: String = row.get(6)?;
Ok(Session {
id: row.get(0)?,
sub: row.get(1)?,
email: row.get(2)?,
role: row.get(3)?,
organization_id: row.get(4)?,
tenant_id: row.get(5)?,
groups: serde_json::from_str(&groups).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(
6,
rusqlite::types::Type::Text,
Box::new(error),
)
})?,
id_token: row.get(7)?,
created_at: row.get(8)?,
expires_at: row.get(9)?,
})
},
)
.optional()?;
Ok(session)
}
/// Remove a session on logout. Idempotent.
pub fn delete_session(&self, id: &str) -> Result<()> {
self.connection()?
.execute("DELETE FROM sessions WHERE id = ?1", [id])?;
Ok(())
}
/// Register an administrator-owned trusted repository. Existing callers
/// (including static-token automation) retain their upsert behavior.
pub fn add_repository(&self, name: &str, url: &str, branch: &str) -> Result<Repository> {
let connection = self.connection()?;
connection.execute(
"INSERT INTO repositories(name, url, default_branch, trusted)
VALUES (?1, ?2, ?3, 1)
ON CONFLICT(name) DO UPDATE SET
url=excluded.url, default_branch=excluded.default_branch",
params![name, url, branch],
)?;
drop(connection);
self.repository(name)
}
/// Register a repository owned by a browser session. It is deliberately
/// untrusted and never upserts an existing name.
pub fn add_repository_owned(
&self,
name: &str,
url: &str,
branch: &str,
owner_sub: &str,
owner_organization_id: Option<&str>,
) -> Result<Repository> {
ensure!(
!owner_sub.is_empty(),
"repository owner subject is required"
);
let connection = self.connection()?;
connection.execute(
"INSERT INTO repositories(
name, url, default_branch, owner_sub, owner_organization_id, trusted
) VALUES (?1, ?2, ?3, ?4, ?5, 0)",
params![name, url, branch, owner_sub, owner_organization_id],
)?;
drop(connection);
self.repository(name)
}
/// Return the highest role granted to a subject for a repository.
/// Ownership grants maintain, organization ownership grants read, and an
/// explicit collaborator grant may raise access further.
pub fn repository_role(
&self,
name: &str,
subject: &str,
organization_id: Option<&str>,
) -> Result<Option<RepositoryRole>> {
let repository = self.repository(name)?;
if repository.owner_sub.as_deref() == Some(subject) {
return Ok(Some(RepositoryRole::Maintain));
}
let mut role = if repository
.owner_organization_id
.as_deref()
.zip(organization_id)
.is_some_and(|(owner, current)| owner == current)
{
Some(RepositoryRole::Read)
} else {
None
};
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT organization_id, role
FROM repository_collaborators
WHERE repository_id=?1 AND subject=?2",
)?;
let grants = statement
.query_map(params![repository.id, subject], |row| {
let organization_id: Option<String> = row.get(0)?;
let role: String = row.get(1)?;
Ok((organization_id, role))
})?
.collect::<rusqlite::Result<Vec<_>>>()?;
drop(statement);
drop(connection);
for (grant_organization, grant_role) in grants {
if grant_organization.is_none() || grant_organization.as_deref() == organization_id {
let grant_role = RepositoryRole::try_from(grant_role.as_str())?;
if role.is_none_or(|current| grant_role > current) {
role = Some(grant_role);
}
}
}
Ok(role)
}
pub fn repository_collaborators(&self, name: &str) -> Result<Vec<RepositoryAccess>> {
let repository = self.repository(name)?;
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT subject, organization_id, role
FROM repository_collaborators
WHERE repository_id=?1 ORDER BY subject",
)?;
let rows = statement
.query_map([repository.id], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Option<String>>(1)?,
row.get::<_, String>(2)?,
))
})?
.collect::<rusqlite::Result<Vec<_>>>()?;
drop(statement);
drop(connection);
rows.into_iter()
.map(|(subject, organization_id, role)| {
Ok(RepositoryAccess {
subject,
organization_id,
role: RepositoryRole::try_from(role.as_str())?,
})
})
.collect()
}
pub fn set_repository_collaborator(
&self,
name: &str,
subject: &str,
organization_id: Option<&str>,
role: RepositoryRole,
) -> Result<()> {
ensure!(
!subject.is_empty() && subject.len() <= 256,
"collaborator subject must be 1..=256 bytes"
);
let repository = self.repository(name)?;
self.connection()?.execute(
"INSERT INTO repository_collaborators(repository_id, subject, organization_id, role)
VALUES (?1, ?2, ?3, ?4)
ON CONFLICT(repository_id, subject) DO UPDATE SET
organization_id=excluded.organization_id, role=excluded.role",
params![repository.id, subject, organization_id, role.as_str()],
)?;
Ok(())
}
pub fn remove_repository_collaborator(&self, name: &str, subject: &str) -> Result<bool> {
let repository = self.repository(name)?;
Ok(self.connection()?.execute(
"DELETE FROM repository_collaborators WHERE repository_id=?1 AND subject=?2",
params![repository.id, subject],
)? == 1)
}
pub fn set_repository_visibility(&self, name: &str, visibility: &str) -> Result<Repository> {
ensure!(
matches!(visibility, "private" | "public"),
"repository visibility must be private or public"
);
let connection = self.connection()?;
ensure!(
connection.execute(
"UPDATE repositories SET visibility=?1 WHERE name=?2",
params![visibility, name],
)? == 1,
"unknown repository {name}"
);
drop(connection);
self.repository(name)
}
pub fn update_repository(
&self,
name: &str,
url: Option<&str>,
default_branch: Option<&str>,
) -> Result<Repository> {
ensure!(
url.is_some() || default_branch.is_some(),
"repository update needs url or default_branch"
);
if let Some(branch) = default_branch {
crate::config::validate_ref(branch)?;
}
let connection = self.connection()?;
ensure!(
connection.execute(
"UPDATE repositories
SET url=COALESCE(?2, url), default_branch=COALESCE(?3, default_branch)
WHERE name=?1",
params![name, url, default_branch],
)? == 1,
"unknown repository {name}"
);
drop(connection);
self.repository(name)
}
pub fn rename_repository(&self, old: &str, new: &str) -> Result<Repository> {
crate::config::validate_repo_name(new)?;
let connection = self.connection()?;
ensure!(
connection.execute(
"UPDATE repositories SET name=?1 WHERE name=?2",
params![new, old],
)? == 1,
"unknown repository {old}"
);
drop(connection);
self.repository(new)
}
/// Unregister a repository, cascading to its runs, jobs, logs, and artifacts.
/// Returns the removed repository record. Errors if the name is unknown.
pub fn remove_repository(&self, name: &str) -> Result<Repository> {
let repository = self.repository(name)?;
let connection = self.connection()?;
ensure!(
connection.execute("DELETE FROM repositories WHERE name=?1", [name])? == 1,
"unknown repository {name}"
);
Ok(repository)
}
pub fn repository(&self, name: &str) -> Result<Repository> {
self.connection()?
.query_row(
"SELECT id, name, url, default_branch, visibility,
owner_sub, owner_organization_id, trusted, protect_default_branch, created_at
FROM repositories WHERE name=?1",
[name],
map_repository,
)
.with_context(|| format!("unknown repository {name}"))
}
/// Whether `git_ref` is the repository's default branch and that branch
/// carries `protect_default_branch=1`. Every push/sync bypass point
/// (HTTP git-receive-pack and hosted_git::sync) must call this before
/// writing, since it is the one source of truth for "protected".
pub fn is_branch_protected(&self, name: &str, git_ref: &str) -> Result<bool> {
let (default_branch, protected): (String, i64) = self.connection()?.query_row(
"SELECT default_branch, protect_default_branch FROM repositories WHERE name=?1",
[name],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
Ok(protected != 0 && git_ref == default_branch)
}
pub fn set_branch_protection(&self, name: &str, protected: bool) -> Result<Repository> {
ensure!(
self.connection()?.execute(
"UPDATE repositories SET protect_default_branch=?2 WHERE name=?1",
params![name, i64::from(protected)],
)? == 1,
"unknown repository {name}"
);
self.repository(name)
}
/// Assign (or reassign) the owning subject of a repository. Owners are
/// what makes "null owners" go away on agent-hosted mirrors.
pub fn set_repository_owner(&self, name: &str, owner_sub: &str) -> Result<Repository> {
ensure!(
!owner_sub.is_empty() && owner_sub.chars().count() <= 256,
"owner subject must be 1..=256 characters"
);
ensure!(
self.connection()?.execute(
"UPDATE repositories SET owner_sub=?2 WHERE name=?1",
params![name, owner_sub],
)? == 1,
"unknown repository {name}"
);
self.repository(name)
}
pub fn repositories(&self) -> Result<Vec<Repository>> {
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT id, name, url, default_branch, visibility,
owner_sub, owner_organization_id, trusted, protect_default_branch, created_at
FROM repositories ORDER BY name",
)?;
Ok(statement
.query_map([], map_repository)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn query_repositories(&self, query: &RepositoryQuery) -> Result<Vec<Repository>> {
ensure!(
(1..=500).contains(&query.limit),
"repository limit must be 1..=500"
);
ensure!(query.offset <= 10_000, "repository offset must be <= 10000");
if let Some(visibility) = &query.visibility {
ensure!(
matches!(visibility.as_str(), "private" | "public"),
"repository visibility must be private or public"
);
}
if let Some(search) = &query.search {
ensure!(
!search.is_empty() && search.len() <= 200,
"search must be 1..=200 bytes"
);
}
let search = query.search.as_ref().map(|value| format!("%{value}%"));
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT id, name, url, default_branch, visibility,
owner_sub, owner_organization_id, trusted, protect_default_branch, created_at
FROM repositories
WHERE (?1 IS NULL OR visibility=?1)
AND (?2 IS NULL OR name LIKE ?2 OR url LIKE ?2 OR default_branch LIKE ?2)
ORDER BY name LIMIT ?3 OFFSET ?4",
)?;
Ok(statement
.query_map(
params![
query.visibility.as_deref(),
search.as_deref(),
i64::try_from(query.limit)?,
i64::try_from(query.offset)?
],
map_repository,
)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn public_repositories(&self) -> Result<Vec<PublicRepository>> {
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT p.id, p.name, p.url, p.default_branch, p.created_at,
COUNT(r.id),
COALESCE(SUM(CASE WHEN r.status='succeeded' THEN 1 ELSE 0 END), 0),
COALESCE(MAX(COALESCE(r.finished_at, r.started_at, r.created_at)), p.created_at)
FROM repositories p
LEFT JOIN runs r ON r.repository_id=p.id
WHERE p.visibility='public'
GROUP BY p.id
ORDER BY COALESCE(MAX(COALESCE(r.finished_at, r.started_at, r.created_at)), p.created_at) DESC,
COUNT(r.id) DESC,
p.name
LIMIT 12",
)?;
Ok(statement
.query_map([], map_public_repository)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn create_run(
&self,
repository_id: i64,
git_ref: &str,
commit: Option<&str>,
trigger: &str,
) -> Result<i64> {
let connection = self.connection()?;
connection.execute(
"INSERT INTO runs(repository_id, git_ref, commit_sha, trigger, status) VALUES (?1, ?2, ?3, ?4, 'queued')",
params![repository_id, git_ref, commit, trigger],
)?;
Ok(connection.last_insert_rowid())
}
/// Create a queued run, or return its existing active equivalent. The
/// immediate transaction serializes concurrent MCP and webhook callers.
pub fn create_or_get_active_run(
&self,
repository_id: i64,
git_ref: &str,
commit: &str,
trigger: &str,
) -> Result<(i64, bool)> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
if let Some(id) = transaction
.query_row(
"SELECT id FROM runs
WHERE repository_id=?1 AND commit_sha=?2
AND status IN ('queued', 'running', 'waiting')
ORDER BY id DESC LIMIT 1",
params![repository_id, commit],
|row| row.get(0),
)
.optional()?
{
transaction.commit()?;
return Ok((id, true));
}
transaction.execute(
"INSERT INTO runs(repository_id, git_ref, commit_sha, trigger, status)
VALUES (?1, ?2, ?3, ?4, 'queued')",
params![repository_id, git_ref, commit, trigger],
)?;
let id = transaction.last_insert_rowid();
transaction.commit()?;
Ok((id, false))
}
/// Create a queued run for a ref whose commit is not yet known, reusing
/// an existing queued run of the same repository and ref that is also
/// still unresolved. Collapses duplicate webhook deliveries that arrive
/// before the first run resolves its commit.
pub fn create_or_get_pending_run(
&self,
repository_id: i64,
git_ref: &str,
trigger: &str,
) -> Result<(i64, bool)> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
if let Some(id) = transaction
.query_row(
"SELECT id FROM runs
WHERE repository_id=?1 AND git_ref=?2 AND commit_sha IS NULL AND status='queued'
ORDER BY id DESC LIMIT 1",
params![repository_id, git_ref],
|row| row.get(0),
)
.optional()?
{
transaction.commit()?;
return Ok((id, true));
}
transaction.execute(
"INSERT INTO runs(repository_id, git_ref, commit_sha, trigger, status)
VALUES (?1, ?2, NULL, ?3, 'queued')",
params![repository_id, git_ref, trigger],
)?;
let id = transaction.last_insert_rowid();
transaction.commit()?;
Ok((id, false))
}
/// Cancel older approval-waiting runs of the same repository and ref that
/// a newly queued run supersedes, cascading to their not-yet-final jobs.
/// Returns the canceled run ids.
pub fn supersede_waiting_runs(
&self,
repository_id: i64,
git_ref: &str,
keep: i64,
) -> Result<Vec<i64>> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let ids: Vec<i64> = transaction
.prepare(
"SELECT id FROM runs
WHERE repository_id=?1 AND git_ref=?2 AND status='waiting' AND id<>?3",
)?
.query_map(params![repository_id, git_ref, keep], |row| row.get(0))?
.collect::<rusqlite::Result<_>>()?;
for id in &ids {
transaction.execute(
"UPDATE runs SET status='canceled', error=NULL, finished_at=unixepoch() WHERE id=?1",
[id],
)?;
transaction.execute(
"UPDATE jobs SET status='canceled', finished_at=unixepoch()
WHERE run_id=?1 AND status IN ('queued','waiting','running')",
[id],
)?;
}
transaction.commit()?;
Ok(ids)
}
pub fn run(&self, id: i64) -> Result<Run> {
self.connection()?
.query_row(
"SELECT r.id, r.repository_id, p.name, r.git_ref, r.commit_sha, r.trigger, r.status,
r.error, r.created_at, r.started_at, r.finished_at
FROM runs r JOIN repositories p ON p.id=r.repository_id WHERE r.id=?1",
[id],
map_run,
)
.with_context(|| format!("unknown run {id}"))
}
pub fn runs(&self, repository: Option<&str>, limit: usize) -> Result<Vec<Run>> {
self.query_runs(&RunQuery {
repository: repository.map(str::to_owned),
limit,
..RunQuery::default()
})
}
pub fn query_runs(&self, query: &RunQuery) -> Result<Vec<Run>> {
let ValidatedRunQuery {
predicate,
mut values,
} = validate_run_query(query)?;
let mut sql = String::from(
"SELECT r.id, r.repository_id, p.name, r.git_ref, r.commit_sha, r.trigger, r.status,
r.error, r.created_at, r.started_at, r.finished_at
FROM runs r JOIN repositories p ON p.id=r.repository_id",
);
sql.push_str(&predicate);
sql.push_str(" ORDER BY r.id DESC LIMIT ? OFFSET ?");
values.push(Value::Integer(i64::try_from(query.limit)?));
values.push(Value::Integer(i64::try_from(query.offset)?));
let connection = self.connection()?;
let mut statement = connection.prepare(&sql)?;
Ok(statement
.query_map(params_from_iter(values), map_run)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn count_runs(&self, query: &RunQuery) -> Result<i64> {
let ValidatedRunQuery { predicate, values } = validate_run_query(query)?;
let sql = format!(
"SELECT COUNT(*) FROM runs r JOIN repositories p ON p.id=r.repository_id{predicate}"
);
Ok(self
.connection()?
.query_row(&sql, params_from_iter(values), |row| row.get(0))?)
}
pub fn max_run_id(&self, query: &RunQuery) -> Result<Option<i64>> {
let ValidatedRunQuery { predicate, values } = validate_run_query(query)?;
let sql = format!(
"SELECT MAX(r.id) FROM runs r JOIN repositories p ON p.id=r.repository_id{predicate}"
);
Ok(self
.connection()?
.query_row(&sql, params_from_iter(values), |row| row.get(0))?)
}
pub fn detail(&self, id: i64) -> Result<RunDetail> {
Ok(RunDetail {
run: self.run(id)?,
jobs: self.jobs(id)?,
artifacts: self.artifacts(id)?,
deployments: self.deployments(id)?,
})
}
pub fn claim_next_queued_run(&self) -> Result<Option<i64>> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let id = transaction
.query_row(
"SELECT id FROM runs WHERE status='queued' ORDER BY id LIMIT 1",
[],
|row| row.get(0),
)
.optional()?;
if let Some(id) = id {
ensure!(
transaction.execute(
"UPDATE runs SET status='running',
started_at=COALESCE(started_at, unixepoch()),
error=CASE WHEN error='controller restarted during execution; retrying incomplete job' THEN error ELSE NULL END
WHERE id=?1 AND status='queued'",
[id],
)? == 1,
"queued run changed during claim"
);
}
transaction.commit()?;
Ok(id)
}
pub fn claim_run(&self, id: i64) -> Result<bool> {
Ok(self.connection()?.execute(
"UPDATE runs SET status='running', started_at=COALESCE(started_at, unixepoch()), error=NULL WHERE id=?1 AND status='queued'",
[id],
)? == 1)
}
pub fn set_run_commit(&self, id: i64, commit: &str) -> Result<()> {
self.connection()?.execute(
"UPDATE runs SET commit_sha=?2 WHERE id=?1",
params![id, commit],
)?;
Ok(())
}
pub fn finish_run(&self, id: i64, status: &str, error: Option<&str>) -> Result<()> {
self.connection()?.execute(
"UPDATE runs SET status=?2, error=?3, finished_at=CASE WHEN ?2 IN ('succeeded','failed','canceled','interrupted') THEN unixepoch() ELSE NULL END WHERE id=?1",
params![id, status, error],
)?;
Ok(())
}
pub fn insert_jobs(&self, run_id: i64, specs: &[crate::config::JobSpec]) -> Result<()> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
for spec in specs {
let spec_json = serde_json::to_string(spec)?;
let needs = serde_json::to_string(&spec.needs)?;
let platform = spec.matrix.get("platform");
transaction.execute(
"INSERT INTO jobs(run_id, base_name, name, executor, image, platform, status, needs_json, spec_json, environment, approval_required)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'queued', ?7, ?8, ?9, ?10)",
params![run_id, spec.base_name, spec.name, spec.executor, spec.image, platform, needs, spec_json, spec.environment, spec.approval],
)?;
}
transaction.commit()?;
Ok(())
}
pub fn jobs(&self, run_id: i64) -> Result<Vec<Job>> {
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT id, run_id, base_name, name, executor, image, platform, status, needs_json,
spec_json, logs, exit_code, environment, approval_required, approved_at, approved_by,
started_at, finished_at
FROM jobs WHERE run_id=?1 ORDER BY id",
)?;
Ok(statement
.query_map([run_id], map_job)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn set_job_status(
&self,
id: i64,
status: &str,
logs: Option<&str>,
exit_code: Option<i32>,
) -> Result<()> {
self.connection()?.execute(
"UPDATE jobs SET status=?2, logs=COALESCE(?3, logs), exit_code=?4,
started_at=CASE WHEN ?2='running' THEN COALESCE(started_at, unixepoch()) ELSE started_at END,
finished_at=CASE WHEN ?2 IN ('succeeded','failed','skipped','canceled','interrupted') THEN unixepoch() ELSE finished_at END
WHERE id=?1",
params![id, status, logs, exit_code],
)?;
Ok(())
}
pub fn approve_environment(&self, run_id: i64, environment: &str) -> Result<usize> {
self.approve_environment_as(run_id, environment, "system")
}
pub fn approve_environment_as(
&self,
run_id: i64,
environment: &str,
approver: &str,
) -> Result<usize> {
ensure!(
!approver.is_empty() && approver.len() <= 256,
"approver identity must be 1..=256 bytes"
);
let connection = self.connection()?;
let changed = connection.execute(
"UPDATE jobs SET approved_at=unixepoch(), approved_by=?3,
status=CASE WHEN status='waiting' THEN 'queued' ELSE status END
WHERE run_id=?1 AND environment=?2 AND approval_required=1 AND status IN ('waiting','queued')",
params![run_id, environment, approver],
)?;
ensure!(
changed > 0,
"no promotable {environment} job in run {run_id}"
);
connection.execute(
"UPDATE runs SET status='queued', finished_at=NULL WHERE id=?1 AND status='waiting'",
[run_id],
)?;
Ok(changed)
}
/// Cancel a run that has not started executing (status queued or
/// waiting), cascading to its not-yet-final jobs. Running and terminal
/// runs are refused: the worker owns a running run's lifecycle.
pub fn cancel_run(&self, id: i64) -> Result<()> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let changed = transaction.execute(
"UPDATE runs SET status='canceled', error=NULL, finished_at=unixepoch()
WHERE id=?1 AND status IN ('queued','waiting')",
[id],
)?;
if changed != 1 {
let status: Option<String> = transaction
.query_row("SELECT status FROM runs WHERE id=?1", [id], |row| {
row.get(0)
})
.optional()?;
transaction.commit()?;
match status {
Some(status) => {
bail!("run {id} is {status}; only queued or waiting runs can be canceled")
}
None => bail!("unknown run {id}"),
}
}
transaction.execute(
"UPDATE jobs SET status='canceled', finished_at=unixepoch()
WHERE run_id=?1 AND status IN ('queued','waiting','running')",
[id],
)?;
transaction.commit()?;
Ok(())
}
/// Cancel a run currently owned by a worker. The worker calls this only
/// after its child process has exited, so all live execution state can be
/// finalized atomically without racing the executor.
pub fn cancel_running_run(&self, id: i64) -> Result<()> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let changed = transaction.execute(
"UPDATE runs SET status='canceled', error=NULL, finished_at=unixepoch()
WHERE id=?1 AND status='running'",
[id],
)?;
if changed != 1 {
let status: Option<String> = transaction
.query_row("SELECT status FROM runs WHERE id=?1", [id], |row| {
row.get(0)
})
.optional()?;
transaction.commit()?;
match status {
Some(status) => {
bail!("run {id} is {status}; only running runs can be canceled by a worker")
}
None => bail!("unknown run {id}"),
}
}
transaction.execute(
"UPDATE jobs SET status='canceled',
logs=CASE WHEN logs='' THEN 'run canceled by operator'
ELSE logs || char(10) || 'run canceled by operator' END,
finished_at=unixepoch()
WHERE run_id=?1 AND status IN ('queued','waiting','running')",
[id],
)?;
transaction.execute(
"UPDATE deployments SET status='failed', result='failure', finished_at=unixepoch()
WHERE run_id=?1 AND status='running'",
[id],
)?;
transaction.commit()?;
Ok(())
}
pub fn retry(&self, run_id: i64) -> Result<i64> {
let run = self.run(run_id)?;
self.create_run(
run.repository_id,
&run.git_ref,
run.commit_sha.as_deref(),
"retry",
)
}
pub fn add_artifact(
&self,
run_id: i64,
job_id: i64,
name: &str,
path: &str,
sha256: &str,
bytes: u64,
) -> Result<i64> {
let connection = self.connection()?;
connection.execute(
"INSERT INTO artifacts(run_id, job_id, name, path, sha256, bytes) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![run_id, job_id, name, path, sha256, i64::try_from(bytes)?],
)?;
Ok(connection.last_insert_rowid())
}
pub fn artifact(&self, id: i64) -> Result<Artifact> {
self.connection()?.query_row(
"SELECT id, run_id, job_id, name, path, sha256, bytes, created_at FROM artifacts WHERE id=?1",
[id],
map_artifact,
).with_context(|| format!("unknown artifact {id}"))
}
pub fn artifacts(&self, run_id: i64) -> Result<Vec<Artifact>> {
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT id, run_id, job_id, name, path, sha256, bytes, created_at FROM artifacts WHERE run_id=?1 ORDER BY id",
)?;
Ok(statement
.query_map([run_id], map_artifact)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn begin_deployment(
&self,
run_id: i64,
job_id: i64,
environment: &str,
artifacts_json: &str,
) -> Result<i64> {
let connection = self.connection()?;
connection.execute(
"INSERT INTO deployments(run_id, job_id, environment, status, artifacts_json) VALUES (?1, ?2, ?3, 'running', ?4)",
params![run_id, job_id, environment, artifacts_json],
)?;
Ok(connection.last_insert_rowid())
}
pub fn register_installation_revision(
&self,
installation_id: &str,
organization_id: &str,
workspace_id: &str,
app_id: &str,
idempotency_key: &str,
) -> Result<i64> {
ensure!(
!installation_id.is_empty()
&& !organization_id.is_empty()
&& !workspace_id.is_empty()
&& !app_id.is_empty(),
"installation identity is required"
);
ensure!(!idempotency_key.is_empty(), "idempotency key is required");
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
if let Some((revision, existing_org, existing_workspace, existing_app)) = transaction
.query_row(
"SELECT revision, organization_id, workspace_id, app_id
FROM installation_revisions
WHERE installation_id=?1 AND idempotency_key=?2",
params![installation_id, idempotency_key],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
))
},
)
.optional()?
{
ensure!(
existing_org == organization_id
&& existing_workspace == workspace_id
&& existing_app == app_id,
"installation tenant identity mismatch"
);
return Ok(revision);
}
let current: Option<(String, String, String)> = transaction
.query_row(
"SELECT organization_id, workspace_id, app_id FROM managed_installations WHERE installation_id=?1",
[installation_id],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.optional()?;
if let Some((existing_org, existing_workspace, existing_app)) = current.as_ref() {
ensure!(
existing_org == organization_id
&& existing_workspace == workspace_id
&& existing_app == app_id,
"installation tenant identity mismatch"
);
}
let revision: i64 = transaction.query_row(
"SELECT COALESCE(MAX(revision), 0) + 1 FROM installation_revisions WHERE installation_id=?1",
[installation_id],
|row| row.get(0),
)?;
if current.is_none() {
transaction.execute(
"INSERT INTO managed_installations(installation_id, organization_id, workspace_id, app_id, revision) VALUES (?1, ?2, ?3, ?4, ?5)",
params![installation_id, organization_id, workspace_id, app_id, revision],
)?;
}
transaction.execute(
"INSERT INTO installation_revisions(installation_id, revision, organization_id, workspace_id, app_id, idempotency_key)
VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![installation_id, revision, organization_id, workspace_id, app_id, idempotency_key],
)?;
transaction.execute(
"UPDATE managed_installations SET revision=?2, updated_at=unixepoch() WHERE installation_id=?1",
params![installation_id, revision],
)?;
transaction.commit()?;
Ok(revision)
}
pub fn begin_managed_deployment(&self, deployment: ManagedDeployment<'_>) -> Result<i64> {
let ManagedDeployment {
run_id,
job_id,
environment,
artifacts_json,
installation_id,
installation_revision,
commit_sha,
artifact_digest,
result,
rollback_of_deployment_id,
recovery_of_deployment_id,
} = deployment;
ensure!(
!installation_id.is_empty() && installation_revision > 0,
"managed installation is required"
);
ensure!(
!environment.is_empty(),
"deployment environment is required"
);
if let Some(commit_sha) = commit_sha {
ensure!(
matches!(commit_sha.len(), 40 | 64)
&& commit_sha.bytes().all(|byte| byte.is_ascii_hexdigit()),
"invalid deployment commit SHA"
);
}
if let Some(artifact_digest) = artifact_digest {
ensure!(
artifact_digest.len() == 64
&& artifact_digest.bytes().all(|byte| byte.is_ascii_hexdigit()),
"invalid deployment artifact digest"
);
}
let artifact_manifest: Vec<JsonValue> =
serde_json::from_str(artifacts_json).context("invalid deployment artifact manifest")?;
let calculated_digest = format!("{:x}", Sha256::digest(artifacts_json.as_bytes()));
ensure!(
artifact_digest == Some(calculated_digest.as_str()),
"deployment artifact digest does not match manifest"
);
if let Some(result) = result {
ensure!(
matches!(result, "success" | "failure" | "rollback" | "recovery"),
"invalid deployment result"
);
}
ensure!(
rollback_of_deployment_id.is_none() || recovery_of_deployment_id.is_none(),
"deployment cannot be both rollback and recovery"
);
let connection = self.connection()?;
let run_commit: Option<String> = connection
.query_row("SELECT commit_sha FROM runs WHERE id=?1", [run_id], |row| {
row.get(0)
})
.with_context(|| format!("unknown run {run_id}"))?;
ensure!(
commit_sha.is_some() && run_commit.is_some(),
"managed deployment requires an exact run commit"
);
ensure!(
artifact_digest.is_some(),
"managed deployment requires an artifact digest"
);
ensure!(
run_commit.as_deref() == commit_sha,
"deployment commit does not match run commit"
);
let job_belongs_to_run: bool = connection.query_row(
"SELECT EXISTS(SELECT 1 FROM jobs WHERE id=?1 AND run_id=?2)",
params![job_id, run_id],
|row| row.get(0),
)?;
ensure!(job_belongs_to_run, "deployment job does not belong to run");
for artifact in &artifact_manifest {
let artifact_id = artifact
.get("id")
.and_then(JsonValue::as_i64)
.context("deployment artifact id is missing")?;
let artifact_sha = artifact
.get("sha256")
.and_then(JsonValue::as_str)
.context("deployment artifact digest is missing")?;
let (stored_run_id, stored_sha): (i64, String) = connection.query_row(
"SELECT run_id, sha256 FROM artifacts WHERE id=?1",
[artifact_id],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
ensure!(
stored_run_id == run_id && stored_sha == artifact_sha,
"deployment artifact does not match stored artifact"
);
}
let tenant: (String, String, String) = connection.query_row(
"SELECT organization_id, workspace_id, app_id FROM installation_revisions WHERE installation_id=?1 AND revision=?2",
params![installation_id, installation_revision],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)?;
for lineage_id in [rollback_of_deployment_id, recovery_of_deployment_id]
.into_iter()
.flatten()
{
let lineage_tenant: (Option<String>, Option<String>) = connection.query_row(
"SELECT organization_id, workspace_id FROM deployments WHERE id=?1",
[lineage_id],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
ensure!(
lineage_tenant.0.as_deref() == Some(tenant.0.as_str())
&& lineage_tenant.1.as_deref() == Some(tenant.1.as_str()),
"deployment lineage tenant mismatch"
);
}
connection.execute(
"INSERT INTO deployments(run_id, job_id, environment, status, artifacts_json, organization_id, workspace_id, installation_id, app_id, installation_revision, commit_sha, artifact_digest, result, rollback_of_deployment_id, recovery_of_deployment_id)
VALUES (?1, ?2, ?3, 'running', ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
params![run_id, job_id, environment, artifacts_json, tenant.0, tenant.1, installation_id, tenant.2, installation_revision, commit_sha, artifact_digest, result, rollback_of_deployment_id, recovery_of_deployment_id],
)?;
Ok(connection.last_insert_rowid())
}
/// Source-derived delivery metrics. Only terminal deployments with a
/// persisted completion timestamp are sampled. Missing timestamps remain
/// unknown rather than being synthesized as zero.
pub fn delivery_metrics(
&self,
repository: Option<&str>,
environment: Option<&str>,
window_seconds: i64,
) -> Result<DeliveryMetrics> {
ensure!(
(1..=30 * 86_400).contains(&window_seconds),
"window must be 1..=30 days"
);
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)?
.as_secs() as i64;
let since = now.saturating_sub(window_seconds);
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT d.id, d.run_id, d.status, d.finished_at, d.result,
d.recovery_of_deployment_id, r.created_at, r.started_at, r.status
FROM deployments d
JOIN runs r ON r.id=d.run_id
JOIN repositories p ON p.id=r.repository_id
WHERE d.status IN ('succeeded', 'failed')
AND d.finished_at IS NOT NULL
AND d.finished_at >= ?1
AND (?2 IS NULL OR p.name=?2)
AND (?3 IS NULL OR d.environment=?3)
ORDER BY d.id
LIMIT ?4",
)?;
let rows = statement.query_map(
params![
since,
repository,
environment,
i64::try_from(MAX_METRIC_ROWS + 1)?
],
|row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, String>(2)?,
row.get::<_, i64>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<i64>>(5)?,
row.get::<_, i64>(6)?,
row.get::<_, Option<i64>>(7)?,
row.get::<_, String>(8)?,
))
},
)?;
let mut deployment_ids = Vec::new();
let mut run_ids = Vec::new();
let mut successful = 0_i64;
let mut failures = 0_i64;
let mut lead_total = 0_i64;
let mut lead_count = 0_i64;
let mut queue_total = 0_i64;
let mut queue_count = 0_i64;
let mut recovery_total = 0_i64;
let mut recovery_count = 0_i64;
let mut completed_deployments = Vec::new();
let mut rows_seen = 0usize;
for row in rows {
rows_seen = rows_seen.saturating_add(1);
if rows_seen > MAX_METRIC_ROWS {
continue;
}
let (
id,
run_id,
status,
finished_at,
result,
recovery_of,
run_created,
run_started,
run_status,
) = row?;
if matches!(run_status.as_str(), "canceled" | "interrupted") {
continue;
}
deployment_ids.push(id);
run_ids.push(run_id);
completed_deployments.push((id, finished_at));
let failed =
status == "failed" || matches!(result.as_deref(), Some("failure" | "rollback"));
if failed {
failures += 1;
} else if status == "succeeded" {
successful += 1;
if finished_at >= run_created {
lead_total += finished_at - run_created;
lead_count += 1;
}
}
if let Some(started) = run_started
&& started >= run_created
{
queue_total += started - run_created;
queue_count += 1;
}
if status == "succeeded"
&& result.as_deref() == Some("recovery")
&& let Some(failed_id) = recovery_of
{
let failed_at = completed_deployments
.iter()
.find(|(candidate, _)| *candidate == failed_id)
.map(|(_, completed)| *completed)
.or_else(|| {
connection
.query_row(
"SELECT finished_at FROM deployments WHERE id=?1
AND status IN ('succeeded', 'failed')",
[failed_id],
|row| row.get::<_, Option<i64>>(0),
)
.optional()
.ok()
.flatten()
.flatten()
});
if let Some(failed_at) = failed_at
&& finished_at >= failed_at
{
recovery_total += finished_at - failed_at;
recovery_count += 1;
}
}
}
let mut job_stages = Vec::new();
if !run_ids.is_empty() {
let placeholders = std::iter::repeat_n("?", run_ids.len())
.collect::<Vec<_>>()
.join(", ");
let mut jobs = connection.prepare(&format!(
"SELECT id, run_id, name, status, spec_json, started_at, finished_at
FROM jobs WHERE run_id IN ({placeholders}) ORDER BY id"
))?;
let rows = jobs.query_map(params_from_iter(run_ids.iter()), |row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, i64>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
row.get::<_, String>(4)?,
row.get::<_, Option<i64>>(5)?,
row.get::<_, Option<i64>>(6)?,
))
})?;
for row in rows {
let (job_id, run_id, name, status, spec_json, started_at, finished_at) = row?;
let cache_paths = serde_json::from_str::<JsonValue>(&spec_json)
.ok()
.and_then(|spec| spec.get("cache").cloned())
.and_then(|cache| cache.as_array().cloned())
.map(|paths| {
paths
.into_iter()
.filter_map(|path| path.as_str().map(str::to_owned))
.collect::<Vec<_>>()
})
.unwrap_or_default();
let duration_seconds =
started_at.zip(finished_at).and_then(|(started, finished)| {
(finished >= started).then_some((finished - started) as f64)
});
job_stages.push(JobStageMetric {
job_id,
run_id,
name,
status,
started_at,
finished_at,
duration_seconds,
cache_paths,
});
}
}
let sample_size = successful + failures;
let days = window_seconds as f64 / 86_400.0;
Ok(DeliveryMetrics {
repository: repository.map(str::to_owned),
environment: environment.map(str::to_owned),
window_seconds,
since,
fresh_at: now,
sample_size,
deployment_frequency_per_day: (successful > 0).then(|| successful as f64 / days),
commit_to_production_seconds: (lead_count > 0)
.then(|| lead_total as f64 / lead_count as f64),
change_failure_rate: (sample_size > 0).then(|| failures as f64 / sample_size as f64),
recovery_time_seconds: (recovery_count > 0)
.then(|| recovery_total as f64 / recovery_count as f64),
queue_duration_seconds: (queue_count > 0)
.then(|| queue_total as f64 / queue_count as f64),
job_stages,
deployment_ids,
run_ids,
truncated: rows_seen > MAX_METRIC_ROWS,
})
}
pub fn finish_deployment(&self, id: i64, status: &str) -> Result<()> {
let result = match status {
"succeeded" => Some("success"),
"failed" => Some("failure"),
"running" => None,
_ => bail!("invalid deployment status"),
};
self.finish_deployment_result(id, status, result)
}
pub fn finish_deployment_result(
&self,
id: i64,
status: &str,
result: Option<&str>,
) -> Result<()> {
ensure!(
matches!(status, "running" | "succeeded" | "failed"),
"invalid deployment status"
);
if let Some(result) = result {
ensure!(
matches!(result, "success" | "failure" | "rollback" | "recovery"),
"invalid deployment result"
);
}
self.connection()?.execute(
"UPDATE deployments SET status=?2, result=?3, finished_at=unixepoch() WHERE id=?1",
params![id, status, result],
)?;
Ok(())
}
pub fn deployments(&self, run_id: i64) -> Result<Vec<Deployment>> {
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT id, run_id, job_id, environment, status, artifacts_json, created_at, finished_at,
organization_id, workspace_id, installation_id, app_id, installation_revision,
commit_sha, artifact_digest, result, rollback_of_deployment_id, recovery_of_deployment_id
FROM deployments WHERE run_id=?1 ORDER BY id",
)?;
Ok(statement
.query_map([run_id], map_deployment)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn deployments_for_tenant(
&self,
run_id: i64,
organization_id: &str,
workspace_id: &str,
) -> Result<Vec<Deployment>> {
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT id, run_id, job_id, environment, status, artifacts_json, created_at, finished_at,
organization_id, workspace_id, installation_id, app_id, installation_revision,
commit_sha, artifact_digest, result, rollback_of_deployment_id, recovery_of_deployment_id
FROM deployments
WHERE run_id=?1 AND organization_id=?2 AND workspace_id=?3
ORDER BY id",
)?;
let deployments = statement
.query_map(
params![run_id, organization_id, workspace_id],
map_deployment,
)?
.collect::<rusqlite::Result<Vec<_>>>()?;
if deployments.is_empty() {
let run_has_deployment: bool = connection.query_row(
"SELECT EXISTS(SELECT 1 FROM deployments WHERE run_id=?1)",
[run_id],
|row| row.get(0),
)?;
ensure!(!run_has_deployment, "tenant access denied");
}
Ok(deployments)
}
pub fn repository_trusted(&self, repository_id: i64) -> Result<bool> {
let trusted: i64 = self.connection()?.query_row(
"SELECT trusted FROM repositories WHERE id=?1",
[repository_id],
|row| row.get(0),
)?;
Ok(trusted != 0)
}
pub fn queued_runs(&self) -> Result<i64> {
Ok(self.connection()?.query_row(
"SELECT COUNT(*) FROM runs WHERE status='queued'",
[],
|row| row.get(0),
)?)
}
pub fn reset_workers(&self) -> Result<()> {
self.connection()?.execute(
"UPDATE workers SET status='offline', current_run_id=NULL,
lease_token=NULL, lease_expires_at=NULL, heartbeat_at=unixepoch()
WHERE COALESCE(lease_expires_at, 0) <= unixepoch()",
[],
)?;
Ok(())
}
/// Delete worker rows whose id is not in `keep`, reconciling the table to
/// the currently configured worker set. Only the executor instance
/// (worker_count > 0) owns the worker table, so a serve-only restart never
/// calls this and cannot wipe a live worker's row. Refuses an empty keep
/// set so a misconfigured caller cannot clear every worker. Returns the
/// number of stale rows removed.
pub fn prune_workers_except(&self, keep: &[String]) -> Result<usize> {
ensure!(
!keep.is_empty(),
"refusing to prune workers with an empty keep set"
);
let placeholders = keep.iter().map(|_| "?").collect::<Vec<_>>().join(", ");
let removed = self.connection()?.execute(
&format!("DELETE FROM workers WHERE id NOT IN ({placeholders})"),
params_from_iter(keep.iter()),
)?;
Ok(removed)
}
pub fn register_worker(
&self,
id: &str,
host: &str,
capabilities: &str,
capacity: usize,
) -> Result<()> {
ensure!(!id.is_empty() && id.len() <= 64, "invalid worker id");
ensure!(!host.is_empty() && host.len() <= 128, "invalid worker host");
ensure!(
!capabilities.is_empty() && capabilities.len() <= 256,
"invalid capabilities"
);
ensure!(
(1..=16).contains(&capacity),
"worker capacity must be 1..=16"
);
self.connection()?.execute(
"INSERT INTO workers(id, host, capabilities, status, capacity, started_at, heartbeat_at)
VALUES (?1, ?2, ?3, 'starting', ?4, unixepoch(), unixepoch())
ON CONFLICT(id) DO UPDATE SET host=excluded.host, capabilities=excluded.capabilities,
capacity=excluded.capacity",
params![id, host, capabilities, i64::try_from(capacity)?],
)?;
Ok(())
}
pub fn acquire_worker_lease(&self, id: &str, lease_token: &str) -> Result<WorkerLease> {
ensure!(
!lease_token.is_empty() && lease_token.len() <= 128,
"invalid worker lease"
);
const LEASE_SECONDS: i64 = 30;
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let now: i64 = transaction.query_row("SELECT unixepoch()", [], |row| row.get(0))?;
let row = transaction
.query_row(
"SELECT current_run_id, lease_token, COALESCE(lease_expires_at, 0)
FROM workers WHERE id=?1",
[id],
|row| {
Ok((
row.get::<_, Option<i64>>(0)?,
row.get::<_, Option<String>>(1)?,
row.get::<_, i64>(2)?,
))
},
)
.optional()?
.context("unknown worker")?;
if row.2 > now && row.1.as_deref() != Some(lease_token) {
transaction.commit()?;
return Ok(WorkerLease {
acquired: false,
previous_run_id: None,
});
}
transaction.execute(
"UPDATE workers SET status='starting', lease_token=?2,
lease_expires_at=?3, heartbeat_at=?4 WHERE id=?1",
params![id, lease_token, now + LEASE_SECONDS, now],
)?;
transaction.commit()?;
Ok(WorkerLease {
acquired: true,
previous_run_id: row.0,
})
}
pub fn set_worker_draining(&self, id: &str, draining: bool) -> Result<()> {
ensure!(
self.connection()?.execute(
"UPDATE workers SET draining=?2, heartbeat_at=unixepoch()
WHERE id=?1",
params![id, i64::from(draining)],
)? == 1,
"unknown worker {id}"
);
Ok(())
}
pub fn heartbeat_worker_lease(
&self,
id: &str,
lease_token: &str,
status: &str,
current_run_id: Option<i64>,
completed: bool,
) -> Result<()> {
ensure!(
matches!(status, "starting" | "idle" | "running" | "offline"),
"invalid worker status"
);
ensure!(
self.connection()?.execute(
"UPDATE workers SET status=?3, current_run_id=?4, heartbeat_at=unixepoch(),
lease_expires_at=unixepoch()+30,
completed_runs=completed_runs + ?5
WHERE id=?1 AND lease_token=?2",
params![
id,
lease_token,
status,
current_run_id,
i64::from(completed)
],
)? == 1,
"worker lease is not held"
);
Ok(())
}
pub fn claim_next_queued_run_for_worker(
&self,
worker_id: &str,
lease_token: &str,
) -> Result<Option<i64>> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let now: i64 = transaction.query_row("SELECT unixepoch()", [], |row| row.get(0))?;
let (lease_valid, draining): (bool, bool) = transaction.query_row(
"SELECT EXISTS(
SELECT 1 FROM workers
WHERE id=?1 AND lease_token=?2
AND COALESCE(lease_expires_at, 0) > ?3
),
COALESCE((SELECT draining FROM workers WHERE id=?1), 0) != 0",
params![worker_id, lease_token, now],
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
ensure!(lease_valid, "worker lease is not eligible to claim");
if draining {
transaction.commit()?;
return Ok(None);
}
let id = transaction
.query_row(
"SELECT id FROM runs WHERE status='queued' ORDER BY id LIMIT 1",
[],
|row| row.get(0),
)
.optional()?;
if let Some(id) = id {
ensure!(
transaction.execute(
"UPDATE runs SET status='running',
started_at=COALESCE(started_at, unixepoch()),
error=CASE WHEN error='controller restarted during execution; retrying incomplete job' THEN error ELSE NULL END
WHERE id=?1 AND status='queued'",
[id],
)? == 1,
"queued run changed during claim"
);
transaction.execute(
"UPDATE workers SET status='running', current_run_id=?2,
heartbeat_at=?3, lease_expires_at=?3+30
WHERE id=?1 AND lease_token=?4",
params![worker_id, id, now, lease_token],
)?;
}
transaction.commit()?;
Ok(id)
}
pub fn recover_worker_run(&self, run_id: i64) -> Result<usize> {
let connection = self.connection()?;
let deployments = connection.execute(
"UPDATE deployments SET status='failed', result='failure', finished_at=unixepoch()
WHERE run_id=?1 AND status='running'",
[run_id],
)?;
let jobs = connection.execute(
"UPDATE jobs SET status='queued',
logs=CASE WHEN logs='' THEN 'controller restarted during execution; retrying job; process exit status unavailable'
ELSE logs || char(10) || 'controller restarted during execution; retrying job; process exit status unavailable' END,
exit_code=NULL, finished_at=NULL
WHERE run_id=?1 AND status='running'",
[run_id],
)?;
let runs = connection.execute(
"UPDATE runs SET status='queued', error='controller restarted during execution; retrying incomplete job',
finished_at=NULL WHERE id=?1 AND status='running'",
[run_id],
)?;
Ok(deployments + jobs + runs)
}
pub fn heartbeat_worker(
&self,
id: &str,
status: &str,
current_run_id: Option<i64>,
completed: bool,
) -> Result<()> {
ensure!(
matches!(status, "starting" | "idle" | "running" | "offline"),
"invalid worker status"
);
ensure!(
self.connection()?.execute(
"UPDATE workers SET status=?2, current_run_id=?3, heartbeat_at=unixepoch(),
completed_runs=completed_runs + ?4
WHERE id=?1 AND (lease_token IS NULL OR COALESCE(lease_expires_at, 0) <= unixepoch())",
params![id, status, current_run_id, i64::from(completed)],
)? == 1,
"unknown worker {id}"
);
Ok(())
}
pub fn workers(&self) -> Result<Vec<Worker>> {
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT id, host, capabilities, status, capacity, current_run_id, started_at,
heartbeat_at, completed_runs, draining, lease_expires_at
FROM workers ORDER BY id",
)?;
Ok(statement
.query_map([], map_worker)?
.collect::<rusqlite::Result<_>>()?)
}
/// Requeue in-flight work after a controller restart.
///
/// Jobs are child processes of the controller, so a service replacement
/// ends them without an exit status. Retrying the interrupted job is safer
/// than reporting a synthetic verification failure: completed jobs remain
/// immutable, while the queued job re-executes from the same run checkout.
/// Keep an explicit diagnostic because an actual command failure never had
/// a chance to flush its output into the database.
pub fn recover_interrupted(&self) -> Result<usize> {
let connection = self.connection()?;
let deployments = connection.execute(
"UPDATE deployments SET status='failed', result='failure', finished_at=unixepoch() WHERE status='running'",
[],
)?;
let jobs = connection.execute(
"UPDATE jobs SET status='queued',
logs=CASE WHEN logs='' THEN 'controller restarted during execution; retrying job; process exit status unavailable'
ELSE logs || char(10) || 'controller restarted during execution; retrying job; process exit status unavailable' END,
exit_code=NULL, finished_at=NULL
WHERE status='running'",
[],
)?;
let runs = connection.execute(
"UPDATE runs SET status='queued', error='controller restarted during execution; retrying incomplete job', finished_at=NULL WHERE status='running'",
[],
)?;
Ok(deployments + jobs + runs)
}
#[allow(clippy::too_many_arguments)]
pub fn record_audit_event(
&self,
actor: &str,
operation: &str,
repository: Option<&str>,
environment: Option<&str>,
target: Option<&str>,
outcome: &str,
details: &JsonValue,
) -> Result<()> {
ensure!(
!actor.is_empty() && actor.len() <= 256,
"invalid audit actor"
);
ensure!(
!operation.is_empty() && operation.len() <= 128,
"invalid audit operation"
);
ensure!(
!outcome.is_empty() && outcome.len() <= 64,
"invalid audit outcome"
);
for (label, value, max) in [
("repository", repository, 64),
("environment", environment, 128),
("target", target, 512),
] {
if let Some(value) = value {
ensure!(
!value.is_empty() && value.len() <= max,
"invalid audit {label}"
);
}
}
let details_json = serde_json::to_string(details)?;
ensure!(details_json.len() <= 8192, "audit details are too large");
self.connection()?.execute(
"INSERT INTO audit_events(
actor, operation, repository, environment, target, outcome, details_json
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
params![
actor,
operation,
repository,
environment,
target,
outcome,
details_json
],
)?;
Ok(())
}
pub fn audit_events(
&self,
repository: Option<&str>,
limit: usize,
offset: usize,
) -> Result<Vec<AuditEvent>> {
ensure!((1..=500).contains(&limit), "audit limit must be 1..=500");
ensure!(offset <= 10_000, "audit offset must be <=10000");
if let Some(repository) = repository {
crate::config::validate_repo_name(repository)?;
}
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT id, actor, operation, repository, environment, target, outcome,
details_json, created_at
FROM audit_events
WHERE (?1 IS NULL OR repository=?1)
ORDER BY created_at DESC, id DESC
LIMIT ?2 OFFSET ?3",
)?;
Ok(statement
.query_map(
params![repository, i64::try_from(limit)?, i64::try_from(offset)?],
|row| {
Ok(AuditEvent {
id: row.get(0)?,
actor: row.get(1)?,
operation: row.get(2)?,
repository: row.get(3)?,
environment: row.get(4)?,
target: row.get(5)?,
outcome: row.get(6)?,
details_json: row.get(7)?,
created_at: row.get(8)?,
})
},
)?
.collect::<rusqlite::Result<_>>()?)
}
fn connection(&self) -> Result<MutexGuard<'_, Connection>> {
self.0
.lock()
.map_err(|_| anyhow!("database mutex poisoned"))
}
}
fn map_repository(row: &rusqlite::Row<'_>) -> rusqlite::Result<Repository> {
Ok(Repository {
id: row.get(0)?,
name: row.get(1)?,
url: row.get(2)?,
default_branch: row.get(3)?,
visibility: row.get(4)?,
owner_sub: row.get(5)?,
owner_organization_id: row.get(6)?,
trusted: row.get::<_, i64>(7)? != 0,
protect_default_branch: row.get::<_, i64>(8)? != 0,
created_at: row.get(9)?,
})
}
fn map_public_repository(row: &rusqlite::Row<'_>) -> rusqlite::Result<PublicRepository> {
Ok(PublicRepository {
id: row.get(0)?,
name: row.get(1)?,
url: row.get(2)?,
default_branch: row.get(3)?,
created_at: row.get(4)?,
run_count: row.get(5)?,
successful_runs: row.get(6)?,
last_activity_at: row.get(7)?,
})
}
fn map_run(row: &rusqlite::Row<'_>) -> rusqlite::Result<Run> {
Ok(Run {
id: row.get(0)?,
repository_id: row.get(1)?,
repository: row.get(2)?,
git_ref: row.get(3)?,
commit_sha: row.get(4)?,
trigger: row.get(5)?,
status: row.get(6)?,
error: row.get(7)?,
created_at: row.get(8)?,
started_at: row.get(9)?,
finished_at: row.get(10)?,
})
}
fn map_job(row: &rusqlite::Row<'_>) -> rusqlite::Result<Job> {
let needs_json: String = row.get(8)?;
let needs = serde_json::from_str(&needs_json).map_err(|error| {
rusqlite::Error::FromSqlConversionFailure(8, rusqlite::types::Type::Text, Box::new(error))
})?;
Ok(Job {
id: row.get(0)?,
run_id: row.get(1)?,
base_name: row.get(2)?,
name: row.get(3)?,
executor: row.get(4)?,
image: row.get(5)?,
platform: row.get(6)?,
status: row.get(7)?,
needs,
spec_json: row.get(9)?,
logs: row.get(10)?,
exit_code: row.get(11)?,
environment: row.get(12)?,
approval_required: row.get(13)?,
approved_at: row.get(14)?,
approved_by: row.get(15)?,
started_at: row.get(16)?,
finished_at: row.get(17)?,
})
}
fn map_artifact(row: &rusqlite::Row<'_>) -> rusqlite::Result<Artifact> {
Ok(Artifact {
id: row.get(0)?,
run_id: row.get(1)?,
job_id: row.get(2)?,
name: row.get(3)?,
path: row.get(4)?,
sha256: row.get(5)?,
bytes: row.get(6)?,
created_at: row.get(7)?,
})
}
fn map_deployment(row: &rusqlite::Row<'_>) -> rusqlite::Result<Deployment> {
Ok(Deployment {
id: row.get(0)?,
run_id: row.get(1)?,
job_id: row.get(2)?,
environment: row.get(3)?,
status: row.get(4)?,
artifacts_json: row.get(5)?,
created_at: row.get(6)?,
finished_at: row.get(7)?,
organization_id: row.get(8)?,
workspace_id: row.get(9)?,
installation_id: row.get(10)?,
app_id: row.get(11)?,
installation_revision: row.get(12)?,
commit_sha: row.get(13)?,
artifact_digest: row.get(14)?,
result: row.get(15)?,
rollback_of_deployment_id: row.get(16)?,
recovery_of_deployment_id: row.get(17)?,
})
}
fn map_worker(row: &rusqlite::Row<'_>) -> rusqlite::Result<Worker> {
Ok(Worker {
id: row.get(0)?,
host: row.get(1)?,
capabilities: row.get(2)?,
status: row.get(3)?,
capacity: row.get(4)?,
current_run_id: row.get(5)?,
started_at: row.get(6)?,
heartbeat_at: row.get(7)?,
completed_runs: row.get(8)?,
draining: row.get::<_, i64>(9)? != 0,
lease_expires_at: row.get(10)?,
})
}
fn prepare_database(path: &Path, create: bool) -> Result<()> {
let parent = path.parent().context("database path has no parent")?;
if create {
fs::create_dir_all(parent)?;
}
ensure!(
parent.is_dir(),
"database parent does not exist: {}",
parent.display()
);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(parent, fs::Permissions::from_mode(0o700))?;
}
match fs::symlink_metadata(path) {
Ok(metadata) => ensure!(
metadata.is_file() && !metadata.file_type().is_symlink(),
"database must be a regular non-symlink file"
),
Err(error) if error.kind() == std::io::ErrorKind::NotFound && create => {}
Err(error) => return Err(error.into()),
}
Ok(())
}
fn secure_database_file(path: &Path) -> Result<()> {
let metadata = fs::symlink_metadata(path)?;
ensure!(
metadata.is_file() && !metadata.file_type().is_symlink(),
"database must be a regular non-symlink file"
);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(path, fs::Permissions::from_mode(0o600))?;
}
Ok(())
}
pub fn load_secret(path: &Path, environment: &str) -> Result<Zeroizing<String>> {
let mut value = Zeroizing::new(String::new());
match fs::symlink_metadata(path) {
Ok(metadata) => {
ensure!(
metadata.is_file() && !metadata.file_type().is_symlink(),
"secret must be a regular non-symlink file"
);
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
ensure!(
metadata.mode() & 0o077 == 0,
"secret file permissions must be 0600"
);
}
ensure!(metadata.len() <= 1024, "secret file is too large");
let mut file = crate::fsguard::open_nofollow(path, false)?;
file.read_to_string(&mut value)?;
crate::fsguard::ensure_path_matches_file(path, &file, "secret")?;
let trimmed = value.trim_end().len();
value.truncate(trimmed);
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
value.push_str(
&env::var(environment)
.with_context(|| format!("missing {environment} or {}", path.display()))?,
);
}
Err(error) => return Err(error.into()),
}
validate_secret(&value)?;
Ok(value)
}
pub fn generate_secret(path: &Path) -> Result<()> {
ensure!(!path.exists(), "refusing to overwrite {}", path.display());
let parent = path.parent().context("secret path has no parent")?;
fs::create_dir_all(parent)?;
#[cfg(unix)]
{
use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
fs::set_permissions(parent, fs::Permissions::from_mode(0o700))?;
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(path)?;
write_secret(&mut file)?;
}
#[cfg(not(unix))]
{
let mut file = OpenOptions::new().write(true).create_new(true).open(path)?;
write_secret(&mut file)?;
}
Ok(())
}
fn write_secret(file: &mut fs::File) -> Result<()> {
let mut bytes = Zeroizing::new([0_u8; 32]);
getrandom::fill(bytes.as_mut()).context("read operating-system randomness")?;
let mut encoded = Zeroizing::new(String::with_capacity(65));
for byte in bytes.iter() {
use std::fmt::Write as _;
write!(encoded, "{byte:02x}")?;
}
encoded.push('\n');
file.write_all(encoded.as_bytes())?;
file.sync_all()?;
Ok(())
}
fn validate_secret(value: &str) -> Result<()> {
ensure!(
value.len() == 64 && value.bytes().all(|byte| byte.is_ascii_hexdigit()),
"secret must be 32 bytes encoded as hexadecimal"
);
Ok(())
}
fn decode_secret(value: &str) -> Result<Zeroizing<Vec<u8>>> {
validate_secret(value)?;
let mut bytes = Zeroizing::new(Vec::with_capacity(32));
for pair in value.as_bytes().chunks_exact(2) {
let text = std::str::from_utf8(pair)?;
bytes.push(u8::from_str_radix(text, 16)?);
}
Ok(bytes)
}
fn configure(connection: &Connection, key: &str) -> Result<()> {
apply_key(connection, key)?;
let cipher: String = connection.pragma_query_value(None, "cipher_version", |row| row.get(0))?;
ensure!(!cipher.is_empty(), "SQLCipher support is unavailable");
connection
.query_row("SELECT count(*) FROM sqlite_schema", [], |_| Ok(()))
.context("database key is incorrect or database is corrupt")?;
connection.busy_timeout(Duration::from_secs(5))?;
connection.pragma_update(None, "foreign_keys", "ON")?;
connection.pragma_update(None, "journal_mode", "WAL")?;
connection.pragma_update(None, "synchronous", "FULL")?;
connection.pragma_update(None, "secure_delete", "ON")?;
connection.set_db_config(DbConfig::SQLITE_DBCONFIG_DEFENSIVE, true)?;
connection.set_db_config(DbConfig::SQLITE_DBCONFIG_TRUSTED_SCHEMA, false)?;
connection.set_limit(Limit::SQLITE_LIMIT_LENGTH, 16 * 1024 * 1024)?;
connection.set_limit(Limit::SQLITE_LIMIT_SQL_LENGTH, 256 * 1024)?;
connection.set_limit(Limit::SQLITE_LIMIT_VARIABLE_NUMBER, 128)?;
Ok(())
}
#[allow(unsafe_code)]
fn apply_key(connection: &Connection, encoded: &str) -> Result<()> {
let key = decode_secret(encoded)?;
let length = i32::try_from(key.len())?;
// SAFETY: SQLCipher copies these bytes during the call; rusqlite owns the live handle.
let result =
unsafe { rusqlite::ffi::sqlite3_key(connection.handle(), key.as_ptr().cast(), length) };
ensure!(
result == rusqlite::ffi::SQLITE_OK,
"SQLCipher rejected the database key"
);
Ok(())
}
pub fn default_paths(root: &Path) -> (PathBuf, PathBuf, PathBuf) {
(
root.join("akurai.db"),
root.join("workspaces"),
root.join("artifacts"),
)
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct Issue {
pub id: i64,
pub repository_id: i64,
pub repository: String,
pub number: i64,
pub title: String,
pub body: String,
pub state: String,
pub author_sub: String,
pub author_display: String,
pub created_at: i64,
pub updated_at: i64,
pub closed_at: Option<i64>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct IssueComment {
pub id: i64,
pub issue_id: i64,
pub repository: String,
pub issue_number: i64,
pub body: String,
pub author_sub: String,
pub author_display: String,
pub created_at: i64,
pub updated_at: i64,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct IssueDetail {
pub issue: Issue,
pub comments: Vec<IssueComment>,
}
impl Database {
pub fn list_issues(
&self,
repository: &str,
state: Option<&str>,
limit: usize,
offset: usize,
) -> Result<Vec<Issue>> {
ensure!((1..=100).contains(&limit), "issue limit must be 1..=100");
ensure!(offset <= 10_000, "issue offset must be <= 10000");
if let Some(state) = state {
ensure!(
matches!(state, "open" | "closed"),
"issue state must be open or closed"
);
}
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT i.id, i.repository_id, r.name, i.number, i.title, i.body,
i.state, i.author_sub, i.author_display, i.created_at,
i.updated_at, i.closed_at
FROM issues i
JOIN repositories r ON r.id=i.repository_id
WHERE r.name=?1 AND (?2 IS NULL OR i.state=?2)
ORDER BY i.number DESC
LIMIT ?3 OFFSET ?4",
)?;
Ok(statement
.query_map(
params![repository, state, limit as i64, offset as i64],
map_issue,
)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn count_issues(&self, repository: &str, state: Option<&str>) -> Result<i64> {
if let Some(state) = state {
ensure!(
matches!(state, "open" | "closed"),
"issue state must be open or closed"
);
}
Ok(self.connection()?.query_row(
"SELECT COUNT(*)
FROM issues i JOIN repositories r ON r.id=i.repository_id
WHERE r.name=?1 AND (?2 IS NULL OR i.state=?2)",
params![repository, state],
|row| row.get(0),
)?)
}
pub fn issue(&self, repository: &str, number: i64) -> Result<IssueDetail> {
ensure!(number > 0, "issue number must be positive");
let connection = self.connection()?;
let issue = connection.query_row(
"SELECT i.id, i.repository_id, r.name, i.number, i.title, i.body,
i.state, i.author_sub, i.author_display, i.created_at,
i.updated_at, i.closed_at
FROM issues i
JOIN repositories r ON r.id=i.repository_id
WHERE r.name=?1 AND i.number=?2",
params![repository, number],
map_issue,
)?;
let mut comments_statement = connection.prepare(
"SELECT c.id, c.issue_id, r.name, i.number, c.body,
c.author_sub, c.author_display, c.created_at, c.updated_at
FROM issue_comments c
JOIN issues i ON i.id=c.issue_id
JOIN repositories r ON r.id=i.repository_id
WHERE c.issue_id=?1
ORDER BY c.created_at ASC, c.id ASC
LIMIT 500",
)?;
let comments = comments_statement
.query_map([issue.id], map_issue_comment)?
.collect::<rusqlite::Result<_>>()?;
Ok(IssueDetail { issue, comments })
}
pub fn create_issue(
&self,
repository: &str,
title: &str,
body: &str,
author_sub: &str,
author_display: &str,
) -> Result<Issue> {
validate_issue_fields(title, body, author_sub, author_display)?;
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let repository_id: i64 = transaction.query_row(
"SELECT id FROM repositories WHERE name=?1",
[repository],
|row| row.get(0),
)?;
let number: i64 = transaction.query_row(
"SELECT COALESCE(MAX(number), 0) + 1 FROM issues WHERE repository_id=?1",
[repository_id],
|row| row.get(0),
)?;
transaction.execute(
"INSERT INTO issues(
repository_id, number, title, body, state,
author_sub, author_display
) VALUES (?1, ?2, ?3, ?4, 'open', ?5, ?6)",
params![
repository_id,
number,
title,
body,
author_sub,
author_display
],
)?;
transaction.commit()?;
drop(connection);
self.issue(repository, number).map(|detail| detail.issue)
}
pub fn update_issue(
&self,
repository: &str,
number: i64,
title: Option<&str>,
body: Option<&str>,
state: Option<&str>,
) -> Result<Issue> {
ensure!(number > 0, "issue number must be positive");
if title.is_none() && body.is_none() && state.is_none() {
return self.issue(repository, number).map(|detail| detail.issue);
}
if let Some(title) = title {
ensure!(
!title.trim().is_empty() && title.chars().count() <= 256,
"issue title must be 1..=256 characters"
);
}
if let Some(body) = body {
ensure!(
body.chars().count() <= 65_536,
"issue body must be at most 65536 characters"
);
}
if let Some(state) = state {
ensure!(
matches!(state, "open" | "closed"),
"issue state must be open or closed"
);
}
let connection = self.connection()?;
let changed = connection.execute(
"UPDATE issues
SET title=COALESCE(?3, title),
body=COALESCE(?4, body),
state=COALESCE(?5, state),
closed_at=CASE
WHEN COALESCE(?5, state)='closed' THEN COALESCE(closed_at, unixepoch())
ELSE NULL
END,
updated_at=unixepoch()
WHERE repository_id=(SELECT id FROM repositories WHERE name=?1)
AND number=?2",
params![repository, number, title, body, state],
)?;
ensure!(changed == 1, "issue not found");
drop(connection);
self.issue(repository, number).map(|detail| detail.issue)
}
pub fn add_issue_comment(
&self,
repository: &str,
number: i64,
body: &str,
author_sub: &str,
author_display: &str,
) -> Result<IssueComment> {
ensure!(
!body.trim().is_empty() && body.chars().count() <= 65_536,
"comment body must be 1..=65536 characters"
);
ensure!(
!author_sub.is_empty() && author_sub.chars().count() <= 256,
"comment author subject must be 1..=256 characters"
);
ensure!(
!author_display.trim().is_empty() && author_display.chars().count() <= 256,
"comment author display must be 1..=256 characters"
);
let connection = self.connection()?;
let issue_id: i64 = connection.query_row(
"SELECT i.id FROM issues i
JOIN repositories r ON r.id=i.repository_id
WHERE r.name=?1 AND i.number=?2",
params![repository, number],
|row| row.get(0),
)?;
connection.execute(
"INSERT INTO issue_comments(issue_id, body, author_sub, author_display)
VALUES (?1, ?2, ?3, ?4)",
params![issue_id, body, author_sub, author_display],
)?;
let id = connection.last_insert_rowid();
connection
.query_row(
"SELECT c.id, c.issue_id, r.name, i.number, c.body,
c.author_sub, c.author_display, c.created_at, c.updated_at
FROM issue_comments c
JOIN issues i ON i.id=c.issue_id
JOIN repositories r ON r.id=i.repository_id
WHERE c.id=?1",
[id],
map_issue_comment,
)
.map_err(Into::into)
}
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct PullRequest {
pub id: i64,
pub repository_id: i64,
pub repository: String,
pub number: i64,
pub title: String,
pub body: String,
pub source_ref: String,
pub target_ref: String,
pub head_sha: String,
pub base_sha: String,
pub state: String,
pub author_sub: String,
pub author_display: String,
pub merge_commit: Option<String>,
pub created_at: i64,
pub updated_at: i64,
pub closed_at: Option<i64>,
pub merged_at: Option<i64>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct PullRequestReview {
pub id: i64,
pub pull_request_id: i64,
pub reviewer_sub: String,
pub reviewer_display: String,
pub state: String,
pub head_sha: String,
pub body: String,
pub created_at: i64,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct PullRequestDetail {
pub pull_request: PullRequest,
pub reviews: Vec<PullRequestReview>,
/// True only when at least one `approved` review exists whose `head_sha`
/// equals the PR's *current* head_sha and whose reviewer is not the PR
/// author. A push that changes head_sha silently makes every prior
/// review stale without deleting review history.
pub approved: bool,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct MergeQueueEntry {
pub id: i64,
pub repository_id: i64,
pub repository: String,
pub pull_request_id: i64,
pub pull_request_number: i64,
pub position: i64,
pub base_sha: String,
pub head_sha: String,
pub status: String,
pub run_id: Option<i64>,
pub error: Option<String>,
pub enqueued_by: String,
pub created_at: i64,
pub started_at: Option<i64>,
pub finished_at: Option<i64>,
}
impl Database {
#[allow(clippy::too_many_arguments)]
pub fn create_pull_request(
&self,
repository: &str,
title: &str,
body: &str,
source_ref: &str,
target_ref: &str,
head_sha: &str,
base_sha: &str,
author_sub: &str,
author_display: &str,
) -> Result<PullRequest> {
validate_issue_fields(title, body, author_sub, author_display)?;
ensure!(
source_ref != target_ref,
"source and target ref must differ"
);
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let repository_id: i64 = transaction.query_row(
"SELECT id FROM repositories WHERE name=?1",
[repository],
|row| row.get(0),
)?;
let number: i64 = transaction.query_row(
"SELECT COALESCE(MAX(number), 0) + 1 FROM pull_requests WHERE repository_id=?1",
[repository_id],
|row| row.get(0),
)?;
transaction.execute(
"INSERT INTO pull_requests(
repository_id, number, title, body, source_ref, target_ref,
head_sha, base_sha, state, author_sub, author_display
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'open', ?9, ?10)",
params![
repository_id,
number,
title,
body,
source_ref,
target_ref,
head_sha,
base_sha,
author_sub,
author_display
],
)?;
transaction.commit()?;
drop(connection);
self.pull_request(repository, number)
.map(|detail| detail.pull_request)
}
pub fn list_pull_requests(
&self,
repository: &str,
state: Option<&str>,
limit: usize,
offset: usize,
) -> Result<Vec<PullRequest>> {
ensure!(
(1..=100).contains(&limit),
"pull request limit must be 1..=100"
);
ensure!(offset <= 10_000, "pull request offset must be <= 10000");
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT p.id, p.repository_id, r.name, p.number, p.title, p.body,
p.source_ref, p.target_ref, p.head_sha, p.base_sha, p.state,
p.author_sub, p.author_display, p.merge_commit,
p.created_at, p.updated_at, p.closed_at, p.merged_at
FROM pull_requests p
JOIN repositories r ON r.id=p.repository_id
WHERE r.name=?1 AND (?2 IS NULL OR p.state=?2)
ORDER BY p.number DESC
LIMIT ?3 OFFSET ?4",
)?;
Ok(statement
.query_map(
params![repository, state, limit as i64, offset as i64],
map_pull_request,
)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn count_pull_requests(&self, repository: &str, state: Option<&str>) -> Result<i64> {
Ok(self.connection()?.query_row(
"SELECT COUNT(*)
FROM pull_requests p JOIN repositories r ON r.id=p.repository_id
WHERE r.name=?1 AND (?2 IS NULL OR p.state=?2)",
params![repository, state],
|row| row.get(0),
)?)
}
pub fn pull_request(&self, repository: &str, number: i64) -> Result<PullRequestDetail> {
ensure!(number > 0, "pull request number must be positive");
let connection = self.connection()?;
let pull_request = connection.query_row(
"SELECT p.id, p.repository_id, r.name, p.number, p.title, p.body,
p.source_ref, p.target_ref, p.head_sha, p.base_sha, p.state,
p.author_sub, p.author_display, p.merge_commit,
p.created_at, p.updated_at, p.closed_at, p.merged_at
FROM pull_requests p
JOIN repositories r ON r.id=p.repository_id
WHERE r.name=?1 AND p.number=?2",
params![repository, number],
map_pull_request,
)?;
let mut statement = connection.prepare(
"SELECT id, pull_request_id, reviewer_sub, reviewer_display, state,
head_sha, body, created_at
FROM pull_request_reviews
WHERE pull_request_id=?1
ORDER BY created_at ASC, id ASC
LIMIT 500",
)?;
let reviews: Vec<PullRequestReview> = statement
.query_map([pull_request.id], map_pull_request_review)?
.collect::<rusqlite::Result<_>>()?;
// A reviewer's decision is their MOST RECENT review at the PR's
// current head_sha (so approve-then-request-changes by the same
// reviewer correctly blocks, and a later approval after changes
// were requested correctly unblocks). Older head_shas are always
// stale regardless of state, per stale-approval invalidation.
// Reviews come back ordered by created_at ASC, id ASC, so a plain
// overwrite on each matching review always leaves the latest one.
let mut latest_at_head: std::collections::HashMap<&str, &PullRequestReview> =
std::collections::HashMap::new();
for review in &reviews {
if review.head_sha == pull_request.head_sha
&& review.reviewer_sub != pull_request.author_sub
{
latest_at_head.insert(review.reviewer_sub.as_str(), review);
}
}
let approved = !latest_at_head.is_empty()
&& latest_at_head
.values()
.all(|review| review.state != "changes_requested")
&& latest_at_head
.values()
.any(|review| review.state == "approved");
Ok(PullRequestDetail {
pull_request,
reviews,
approved,
})
}
/// Server-side counterpart of `update_pull_request_head`, used after a
/// real push or mirror sync: resolve open PRs whose `source_ref`
/// matches `branch` in this repository and stamp the SHA the caller
/// already resolved from the hosted mirror (never a client-supplied
/// value). No-op if no open PR uses this branch as its source.
pub fn update_pull_requests_head_for_source_ref(
&self,
repository: &str,
branch: &str,
head_sha: &str,
) -> Result<()> {
config::validate_commit(head_sha)?;
let source_ref = format!("refs/heads/{branch}");
let connection = self.connection()?;
connection.execute(
"UPDATE pull_requests
SET head_sha=?3, updated_at=unixepoch()
WHERE repository_id=(SELECT id FROM repositories WHERE name=?1)
AND (source_ref=?2 OR source_ref=?4) AND state='open' AND head_sha != ?3",
params![repository, source_ref, head_sha, branch],
)?;
Ok(())
}
/// Record a new head_sha for a pull request after a push to its source
/// ref. Existing reviews are never deleted or rewritten; because their
/// `head_sha` no longer matches, `pull_request()` stops counting them
/// toward approval the instant this call lands (stale-approval
/// invalidation without a separate revocation pass).
pub fn update_pull_request_head(
&self,
repository: &str,
number: i64,
head_sha: &str,
) -> Result<PullRequest> {
config::validate_commit(head_sha)?;
let connection = self.connection()?;
let changed = connection.execute(
"UPDATE pull_requests
SET head_sha=?3, updated_at=unixepoch()
WHERE repository_id=(SELECT id FROM repositories WHERE name=?1)
AND number=?2 AND state='open'",
params![repository, number, head_sha],
)?;
ensure!(changed == 1, "pull request not found or not open");
drop(connection);
self.pull_request(repository, number)
.map(|detail| detail.pull_request)
}
/// Add a review bound to the pull request's *current* head_sha. A
/// reviewer may not review their own pull request: a shared static
/// admin identity therefore cannot satisfy independent review, because
/// `reviewer_sub` for the static bearer principal never equals a human
/// `author_sub`, but self-authored PRs are still explicitly rejected.
pub fn add_pull_request_review(
&self,
repository: &str,
number: i64,
reviewer_sub: &str,
reviewer_display: &str,
state: &str,
body: &str,
) -> Result<PullRequestReview> {
ensure!(
matches!(state, "approved" | "changes_requested" | "commented"),
"review state must be approved, changes_requested, or commented"
);
ensure!(
!reviewer_sub.is_empty() && reviewer_sub.chars().count() <= 256,
"reviewer subject must be 1..=256 characters"
);
ensure!(
!reviewer_display.trim().is_empty() && reviewer_display.chars().count() <= 256,
"reviewer display must be 1..=256 characters"
);
ensure!(
body.chars().count() <= 65_536,
"review body must be at most 65536 characters"
);
let detail = self.pull_request(repository, number)?;
ensure!(
detail.pull_request.author_sub != reviewer_sub,
"reviewer must be independent of the pull request author"
);
let head_sha = detail.pull_request.head_sha.clone();
let connection = self.connection()?;
connection.execute(
"INSERT INTO pull_request_reviews(
pull_request_id, reviewer_sub, reviewer_display, state, head_sha, body
) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
params![
detail.pull_request.id,
reviewer_sub,
reviewer_display,
state,
head_sha,
body
],
)?;
let id = connection.last_insert_rowid();
connection
.query_row(
"SELECT id, pull_request_id, reviewer_sub, reviewer_display, state,
head_sha, body, created_at
FROM pull_request_reviews WHERE id=?1",
[id],
map_pull_request_review,
)
.map_err(Into::into)
}
pub fn close_pull_request(&self, repository: &str, number: i64) -> Result<PullRequest> {
let connection = self.connection()?;
let changed = connection.execute(
"UPDATE pull_requests
SET state='closed', closed_at=unixepoch(), updated_at=unixepoch()
WHERE repository_id=(SELECT id FROM repositories WHERE name=?1)
AND number=?2 AND state IN ('open', 'queued')",
params![repository, number],
)?;
ensure!(changed == 1, "pull request not found or already closed");
drop(connection);
self.pull_request(repository, number)
.map(|detail| detail.pull_request)
}
/// Enqueue a pull request into its repository's serialized merge queue.
/// Position is assigned as `max(position)+1` inside the same immediate
/// transaction that flips the PR to `queued`, so two concurrent enqueues
/// for the same repository can never receive the same position. The
/// independent-approval check is re-verified inside this same
/// transaction (not just by the HTTP handler beforehand), closing the
/// race where a `changes_requested` review or a head-changing push
/// lands between an earlier approval check and this call.
pub fn enqueue_merge(
&self,
repository: &str,
number: i64,
enqueued_by: &str,
) -> Result<MergeQueueEntry> {
ensure!(
!enqueued_by.is_empty() && enqueued_by.chars().count() <= 256,
"enqueued_by must be 1..=256 characters"
);
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
let (pr_id, repository_id, base_sha, head_sha, state, author_sub): (
i64,
i64,
String,
String,
String,
String,
) = transaction.query_row(
"SELECT p.id, p.repository_id, p.base_sha, p.head_sha, p.state, p.author_sub
FROM pull_requests p
JOIN repositories r ON r.id=p.repository_id
WHERE r.name=?1 AND p.number=?2",
params![repository, number],
|row| {
Ok((
row.get(0)?,
row.get(1)?,
row.get(2)?,
row.get(3)?,
row.get(4)?,
row.get(5)?,
))
},
)?;
ensure!(
state == "open",
"pull request must be open to enqueue: was {state}"
);
// Each independent reviewer's decision is their most-recent review
// at the PR's *current* head_sha (mirrors PullRequestDetail::approved
// in pull_request()); this must hold inside the same transaction as
// the enqueue, not just as an earlier, separately-committed read.
let latest_reviews_at_head: Vec<(String, String)> = {
let mut statement = transaction.prepare(
"SELECT reviewer_sub, state FROM pull_request_reviews
WHERE pull_request_id=?1 AND head_sha=?2 AND reviewer_sub != ?3
AND id = (
SELECT id FROM pull_request_reviews AS latest
WHERE latest.pull_request_id = pull_request_reviews.pull_request_id
AND latest.reviewer_sub = pull_request_reviews.reviewer_sub
AND latest.head_sha = pull_request_reviews.head_sha
ORDER BY created_at DESC, id DESC LIMIT 1
)",
)?;
statement
.query_map(params![pr_id, head_sha, author_sub], |row| {
Ok((row.get(0)?, row.get(1)?))
})?
.collect::<rusqlite::Result<_>>()?
};
let approved = !latest_reviews_at_head.is_empty()
&& latest_reviews_at_head
.iter()
.all(|(_, state)| state != "changes_requested")
&& latest_reviews_at_head
.iter()
.any(|(_, state)| state == "approved");
ensure!(
approved,
"pull request needs an independent approval at its current head_sha before it can be queued"
);
let position: i64 = transaction.query_row(
"SELECT COALESCE(MAX(position), 0) + 1 FROM merge_queue_entries WHERE repository_id=?1",
[repository_id],
|row| row.get(0),
)?;
transaction.execute(
"UPDATE pull_requests SET state='queued', updated_at=unixepoch() WHERE id=?1",
[pr_id],
)?;
transaction.execute(
"INSERT INTO merge_queue_entries(
repository_id, pull_request_id, position, base_sha, head_sha,
status, enqueued_by
) VALUES (?1, ?2, ?3, ?4, ?5, 'queued', ?6)",
params![
repository_id,
pr_id,
position,
base_sha,
head_sha,
enqueued_by
],
)?;
let id = transaction.last_insert_rowid();
let entry = transaction
.query_row(
"SELECT m.id, m.repository_id, r.name, m.pull_request_id, p.number,
m.position, m.base_sha, m.head_sha, m.status, m.run_id, m.error,
m.enqueued_by, m.created_at, m.started_at, m.finished_at
FROM merge_queue_entries m
JOIN repositories r ON r.id=m.repository_id
JOIN pull_requests p ON p.id=m.pull_request_id
WHERE m.id=?1",
[id],
map_merge_queue_entry,
)
.map_err(anyhow::Error::from)?;
transaction.commit()?;
Ok(entry)
}
/// Distinct repository ids that currently have at least one queued or
/// running merge-queue entry — the merge-queue worker's outer loop
/// iterates exactly this set instead of scanning every repository.
pub fn repositories_with_queued_merge(&self) -> Result<Vec<i64>> {
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT DISTINCT repository_id FROM merge_queue_entries
WHERE status IN ('queued', 'running')",
)?;
Ok(statement
.query_map([], |row| row.get(0))?
.collect::<rusqlite::Result<_>>()?)
}
/// The single entry a repository's queue worker should act on next:
/// lowest position with status `queued`, only when no earlier entry for
/// the same repository is still `running` (strict FIFO serialization).
pub fn next_queued_merge(&self, repository_id: i64) -> Result<Option<MergeQueueEntry>> {
let connection = self.connection()?;
let running: i64 = connection.query_row(
"SELECT COUNT(*) FROM merge_queue_entries WHERE repository_id=?1 AND status='running'",
[repository_id],
|row| row.get(0),
)?;
if running > 0 {
return Ok(None);
}
connection
.query_row(
"SELECT m.id, m.repository_id, r.name, m.pull_request_id, p.number,
m.position, m.base_sha, m.head_sha, m.status, m.run_id, m.error,
m.enqueued_by, m.created_at, m.started_at, m.finished_at
FROM merge_queue_entries m
JOIN repositories r ON r.id=m.repository_id
JOIN pull_requests p ON p.id=m.pull_request_id
WHERE m.repository_id=?1 AND m.status='queued'
ORDER BY m.position ASC LIMIT 1",
[repository_id],
map_merge_queue_entry,
)
.optional()
.map_err(Into::into)
}
/// Atomically claim a queued entry: `queued` -> `running`. Returns false
/// if another worker already claimed it (compare-and-swap on status).
pub fn claim_merge_queue_entry(&self, id: i64) -> Result<bool> {
Ok(self.connection()?.execute(
"UPDATE merge_queue_entries SET status='running', started_at=unixepoch()
WHERE id=?1 AND status='queued'
AND NOT EXISTS (
SELECT 1 FROM merge_queue_entries AS earlier
WHERE earlier.repository_id=merge_queue_entries.repository_id
AND (earlier.status='running' OR
(earlier.status='queued' AND earlier.position<merge_queue_entries.position))
)",
[id],
)? == 1)
}
pub fn finish_merge_queue_entry(
&self,
id: i64,
status: &str,
run_id: Option<i64>,
error: Option<&str>,
) -> Result<()> {
ensure!(
matches!(status, "succeeded" | "failed" | "canceled"),
"merge queue terminal status must be succeeded, failed, or canceled"
);
ensure!(
self.connection()?.execute(
"UPDATE merge_queue_entries
SET status=?2, run_id=?3, error=?4, finished_at=unixepoch()
WHERE id=?1 AND status='running'",
params![id, status, run_id, error],
)? == 1,
"merge queue entry {id} was not running"
);
Ok(())
}
/// Crash recovery: any entry left `running` when the process restarts
/// (worker died mid-merge) is reset to `queued` so the next tick retries
/// it, preserving strict per-repository ordering instead of skipping it.
pub fn attach_merge_queue_run(&self, id: i64, run_id: i64) -> Result<()> {
ensure!(
self.connection()?.execute(
"UPDATE merge_queue_entries SET run_id=?2 WHERE id=?1 AND status='running'",
params![id, run_id]
)? == 1,
"merge queue entry was not running"
);
Ok(())
}
pub fn recover_merge_queue(&self) -> Result<usize> {
Ok(self.connection()?.execute(
"UPDATE merge_queue_entries SET status='queued', started_at=NULL
WHERE status='running'",
[],
)?)
}
pub fn list_merge_queue(&self, repository: &str) -> Result<Vec<MergeQueueEntry>> {
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT m.id, m.repository_id, r.name, m.pull_request_id, p.number,
m.position, m.base_sha, m.head_sha, m.status, m.run_id, m.error,
m.enqueued_by, m.created_at, m.started_at, m.finished_at
FROM merge_queue_entries m
JOIN repositories r ON r.id=m.repository_id
JOIN pull_requests p ON p.id=m.pull_request_id
WHERE r.name=?1 AND m.status IN ('queued', 'running')
ORDER BY m.position ASC",
)?;
Ok(statement
.query_map([repository], map_merge_queue_entry)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn merge_queue_entry(&self, id: i64) -> Result<MergeQueueEntry> {
self.connection()?
.query_row(
"SELECT m.id, m.repository_id, r.name, m.pull_request_id, p.number,
m.position, m.base_sha, m.head_sha, m.status, m.run_id, m.error,
m.enqueued_by, m.created_at, m.started_at, m.finished_at
FROM merge_queue_entries m
JOIN repositories r ON r.id=m.repository_id
JOIN pull_requests p ON p.id=m.pull_request_id
WHERE m.id=?1",
[id],
map_merge_queue_entry,
)
.map_err(Into::into)
}
/// Record a successful merge: PR -> merged with its commit, and the
/// queue entry -> succeeded, in one immediate transaction so a crash
/// between the two updates cannot happen.
pub fn finish_merged_pull_request(
&self,
queue_entry_id: i64,
pull_request_id: i64,
merge_commit: &str,
run_id: Option<i64>,
) -> Result<()> {
config::validate_commit(merge_commit)?;
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
ensure!(
transaction.execute(
"UPDATE pull_requests
SET state='merged', merge_commit=?2, merged_at=unixepoch(), updated_at=unixepoch()
WHERE id=?1 AND state='queued'",
params![pull_request_id, merge_commit],
)? == 1,
"pull request {pull_request_id} was not queued"
);
ensure!(
transaction.execute(
"UPDATE merge_queue_entries
SET status='succeeded', run_id=?2, finished_at=unixepoch()
WHERE id=?1 AND status='running'",
params![queue_entry_id, run_id],
)? == 1,
"merge queue entry {queue_entry_id} was not running"
);
transaction.commit()?;
Ok(())
}
/// Failure path: PR returns to `open` (so it can be fixed and
/// re-enqueued) and the queue entry records the failure reason.
pub fn fail_merge_queue_pull_request(
&self,
queue_entry_id: i64,
pull_request_id: i64,
error: &str,
run_id: Option<i64>,
) -> Result<()> {
let mut connection = self.connection()?;
let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
transaction.execute(
"UPDATE pull_requests SET state='open', updated_at=unixepoch()
WHERE id=?1 AND state='queued'",
[pull_request_id],
)?;
ensure!(
transaction.execute(
"UPDATE merge_queue_entries
SET status='failed', run_id=?2, error=?3, finished_at=unixepoch()
WHERE id=?1 AND status='running'",
params![queue_entry_id, run_id, error],
)? == 1,
"merge queue entry {queue_entry_id} was not running"
);
transaction.commit()?;
Ok(())
}
}
fn map_pull_request(row: &rusqlite::Row<'_>) -> rusqlite::Result<PullRequest> {
Ok(PullRequest {
id: row.get(0)?,
repository_id: row.get(1)?,
repository: row.get(2)?,
number: row.get(3)?,
title: row.get(4)?,
body: row.get(5)?,
source_ref: row.get(6)?,
target_ref: row.get(7)?,
head_sha: row.get(8)?,
base_sha: row.get(9)?,
state: row.get(10)?,
author_sub: row.get(11)?,
author_display: row.get(12)?,
merge_commit: row.get(13)?,
created_at: row.get(14)?,
updated_at: row.get(15)?,
closed_at: row.get(16)?,
merged_at: row.get(17)?,
})
}
fn map_pull_request_review(row: &rusqlite::Row<'_>) -> rusqlite::Result<PullRequestReview> {
Ok(PullRequestReview {
id: row.get(0)?,
pull_request_id: row.get(1)?,
reviewer_sub: row.get(2)?,
reviewer_display: row.get(3)?,
state: row.get(4)?,
head_sha: row.get(5)?,
body: row.get(6)?,
created_at: row.get(7)?,
})
}
fn map_merge_queue_entry(row: &rusqlite::Row<'_>) -> rusqlite::Result<MergeQueueEntry> {
Ok(MergeQueueEntry {
id: row.get(0)?,
repository_id: row.get(1)?,
repository: row.get(2)?,
pull_request_id: row.get(3)?,
pull_request_number: row.get(4)?,
position: row.get(5)?,
base_sha: row.get(6)?,
head_sha: row.get(7)?,
status: row.get(8)?,
run_id: row.get(9)?,
error: row.get(10)?,
enqueued_by: row.get(11)?,
created_at: row.get(12)?,
started_at: row.get(13)?,
finished_at: row.get(14)?,
})
}
fn validate_issue_fields(
title: &str,
body: &str,
author_sub: &str,
author_display: &str,
) -> Result<()> {
ensure!(
!title.trim().is_empty() && title.chars().count() <= 256,
"issue title must be 1..=256 characters"
);
ensure!(
body.chars().count() <= 65_536,
"issue body must be at most 65536 characters"
);
ensure!(
!author_sub.is_empty() && author_sub.chars().count() <= 256,
"issue author subject must be 1..=256 characters"
);
ensure!(
!author_display.trim().is_empty() && author_display.chars().count() <= 256,
"issue author display must be 1..=256 characters"
);
Ok(())
}
fn map_issue(row: &rusqlite::Row<'_>) -> rusqlite::Result<Issue> {
Ok(Issue {
id: row.get(0)?,
repository_id: row.get(1)?,
repository: row.get(2)?,
number: row.get(3)?,
title: row.get(4)?,
body: row.get(5)?,
state: row.get(6)?,
author_sub: row.get(7)?,
author_display: row.get(8)?,
created_at: row.get(9)?,
updated_at: row.get(10)?,
closed_at: row.get(11)?,
})
}
fn map_issue_comment(row: &rusqlite::Row<'_>) -> rusqlite::Result<IssueComment> {
Ok(IssueComment {
id: row.get(0)?,
issue_id: row.get(1)?,
repository: row.get(2)?,
issue_number: row.get(3)?,
body: row.get(4)?,
author_sub: row.get(5)?,
author_display: row.get(6)?,
created_at: row.get(7)?,
updated_at: row.get(8)?,
})
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
const KEY: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
const KEY2: &str = "100102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
fn open_db(dir: &tempfile::TempDir) -> Result<Database> {
let path = dir.path().join("akurai.db");
let db = Database::open(&path, KEY, true)?;
db.migrate()?;
Ok(db)
}
// ── migrate idempotent ──
#[test]
fn migrate_is_idempotent() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let count = db.migrate()?;
assert_eq!(count, 0, "second migrate should apply zero migrations");
db.validate_schema()?;
// schema still usable
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
assert_eq!(repo.name, "app");
Ok(())
}
// ── wrong key ──
#[test]
fn open_with_wrong_key_fails() -> Result<()> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("akurai.db");
// create with correct key
let _db = Database::open(&path, KEY, true)?;
// reopen with wrong key
match Database::open(&path, KEY2, false) {
Ok(_) => panic!("opening with wrong key should have failed"),
Err(err) => {
let msg = format!("{err:#}");
assert!(
msg.contains("database key is incorrect") || msg.contains("corrupt"),
"expected key-error message, got: {msg}"
);
}
}
Ok(())
}
// ── secret generate → load round-trip ──
#[test]
fn secret_generate_then_load_round_trips() -> Result<()> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("secret.key");
generate_secret(&path)?;
let loaded = load_secret(&path, "TEST_SECRET_ENV")?;
// load_secret returns the raw hex string; validate it
validate_secret(&loaded)?;
assert_eq!(loaded.len(), 64);
Ok(())
}
#[test]
fn load_secret_missing_file_and_env_errors() -> Result<()> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("nonexistent.key");
let err = load_secret(&path, "NONEXISTENT_ENV_VAR_AKBUILD_TEST")
.expect_err("loading nonexistent secret should fail");
let msg = format!("{err:#}");
assert!(
msg.contains("missing NONEXISTENT_ENV_VAR_AKBUILD_TEST")
|| msg.contains(&path.display().to_string()),
"expected missing-env-or-file message, got: {msg}"
);
Ok(())
}
#[test]
fn generate_secret_refuses_overwrite() -> Result<()> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("secret.key");
generate_secret(&path)?;
let err = generate_secret(&path).expect_err("generating over existing secret should fail");
let msg = format!("{err:#}");
assert!(msg.contains("refusing to overwrite"), "got: {msg}");
Ok(())
}
// ── repositories ──
#[test]
fn repository_add_then_list() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.add_repository("app", "https://example.com/app.git", "main")?;
db.add_repository("lib", "https://example.com/lib.git", "develop")?;
let repos = db.repositories()?;
assert_eq!(repos.len(), 2);
// sorted by name
assert_eq!(repos[0].name, "app");
assert_eq!(repos[1].name, "lib");
Ok(())
}
#[test]
fn repository_access_preserves_trusted_admin_rows_and_scopes_owned_rows() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let trusted = db.add_repository("trusted", "https://example.com/trusted.git", "main")?;
assert!(trusted.trusted);
assert!(trusted.owner_sub.is_none());
assert_eq!(
db.repository_role("trusted", "member-a", Some("org-a"))?,
None
);
let owned = db.add_repository_owned(
"owned",
"https://example.com/owned.git",
"main",
"owner-a",
Some("org-a"),
)?;
assert!(!owned.trusted);
assert_eq!(owned.owner_sub.as_deref(), Some("owner-a"));
assert_eq!(
db.repository_role("owned", "owner-a", Some("org-a"))?,
Some(RepositoryRole::Maintain)
);
assert_eq!(
db.repository_role("owned", "member-a", Some("org-a"))?,
Some(RepositoryRole::Read)
);
assert_eq!(
db.repository_role("owned", "member-a", Some("org-b"))?,
None
);
db.set_repository_collaborator("owned", "member-b", Some("org-b"), RepositoryRole::Write)?;
assert_eq!(
db.repository_role("owned", "member-b", Some("org-b"))?,
Some(RepositoryRole::Write)
);
assert_eq!(db.repository_collaborators("owned")?.len(), 1);
Ok(())
}
#[test]
fn repository_add_upserts_on_duplicate_name() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.add_repository("app", "https://example.com/app.git", "main")?;
// same name, different url — upserts
let updated = db.add_repository("app", "https://example.com/app2.git", "develop")?;
assert_eq!(updated.url, "https://example.com/app2.git");
assert_eq!(updated.default_branch, "develop");
let repos = db.repositories()?;
assert_eq!(repos.len(), 1, "upsert should not increase count");
Ok(())
}
#[test]
fn repository_rename_works() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.add_repository("old", "https://example.com/old.git", "main")?;
let renamed = db.rename_repository("old", "new")?;
assert_eq!(renamed.name, "new");
// old name gone
let err = db
.repository("old")
.expect_err("old name should not exist after rename");
let msg = format!("{err:#}");
assert!(msg.contains("unknown repository"), "got: {msg}");
// new name works
let found = db.repository("new")?;
assert_eq!(found.url, "https://example.com/old.git");
Ok(())
}
#[test]
fn repository_rename_to_existing_name_fails() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.add_repository("alpha", "https://example.com/alpha.git", "main")?;
db.add_repository("beta", "https://example.com/beta.git", "main")?;
let err = db
.rename_repository("alpha", "beta")
.expect_err("renaming to existing name should fail");
let msg = format!("{err:#}");
assert!(
msg.contains("UNIQUE constraint") || msg.contains("constraint"),
"expected constraint violation, got: {msg}"
);
Ok(())
}
#[test]
fn repository_rename_invalid_name_fails() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.add_repository("valid", "https://example.com/valid.git", "main")?;
// validate_repo_name rejects empty and names with bad chars
let err = db
.rename_repository("valid", "")
.expect_err("empty name should be rejected");
let msg = format!("{err:#}");
assert!(msg.contains("invalid repository name"), "got: {msg}");
Ok(())
}
#[test]
fn repository_unknown_name_errors() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let err = db
.repository("nonexistent")
.expect_err("unknown repository should error");
let msg = format!("{err:#}");
assert!(msg.contains("unknown repository"), "got: {msg}");
Ok(())
}
#[test]
fn remove_repository_deletes_and_returns_it() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.add_repository("gone", "https://example.com/gone.git", "main")?;
let removed = db.remove_repository("gone")?;
assert_eq!(removed.name, "gone");
let err = db
.repository("gone")
.expect_err("removed repository should be gone");
assert!(format!("{err:#}").contains("unknown repository"));
Ok(())
}
#[test]
fn remove_repository_unknown_errors() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let err = db
.remove_repository("ghost")
.expect_err("removing unknown repository should error");
assert!(format!("{err:#}").contains("unknown repository"));
Ok(())
}
#[test]
fn remove_repository_cascades_runs() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("casc", "https://example.com/casc.git", "main")?;
let run_id = db.create_run(
repo.id,
"main",
Some("deadbeefdeadbeefdeadbeefdeadbeefdeadbeef"),
"manual",
)?;
assert!(db.run(run_id).is_ok(), "run should exist before removal");
db.remove_repository("casc")?;
assert!(
db.run(run_id).is_err(),
"run row must cascade-delete when its repository is removed"
);
Ok(())
}
#[test]
fn repository_query_by_visibility_and_search() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.add_repository("alpha", "https://gh.com/alpha.git", "main")?;
db.add_repository("beta", "https://gh.com/beta.git", "main")?;
db.set_repository_visibility("beta", "public")?;
let public = db.query_repositories(&RepositoryQuery {
visibility: Some("public".into()),
..RepositoryQuery::default()
})?;
assert_eq!(public.len(), 1);
assert_eq!(public[0].name, "beta");
let by_name = db.query_repositories(&RepositoryQuery {
search: Some("alpha".into()),
..RepositoryQuery::default()
})?;
assert_eq!(by_name.len(), 1);
assert_eq!(by_name[0].name, "alpha");
Ok(())
}
// ── runs ──
#[test]
fn run_insert_then_query_by_repo() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let alpha = db.add_repository("alpha", "https://example.com/alpha.git", "main")?;
let beta = db.add_repository("beta", "https://example.com/beta.git", "main")?;
let r1 = db.create_run(alpha.id, "main", None, "manual")?;
let _r2 = db.create_run(beta.id, "release", None, "webhook")?;
let r3 = db.create_run(alpha.id, "feature/x", None, "manual")?;
// by repo filter
let alpha_runs = db.query_runs(&RunQuery {
repository: Some("alpha".into()),
limit: 20,
..RunQuery::default()
})?;
assert_eq!(alpha_runs.len(), 2);
// ordered by id DESC
assert_eq!(alpha_runs[0].id, r3);
assert_eq!(alpha_runs[1].id, r1);
// by status filter
db.finish_run(r1, "succeeded", None)?;
let succeeded = db.query_runs(&RunQuery {
statuses: vec!["succeeded".into()],
limit: 20,
..RunQuery::default()
})?;
assert_eq!(succeeded.len(), 1);
assert_eq!(succeeded[0].id, r1);
Ok(())
}
#[test]
fn run_query_limit_respected() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
for i in 0..10 {
db.create_run(repo.id, &format!("ref{i}"), None, "manual")?;
}
let limited = db.query_runs(&RunQuery {
limit: 3,
..RunQuery::default()
})?;
assert_eq!(limited.len(), 3);
// first 3 by DESC id
let ids: Vec<i64> = limited.iter().map(|r| r.id).collect();
assert!(ids[0] > ids[1] && ids[1] > ids[2], "not DESC ordered");
let oldest = db.query_runs(&RunQuery {
limit: 3,
offset: 9,
..RunQuery::default()
})?;
assert_eq!(oldest.len(), 1);
assert_eq!(db.count_runs(&RunQuery::default())?, 10);
let cursor = db.max_run_id(&RunQuery::default())?.expect("run cursor");
db.create_run(repo.id, "new-after-snapshot", None, "manual")?;
let snapshot = RunQuery {
before_id: Some(cursor),
..RunQuery::default()
};
assert_eq!(db.count_runs(&snapshot)?, 10);
assert!(db.query_runs(&snapshot)?.iter().all(|run| run.id <= cursor));
assert_eq!(db.count_runs(&RunQuery::default())?, 11);
Ok(())
}
#[test]
fn run_count_uses_exactly_the_listing_predicates_without_pagination() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let alpha = db.add_repository("alpha", "https://example.com/alpha.git", "main")?;
let beta = db.add_repository("beta", "https://example.com/beta.git", "main")?;
let first = db.create_run(
alpha.id,
"main",
Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
"manual",
)?;
db.finish_run(first, "succeeded", None)?;
let excluded = db.create_run(
alpha.id,
"feature",
Some("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
"webhook",
)?;
db.finish_run(excluded, "failed", Some("boom"))?;
let last = db.create_run(
beta.id,
"main",
Some("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
"retry",
)?;
db.finish_run(last, "canceled", None)?;
let query = RunQuery {
repository: Some("alpha,beta".into()),
statuses: vec!["succeeded".into(), "canceled".into()],
git_ref: Some("main".into()),
triggers: vec!["manual".into(), "retry".into()],
search: Some("aaa".into()),
before_id: Some(last),
limit: 1,
offset: 1,
public_only: false,
};
let page = db.query_runs(&query)?;
assert_eq!(page.len(), 1);
assert_eq!(page[0].id, first);
assert_eq!(db.count_runs(&query)?, 2);
assert_eq!(db.max_run_id(&query)?, Some(last));
Ok(())
}
#[test]
fn create_or_get_active_run_deduplicates_only_active_runs() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repository = db.add_repository("app", "https://example.com/app.git", "main")?;
let commit = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
let (first, deduplicated) =
db.create_or_get_active_run(repository.id, "main", commit, "manual")?;
assert!(!deduplicated);
let (second, deduplicated) =
db.create_or_get_active_run(repository.id, "main", commit, "webhook")?;
assert!(deduplicated);
assert_eq!(first, second);
db.finish_run(first, "succeeded", None)?;
let (next, deduplicated) =
db.create_or_get_active_run(repository.id, "main", commit, "manual")?;
assert!(!deduplicated);
assert_ne!(first, next);
Ok(())
}
#[test]
fn create_or_get_pending_run_deduplicates_unresolved_runs() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repository = db.add_repository("app", "https://example.com/app.git", "main")?;
let (first, deduplicated) =
db.create_or_get_pending_run(repository.id, "main", "webhook")?;
assert!(!deduplicated);
let (second, deduplicated) =
db.create_or_get_pending_run(repository.id, "main", "webhook")?;
assert!(deduplicated);
assert_eq!(first, second);
// a different ref is not deduplicated
let (other, deduplicated) =
db.create_or_get_pending_run(repository.id, "dev", "webhook")?;
assert!(!deduplicated);
assert_ne!(first, other);
// once the commit resolves, new deliveries create a fresh run
db.set_run_commit(first, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")?;
let (next, deduplicated) =
db.create_or_get_pending_run(repository.id, "main", "webhook")?;
assert!(!deduplicated);
assert_ne!(first, next);
Ok(())
}
#[test]
fn supersede_waiting_runs_cancels_only_older_waiting_same_ref() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repository = db.add_repository("app", "https://example.com/app.git", "main")?;
let waiting = db.create_run(repository.id, "main", None, "webhook")?;
db.finish_run(waiting, "waiting", None)?;
let other_ref = db.create_run(repository.id, "dev", None, "webhook")?;
db.finish_run(other_ref, "waiting", None)?;
let succeeded = db.create_run(repository.id, "main", None, "webhook")?;
db.finish_run(succeeded, "succeeded", None)?;
let fresh = db.create_run(repository.id, "main", None, "webhook")?;
let canceled = db.supersede_waiting_runs(repository.id, "main", fresh)?;
assert_eq!(canceled, vec![waiting]);
assert_eq!(db.run(waiting)?.status, "canceled");
assert_eq!(db.run(other_ref)?.status, "waiting");
assert_eq!(db.run(succeeded)?.status, "succeeded");
assert_eq!(db.run(fresh)?.status, "queued");
Ok(())
}
#[test]
fn run_claim_optimistic_concurrency() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
// first claim succeeds
let claimed = db.claim_run(run_id)?;
assert!(claimed, "first claim should succeed");
assert_eq!(db.run(run_id)?.status, "running");
// second claim fails (already claimed)
let again = db.claim_run(run_id)?;
assert!(!again, "second claim of same run should fail");
Ok(())
}
#[test]
fn run_finish_transitions() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
db.finish_run(run_id, "failed", Some("build error"))?;
let run = db.run(run_id)?;
assert_eq!(run.status, "failed");
assert_eq!(run.error.as_deref(), Some("build error"));
assert!(run.finished_at.is_some(), "finished_at should be set");
Ok(())
}
#[test]
fn cancel_run_cancels_queued_run_and_jobs() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
let spec: crate::config::JobSpec = serde_json::from_value(serde_json::json!({
"base_name": "build",
"name": "build",
"needs": [],
"executor": "native",
"image": null,
"shell": null,
"command": "true",
"matrix": {},
"artifacts": [],
"cache": [],
"network": false,
"secrets": [],
"environment": null,
"approval": false,
"branches": [],
"timeout": 60
}))?;
db.insert_jobs(run_id, std::slice::from_ref(&spec))?;
db.cancel_run(run_id)?;
let detail = db.detail(run_id)?;
assert_eq!(detail.run.status, "canceled");
assert!(detail.run.finished_at.is_some());
assert!(detail.jobs.iter().all(|job| job.status == "canceled"));
Ok(())
}
#[test]
fn cancel_running_run_finalizes_jobs_and_deployments() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
assert!(db.claim_run(run_id)?);
let spec: crate::config::JobSpec = serde_json::from_value(serde_json::json!({
"base_name": "deploy",
"name": "deploy",
"needs": [],
"executor": "native",
"image": null,
"shell": null,
"command": "true",
"matrix": {},
"artifacts": [],
"cache": [],
"network": false,
"secrets": [],
"environment": "production",
"approval": false,
"branches": [],
"timeout": 60
}))?;
db.insert_jobs(run_id, std::slice::from_ref(&spec))?;
let job_id = db.jobs(run_id)?[0].id;
db.set_job_status(job_id, "running", None, None)?;
let deployment_id = db.begin_deployment(run_id, job_id, "production", "[]")?;
db.cancel_running_run(run_id)?;
let detail = db.detail(run_id)?;
assert_eq!(detail.run.status, "canceled");
assert_eq!(detail.jobs[0].status, "canceled");
assert!(detail.jobs[0].logs.contains("canceled by operator"));
assert_eq!(detail.deployments[0].id, deployment_id);
assert_eq!(detail.deployments[0].status, "failed");
assert_eq!(detail.deployments[0].result.as_deref(), Some("failure"));
Ok(())
}
#[test]
fn cancel_run_refuses_running_and_terminal_runs() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let running = db.create_run(repo.id, "main", None, "manual")?;
assert!(db.claim_run(running)?);
let err = db.cancel_run(running).expect_err("running must be refused");
assert!(format!("{err:#}").contains("running"), "got: {err:#}");
let finished = db.create_run(repo.id, "main", None, "manual")?;
db.finish_run(finished, "succeeded", None)?;
let err = db
.cancel_run(finished)
.expect_err("terminal must be refused");
assert!(format!("{err:#}").contains("succeeded"), "got: {err:#}");
let err = db.cancel_run(99999).expect_err("unknown must error");
assert!(format!("{err:#}").contains("unknown run"), "got: {err:#}");
Ok(())
}
#[test]
fn run_unknown_id_errors() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let err = db.run(99999).expect_err("unknown run should error");
let msg = format!("{err:#}");
assert!(msg.contains("unknown run"), "got: {msg}");
Ok(())
}
// ── retry creates a new run ──
#[test]
fn run_retry_creates_new_run_with_retry_trigger() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
db.finish_run(run_id, "failed", Some("oops"))?;
let retry_id = db.retry(run_id)?;
assert_ne!(retry_id, run_id, "retry creates a distinct run");
let retry_run = db.run(retry_id)?;
assert_eq!(retry_run.trigger, "retry");
assert_eq!(retry_run.status, "queued");
Ok(())
}
// ── jobs ──
fn dummy_job_spec(name: &str, base: &str) -> crate::config::JobSpec {
crate::config::JobSpec {
base_name: base.into(),
name: name.into(),
needs: vec![],
executor: "docker".into(),
image: Some("alpine".into()),
shell: None,
command: "echo ok".into(),
matrix: BTreeMap::new(),
artifacts: vec![],
cache: vec![],
network: false,
installation_id: None,
installation_revision: None,
rollback_of_deployment_id: None,
recovery_of_deployment_id: None,
secrets: vec![],
environment: None,
approval: false,
branches: vec![],
timeout: 60,
profile: None,
}
}
#[test]
fn job_insert_and_status_transitions() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
let specs = vec![dummy_job_spec("test", "test")];
db.insert_jobs(run_id, &specs)?;
let jobs = db.jobs(run_id)?;
assert_eq!(jobs.len(), 1);
let job_id = jobs[0].id;
assert_eq!(jobs[0].status, "queued");
assert_eq!(jobs[0].name, "test");
// transition to running
db.set_job_status(job_id, "running", None, None)?;
let jobs = db.jobs(run_id)?;
assert_eq!(jobs[0].status, "running");
assert!(jobs[0].started_at.is_some());
// transition to succeeded with logs and exit code
db.set_job_status(job_id, "succeeded", Some("all good"), Some(0))?;
let jobs = db.jobs(run_id)?;
assert_eq!(jobs[0].status, "succeeded");
assert_eq!(jobs[0].logs.as_str(), "all good");
assert_eq!(jobs[0].exit_code, Some(0));
assert!(jobs[0].finished_at.is_some());
Ok(())
}
// ── artifacts ──
#[test]
fn artifact_store_and_retrieve_by_id() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
let specs = vec![dummy_job_spec("build", "build")];
db.insert_jobs(run_id, &specs)?;
let job_id = db.jobs(run_id)?[0].id;
let artifact_id = db.add_artifact(
run_id,
job_id,
"binary",
"dist/binary",
"abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789",
1024,
)?;
let art = db.artifact(artifact_id)?;
assert_eq!(art.name, "binary");
assert_eq!(
art.sha256,
"abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789"
);
assert_eq!(art.bytes, 1024);
assert_eq!(art.path, "dist/binary");
Ok(())
}
#[test]
fn artifact_unknown_id_errors() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let err = db
.artifact(99999)
.expect_err("unknown artifact should error");
let msg = format!("{err:#}");
assert!(msg.contains("unknown artifact"), "got: {msg}");
Ok(())
}
#[test]
fn artifacts_by_run_empty_when_no_artifacts() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
let arts = db.artifacts(run_id)?;
assert!(arts.is_empty());
Ok(())
}
// ── deployments ──
#[test]
fn deployment_begin_and_finish() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
let specs = vec![dummy_job_spec("deploy", "deploy")];
db.insert_jobs(run_id, &specs)?;
let job_id = db.jobs(run_id)?[0].id;
let dep_id = db.begin_deployment(run_id, job_id, "production", "[]")?;
let deps = db.deployments(run_id)?;
assert_eq!(deps.len(), 1);
assert_eq!(deps[0].status, "running");
assert_eq!(deps[0].environment, "production");
db.finish_deployment(dep_id, "succeeded")?;
let deps = db.deployments(run_id)?;
assert_eq!(deps[0].status, "succeeded");
assert!(deps[0].finished_at.is_some());
Ok(())
}
// ── workers ──
#[test]
fn worker_register_heartbeat_and_list() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.register_worker("w1", "host1", "docker,native", 2)?;
let workers = db.workers()?;
assert_eq!(workers.len(), 1);
assert_eq!(workers[0].id, "w1");
assert_eq!(workers[0].capacity, 2);
assert_eq!(workers[0].status, "starting");
assert_eq!(workers[0].completed_runs, 0);
db.heartbeat_worker("w1", "running", Some(42), false)?;
let workers = db.workers()?;
assert_eq!(workers[0].status, "running");
assert_eq!(workers[0].current_run_id, Some(42));
db.heartbeat_worker("w1", "idle", None, true)?;
let workers = db.workers()?;
assert_eq!(workers[0].completed_runs, 1);
Ok(())
}
#[test]
fn worker_leases_are_exclusive_and_draining_blocks_claims() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.register_worker("w1", "h1", "docker", 1)?;
db.register_worker("w2", "h2", "docker", 1)?;
let first = db.acquire_worker_lease("w1", "token-one")?;
assert!(first.acquired);
assert!(!db.acquire_worker_lease("w1", "token-two")?.acquired);
assert!(db.acquire_worker_lease("w2", "token-two")?.acquired);
db.set_worker_draining("w1", true)?;
assert_eq!(
db.claim_next_queued_run_for_worker("w1", "token-one")?,
None
);
let workers = db.workers()?;
let w1 = workers
.iter()
.find(|worker| worker.id == "w1")
.context("w1")?;
assert!(w1.draining);
let w2 = workers
.iter()
.find(|worker| worker.id == "w2")
.context("w2")?;
assert!(w2.current_run_id.is_none());
Ok(())
}
#[test]
fn worker_heartbeat_unknown_id_errors() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let err = db
.heartbeat_worker("ghost", "idle", None, false)
.expect_err("heartbeat for unknown worker should fail");
let msg = format!("{err:#}");
assert!(msg.contains("unknown worker"), "got: {msg}");
Ok(())
}
#[test]
fn worker_reset_sets_all_offline() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.register_worker("w1", "h1", "docker", 1)?;
db.register_worker("w2", "h2", "native", 1)?;
db.heartbeat_worker("w1", "running", Some(1), false)?;
db.reset_workers()?;
for w in db.workers()? {
assert_eq!(w.status, "offline");
}
Ok(())
}
#[test]
fn worker_prune_removes_rows_outside_configured_set() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.register_worker("titan-1", "titan", "docker,native", 1)?;
db.register_worker("titan-2", "titan", "docker,native", 1)?;
let removed = db.prune_workers_except(&["titan-1".to_owned()])?;
assert_eq!(removed, 1);
let workers = db.workers()?;
assert_eq!(workers.len(), 1);
assert_eq!(workers[0].id, "titan-1");
// Idempotent: a second reconcile removes nothing.
assert_eq!(db.prune_workers_except(&["titan-1".to_owned()])?, 0);
// Refuses an empty keep set rather than wiping the table.
assert!(db.prune_workers_except(&[]).is_err());
assert_eq!(db.workers()?.len(), 1);
Ok(())
}
// ── recover interrupted ──
#[test]
fn recover_interrupted_requeues_running_work_with_diagnostic() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
// Set to running directly (bypassing claim).
db.finish_run(run_id, "running", None)?;
let specs = vec![dummy_job_spec("job", "job")];
db.insert_jobs(run_id, &specs)?;
let job_id = db.jobs(run_id)?[0].id;
db.set_job_status(job_id, "running", None, None)?;
let deployment_id = db.begin_deployment(run_id, job_id, "staging", "[]")?;
let recovered = db.recover_interrupted()?;
assert!(recovered >= 3, "should recover run + job + deployment");
let run = db.run(run_id)?;
assert_eq!(run.status, "queued");
assert_eq!(
run.error.as_deref(),
Some("controller restarted during execution; retrying incomplete job")
);
let jobs = db.jobs(run_id)?;
assert_eq!(jobs[0].status, "queued");
assert_eq!(jobs[0].exit_code, None);
assert!(
jobs[0]
.logs
.contains("controller restarted during execution; retrying job")
);
let deployments = db.deployments(run_id)?;
assert_eq!(deployments[0].id, deployment_id);
assert_eq!(deployments[0].status, "failed");
assert_eq!(deployments[0].result.as_deref(), Some("failure"));
Ok(())
}
#[test]
fn recovered_run_keeps_restart_diagnostic_when_claimed() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
db.finish_run(run_id, "running", None)?;
db.recover_interrupted()?;
assert_eq!(db.claim_next_queued_run()?, Some(run_id));
assert_eq!(db.run(run_id)?.status, "running");
assert_eq!(
db.run(run_id)?.error.as_deref(),
Some("controller restarted during execution; retrying incomplete job")
);
Ok(())
}
// ── claim_next_queued_run ──
#[test]
fn audit_events_retain_actor_and_targets() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.add_repository("app", "https://example.com/app.git", "main")?;
db.record_audit_event(
"user-123",
"akurai_run_promote",
Some("app"),
Some("production"),
Some("42"),
"succeeded",
&serde_json::json!({"wait": false}),
)?;
let events = db.audit_events(Some("app"), 10, 0)?;
assert_eq!(events.len(), 1);
assert_eq!(events[0].actor, "user-123");
assert_eq!(events[0].environment.as_deref(), Some("production"));
assert_eq!(events[0].target.as_deref(), Some("42"));
assert!(events[0].created_at > 0);
Ok(())
}
#[test]
fn delivery_metrics_are_source_derived_and_unknown_when_empty() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let empty = db.delivery_metrics(None, None, 86_400)?;
assert_eq!(empty.sample_size, 0);
assert!(empty.change_failure_rate.is_none());
assert!(empty.deployment_ids.is_empty());
assert!(empty.run_ids.is_empty());
assert!(!empty.truncated);
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", Some(&"a".repeat(40)), "manual")?;
db.insert_jobs(run_id, &[dummy_job_spec("deploy", "deploy")])?;
let job_id = db.jobs(run_id)?[0].id;
let deployment_id = db.begin_deployment(run_id, job_id, "production", "[]")?;
db.finish_deployment(deployment_id, "succeeded")?;
let metrics = db.delivery_metrics(Some("app"), Some("production"), 86_400)?;
assert_eq!(metrics.sample_size, 1);
assert_eq!(metrics.deployment_ids, vec![deployment_id]);
assert_eq!(metrics.run_ids, vec![run_id]);
assert_eq!(metrics.change_failure_rate, Some(0.0));
assert!(metrics.deployment_frequency_per_day.is_some());
Ok(())
}
#[test]
fn installation_revision_and_managed_deployment_are_idempotent_and_tenant_bound() -> Result<()>
{
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let revision =
db.register_installation_revision("inst-1", "org-1", "ws-1", "app-1", "request-1")?;
assert_eq!(revision, 1);
assert_eq!(
db.register_installation_revision("inst-1", "org-1", "ws-1", "app-1", "request-1")?,
1
);
assert!(
db.register_installation_revision("inst-1", "org-2", "ws-1", "app-1", "request-2")
.is_err()
);
assert!(
db.register_installation_revision("inst-1", "org-2", "ws-1", "app-1", "request-1")
.is_err()
);
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", Some(&"b".repeat(40)), "manual")?;
db.insert_jobs(run_id, &[dummy_job_spec("deploy", "deploy")])?;
let job_id = db.jobs(run_id)?[0].id;
let deployment_id = db.begin_managed_deployment(ManagedDeployment {
run_id,
job_id,
environment: "production",
artifacts_json: "[]",
installation_id: "inst-1",
installation_revision: 1,
commit_sha: Some(&"b".repeat(40)),
artifact_digest: Some(
"4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945",
),
result: Some("success"),
rollback_of_deployment_id: None,
recovery_of_deployment_id: None,
})?;
let deployment = &db.deployments(run_id)?[0];
assert_eq!(deployment.id, deployment_id);
assert_eq!(deployment.organization_id.as_deref(), Some("org-1"));
assert_eq!(deployment.installation_revision, Some(1));
assert_eq!(
deployment.artifact_digest.as_deref(),
Some("4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945")
);
assert_eq!(db.deployments_for_tenant(run_id, "org-1", "ws-1")?.len(), 1);
assert!(db.deployments_for_tenant(run_id, "org-2", "ws-2").is_err());
Ok(())
}
#[test]
fn claim_next_queued_run_returns_none_when_empty() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
assert_eq!(db.claim_next_queued_run()?, None);
Ok(())
}
#[test]
fn claim_next_queued_run_picks_oldest_queued() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let r1 = db.create_run(repo.id, "ref1", None, "manual")?;
let r2 = db.create_run(repo.id, "ref2", None, "manual")?;
// both queued; claim_next should pick oldest (r1)
let claimed = db.claim_next_queued_run()?;
assert_eq!(claimed, Some(r1));
// next claim gets r2
let next = db.claim_next_queued_run()?;
assert_eq!(next, Some(r2));
Ok(())
}
// ── queued_runs count ──
#[test]
fn queued_runs_counts_correctly() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
assert_eq!(db.queued_runs()?, 0);
db.create_run(repo.id, "ref1", None, "manual")?;
db.create_run(repo.id, "ref2", None, "manual")?;
assert_eq!(db.queued_runs()?, 2);
db.claim_next_queued_run()?;
assert_eq!(db.queued_runs()?, 1);
Ok(())
}
// ── detail aggregates ──
#[test]
fn detail_aggregates_run_jobs_artifacts_deployments() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
let specs = vec![dummy_job_spec("build", "build")];
db.insert_jobs(run_id, &specs)?;
let job_id = db.jobs(run_id)?[0].id;
db.add_artifact(
run_id,
job_id,
"bin",
"dist/bin",
"a".repeat(64).as_str(),
42,
)?;
db.begin_deployment(run_id, job_id, "staging", "[]")?;
let detail = db.detail(run_id)?;
assert_eq!(detail.run.id, run_id);
assert_eq!(detail.jobs.len(), 1);
assert_eq!(detail.artifacts.len(), 1);
assert_eq!(detail.deployments.len(), 1);
Ok(())
}
// ── set_run_commit ──
#[test]
fn set_run_commit_updates_sha() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
assert!(db.run(run_id)?.commit_sha.is_none());
db.set_run_commit(run_id, &"a".repeat(40))?;
assert_eq!(
db.run(run_id)?.commit_sha.as_deref(),
Some("a".repeat(40).as_str())
);
Ok(())
}
// ── approve_environment ──
#[test]
fn approve_environment_promotes_waiting_jobs() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
let repo = db.add_repository("app", "https://example.com/app.git", "main")?;
let run_id = db.create_run(repo.id, "main", None, "manual")?;
let mut spec = dummy_job_spec("deploy", "deploy");
spec.approval = true;
spec.environment = Some("production".into());
db.insert_jobs(run_id, &[spec])?;
let job_id = db.jobs(run_id)?[0].id;
// jobs inserted with approval_required are created as 'queued'
// set to waiting to test approval promotion
db.set_job_status(job_id, "waiting", None, None)?;
db.finish_run(run_id, "waiting", None)?;
let changed = db.approve_environment(run_id, "production")?;
assert_eq!(changed, 1);
let jobs = db.jobs(run_id)?;
assert_eq!(jobs[0].status, "queued");
assert!(jobs[0].approved_at.is_some());
Ok(())
}
// ── open empty db readable ──
#[test]
fn check_ready_returns_ok_on_fresh_db() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.check_ready()?;
Ok(())
}
// ── public_repositories ──
#[test]
fn public_repositories_include_real_build_activity() -> Result<()> {
let dir = tempfile::tempdir()?;
let db = open_db(&dir)?;
db.add_repository("priv", "https://example.com/priv.git", "main")?;
let repository = db.add_repository("pub", "https://github.com/olibuijr/pub.git", "main")?;
db.set_repository_visibility("pub", "public")?;
let succeeded = db.create_run(repository.id, "main", None, "manual")?;
db.finish_run(succeeded, "succeeded", None)?;
db.create_run(repository.id, "feature", None, "manual")?;
let public = db.public_repositories()?;
assert_eq!(public.len(), 1);
assert_eq!(public[0].name, "pub");
assert_eq!(public[0].run_count, 2);
assert_eq!(public[0].successful_runs, 1);
assert!(public[0].last_activity_at >= public[0].created_at);
Ok(())
}
// ── original tests preserved ──
#[test]
fn stores_ci_state_and_reopens_details() -> Result<()> {
let database = Database::memory(KEY)?;
let repo = database.add_repository("app", "https://example.com/app.git", "main")?;
assert_eq!(repo.visibility, "private");
assert_eq!(
database
.set_repository_visibility("app", "public")?
.visibility,
"public"
);
let run = database.create_run(repo.id, "main", None, "manual")?;
assert_eq!(database.detail(run)?.run.status, "queued");
Ok(())
}
#[test]
fn queries_runs_and_tracks_workers() -> Result<()> {
let database = Database::memory(KEY)?;
let alpha = database.add_repository("alpha", "https://example.com/alpha.git", "main")?;
let beta = database.add_repository("beta", "https://example.com/beta.git", "main")?;
let succeeded = database.create_run(alpha.id, "main", None, "manual")?;
database.finish_run(succeeded, "succeeded", None)?;
let failed = database.create_run(beta.id, "release", None, "webhook")?;
database.finish_run(failed, "failed", Some("compiler error"))?;
let runs = database.query_runs(&RunQuery {
repository: Some("alpha".into()),
statuses: vec!["succeeded".into()],
search: Some("main".into()),
limit: 20,
..RunQuery::default()
})?;
assert_eq!(runs.len(), 1);
assert_eq!(runs[0].id, succeeded);
let multi = database.query_runs(&RunQuery {
repository: Some("alpha,beta".into()),
statuses: vec!["succeeded".into(), "failed".into()],
limit: 20,
..RunQuery::default()
})?;
assert_eq!(multi.len(), 2);
let repositories = database.query_repositories(&RepositoryQuery {
search: Some("example.com/beta".into()),
visibility: Some("private".into()),
..RepositoryQuery::default()
})?;
assert_eq!(repositories.len(), 1);
assert_eq!(repositories[0].name, "beta");
let queued = database.create_run(alpha.id, "feature", None, "manual")?;
assert_eq!(database.claim_next_queued_run()?, Some(queued));
assert_eq!(database.run(queued)?.status, "running");
assert_eq!(database.claim_next_queued_run()?, None);
database.register_worker("titan-1", "titan", "docker,native", 1)?;
database.heartbeat_worker("titan-1", "running", Some(succeeded), false)?;
database.heartbeat_worker("titan-1", "idle", None, true)?;
let workers = database.workers()?;
assert_eq!(workers.len(), 1);
assert_eq!(workers[0].completed_runs, 1);
assert_eq!(workers[0].status, "idle");
database.reset_workers()?;
assert_eq!(database.workers()?[0].status, "offline");
Ok(())
}
#[test]
fn community_issues_are_numbered_bounded_and_cascading() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let repository =
database.add_repository("community", "https://example.com/community.git", "main")?;
let first = database.create_issue(
&repository.name,
"First issue",
"A **safe** description",
"subject-1",
"Ada",
)?;
let second =
database.create_issue(&repository.name, "Second issue", "", "subject-2", "Grace")?;
assert_eq!(first.number, 1);
assert_eq!(second.number, 2);
let comment = database.add_issue_comment(
&repository.name,
first.number,
"A comment",
"subject-2",
"Grace",
)?;
assert_eq!(comment.issue_number, first.number);
let detail = database.issue(&repository.name, first.number)?;
assert_eq!(detail.comments.len(), 1);
database.update_issue(&repository.name, first.number, None, None, Some("closed"))?;
assert_eq!(
database.issue(&repository.name, first.number)?.issue.state,
"closed"
);
database.remove_repository(&repository.name)?;
let remaining: i64 =
database
.connection()?
.query_row("SELECT COUNT(*) FROM issues", [], |row| row.get(0))?;
assert_eq!(remaining, 0);
Ok(())
}
// ── pull requests: persisted CRUD, SHA-bound review, stale invalidation ──
fn open_pr(database: &Database) -> Result<(crate::db::Repository, PullRequest)> {
let repository = database.add_repository("app", "https://example.com/app.git", "main")?;
let pr = database.create_pull_request(
&repository.name,
"Add feature",
"does a thing",
"feature-a",
"main",
"1111111111111111111111111111111111111aaa",
"0000000000000000000000000000000000000aaa",
"author-sub",
"Author Name",
)?;
database.add_pull_request_review(
&repository.name,
pr.number,
"reviewer-sub",
"Reviewer Name",
"approved",
"ship it",
)?;
Ok((repository, pr))
}
#[test]
fn pull_request_create_and_read_round_trips() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let (repository, pr) = open_pr(&database)?;
assert_eq!(pr.number, 1);
assert_eq!(pr.state, "open");
let detail = database.pull_request(&repository.name, pr.number)?;
assert_eq!(detail.pull_request.id, pr.id);
assert_eq!(detail.reviews.len(), 1);
assert!(detail.approved, "open_pr seeds one independent approval");
Ok(())
}
#[test]
fn pull_request_review_from_author_is_rejected() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let (repository, pr) = open_pr(&database)?;
let result = database.add_pull_request_review(
&repository.name,
pr.number,
"author-sub",
"Author Name",
"approved",
"looks good to me",
);
assert!(result.is_err(), "author must not be able to review own PR");
Ok(())
}
#[test]
fn independent_approval_at_current_head_counts() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let (repository, pr) = open_pr(&database)?;
database.add_pull_request_review(
&repository.name,
pr.number,
"reviewer-sub",
"Reviewer Name",
"approved",
"ship it",
)?;
let detail = database.pull_request(&repository.name, pr.number)?;
assert!(
detail.approved,
"independent approval at current head must count"
);
Ok(())
}
#[test]
fn push_after_approval_makes_it_stale() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let (repository, pr) = open_pr(&database)?;
database.add_pull_request_review(
&repository.name,
pr.number,
"reviewer-sub",
"Reviewer Name",
"approved",
"ship it",
)?;
assert!(database.pull_request(&repository.name, pr.number)?.approved);
// New commit pushed to the source ref: head_sha changes.
database.update_pull_request_head(
&repository.name,
pr.number,
"2222222222222222222222222222222222222bbb",
)?;
let detail = database.pull_request(&repository.name, pr.number)?;
assert!(
!detail.approved,
"approval bound to the old head_sha must not carry over a new push"
);
assert_eq!(
detail.reviews.len(),
2,
"review history is preserved, not deleted"
);
Ok(())
}
#[test]
fn enqueue_requires_open_pull_request() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let (repository, pr) = open_pr(&database)?;
database.enqueue_merge(&repository.name, pr.number, "reviewer-sub")?;
assert_eq!(
database
.pull_request(&repository.name, pr.number)?
.pull_request
.state,
"queued"
);
let second = database.enqueue_merge(&repository.name, pr.number, "reviewer-sub");
assert!(
second.is_err(),
"already-queued PR cannot be enqueued again"
);
Ok(())
}
#[test]
fn merge_queue_serializes_per_repository_fifo() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let repository = database.add_repository("app", "https://example.com/app.git", "main")?;
let mut prs = Vec::new();
for index in 0..3 {
let pr = database.create_pull_request(
&repository.name,
&format!("PR {index}"),
"body",
&format!("feature-{index}"),
"main",
&format!("{index:040x}").replacen('0', "1", 1),
"0000000000000000000000000000000000000aaa",
"author-sub",
"Author",
)?;
database.add_pull_request_review(
&repository.name,
pr.number,
"reviewer-sub",
"Reviewer",
"approved",
"ok",
)?;
database.enqueue_merge(&repository.name, pr.number, "reviewer-sub")?;
prs.push(pr);
}
let entries = database.list_merge_queue(&repository.name)?;
assert!(
!database.claim_merge_queue_entry(entries[1].id)?,
"direct claims must not bypass the FIFO head"
);
// Strict FIFO: next_queued_merge always returns the lowest position.
let first = database
.next_queued_merge(repository.id)?
.expect("first entry queued");
assert_eq!(first.pull_request_id, prs[0].id);
assert!(database.claim_merge_queue_entry(first.id)?);
// A second worker cannot claim the same entry (CAS).
assert!(!database.claim_merge_queue_entry(first.id)?);
// While the first entry is running, no other entry is offered —
// this is what makes the queue *serialized* rather than parallel.
assert!(
database.next_queued_merge(repository.id)?.is_none(),
"queue must not offer a second entry while one is running"
);
database.finish_merge_queue_entry(first.id, "succeeded", None, None)?;
let second = database
.next_queued_merge(repository.id)?
.expect("second entry queued after first finished");
assert_eq!(second.pull_request_id, prs[1].id);
Ok(())
}
#[test]
fn crash_recovery_resets_running_entries_to_queued() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let (repository, pr) = open_pr(&database)?;
let entry = database.enqueue_merge(&repository.name, pr.number, "reviewer-sub")?;
assert!(database.claim_merge_queue_entry(entry.id)?);
assert_eq!(database.merge_queue_entry(entry.id)?.status, "running");
// Simulate a worker crash: process restarts and runs recovery.
let recovered = database.recover_merge_queue()?;
assert_eq!(recovered, 1);
assert_eq!(database.merge_queue_entry(entry.id)?.status, "queued");
Ok(())
}
#[test]
fn failed_merge_returns_pull_request_to_open() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let (repository, pr) = open_pr(&database)?;
let entry = database.enqueue_merge(&repository.name, pr.number, "reviewer-sub")?;
database.claim_merge_queue_entry(entry.id)?;
database.fail_merge_queue_pull_request(entry.id, pr.id, "CI failed", None)?;
let detail = database.pull_request(&repository.name, pr.number)?;
assert_eq!(detail.pull_request.state, "open");
assert_eq!(database.merge_queue_entry(entry.id)?.status, "failed");
Ok(())
}
#[test]
fn pull_request_can_be_re_enqueued_after_a_failed_attempt() -> Result<()> {
// Regression for the review finding that the milestone-1 unique
// constraint was UNIQUE(repository_id, pull_request_id, status),
// which made a second 'failed' row for the same PR collide with
// the first one and blew up the whole retry path.
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let (repository, pr) = open_pr(&database)?;
let first = database.enqueue_merge(&repository.name, pr.number, "reviewer-sub")?;
database.claim_merge_queue_entry(first.id)?;
database.fail_merge_queue_pull_request(first.id, pr.id, "CI failed", None)?;
assert_eq!(
database
.pull_request(&repository.name, pr.number)?
.pull_request
.state,
"open"
);
let second = database.enqueue_merge(&repository.name, pr.number, "reviewer-sub")?;
database.claim_merge_queue_entry(second.id)?;
database.fail_merge_queue_pull_request(second.id, pr.id, "CI failed again", None)?;
assert_eq!(database.merge_queue_entry(first.id)?.status, "failed");
assert_eq!(database.merge_queue_entry(second.id)?.status, "failed");
assert_ne!(first.id, second.id);
Ok(())
}
#[test]
fn approval_then_changes_requested_by_same_reviewer_blocks_enqueue() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let (repository, pr) = open_pr(&database)?;
database.add_pull_request_review(
&repository.name,
pr.number,
"reviewer-sub",
"Reviewer",
"approved",
"lgtm",
)?;
assert!(database.pull_request(&repository.name, pr.number)?.approved);
database.add_pull_request_review(
&repository.name,
pr.number,
"reviewer-sub",
"Reviewer",
"changes_requested",
"actually no",
)?;
let detail = database.pull_request(&repository.name, pr.number)?;
assert!(
!detail.approved,
"a later changes_requested from the same reviewer must revoke their earlier approval"
);
assert!(
database
.enqueue_merge(&repository.name, pr.number, "reviewer-sub")
.is_err(),
"enqueue_merge must re-verify approval inside its own transaction"
);
Ok(())
}
#[test]
fn successful_merge_marks_pull_request_merged() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let (repository, pr) = open_pr(&database)?;
let entry = database.enqueue_merge(&repository.name, pr.number, "reviewer-sub")?;
database.claim_merge_queue_entry(entry.id)?;
database.finish_merged_pull_request(
entry.id,
pr.id,
"3333333333333333333333333333333333333ccc",
None,
)?;
let detail = database.pull_request(&repository.name, pr.number)?;
assert_eq!(detail.pull_request.state, "merged");
assert_eq!(
detail.pull_request.merge_commit.as_deref(),
Some("3333333333333333333333333333333333333ccc")
);
assert_eq!(database.merge_queue_entry(entry.id)?.status, "succeeded");
Ok(())
}
#[test]
fn set_branch_protection_and_check_protected_branch() -> Result<()> {
let dir = tempfile::tempdir()?;
let database = open_db(&dir)?;
let repository = database.add_repository("app", "https://example.com/app.git", "main")?;
assert!(!database.is_branch_protected(&repository.name, "main")?);
database.set_branch_protection(&repository.name, true)?;
assert!(database.is_branch_protected(&repository.name, "main")?);
assert!(
!database.is_branch_protected(&repository.name, "feature")?,
"only the default branch is protected"
);
database.set_branch_protection(&repository.name, false)?;
assert!(!database.is_branch_protected(&repository.name, "main")?);
Ok(())
}
}