Menu
AkurAI-Build
publicLatest change 30e6b8c347048d64c2c048dfd05a8ded219d37f6 - Host authenticated Git repositories over Smart HTTP with repo host/sync CLI by Ólafur Búi Ólafsson
use std::{
env,
fs::{self, OpenOptions},
io::{Read, Write},
path::{Path, PathBuf},
sync::{Arc, Mutex, MutexGuard},
time::Duration,
};
use anyhow::{Context, Result, anyhow, ensure};
use rusqlite::{
Connection, OpenFlags, TransactionBehavior, config::DbConfig, limits::Limit, params,
params_from_iter, types::Value,
};
use serde::{Deserialize, Serialize};
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"),
),
];
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>>);
#[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 created_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 limit: usize,
pub offset: usize,
}
impl Default for RunQuery {
fn default() -> Self {
Self {
repository: None,
statuses: Vec::new(),
git_ref: None,
triggers: Vec::new(),
search: None,
limit: 20,
offset: 0,
}
}
}
#[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,
}
#[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 started_at: Option<i64>,
pub finished_at: Option<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>,
}
#[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 == 3,
"database schema is not AkurAI Build v3"
);
for table in [
"repositories",
"runs",
"jobs",
"artifacts",
"deployments",
"workers",
"_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(())
}
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) VALUES (?1, ?2, ?3)
ON CONFLICT(name) DO UPDATE SET url=excluded.url, default_branch=excluded.default_branch",
params![name, url, branch],
)?;
drop(connection);
self.repository(name)
}
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 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)
}
pub fn repository(&self, name: &str) -> Result<Repository> {
self.connection()?
.query_row(
"SELECT id, name, url, default_branch, visibility, created_at FROM repositories WHERE name=?1",
[name],
map_repository,
)
.with_context(|| format!("unknown 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, 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, 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<Repository>> {
let connection = self.connection()?;
let mut statement = connection.prepare(
"SELECT id, name, url, default_branch, visibility, created_at
FROM repositories WHERE visibility='public' ORDER BY name",
)?;
Ok(statement
.query_map([], map_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())
}
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>> {
ensure!(
(1..=200).contains(&query.limit),
"run limit must be 1..=200"
);
ensure!(query.offset <= 10_000, "run offset must be <= 10000");
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 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",
);
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 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));
}
if clauses.is_empty() {
sql.push_str(" WHERE 1=1");
} else {
sql.push_str(" WHERE ");
sql.push_str(&clauses.join(" AND "));
}
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 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=NULL
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, 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> {
let connection = self.connection()?;
let changed = connection.execute(
"UPDATE jobs SET approved_at=unixepoch(), 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],
)?;
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)
}
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 finish_deployment(&self, id: i64, status: &str) -> Result<()> {
self.connection()?.execute(
"UPDATE deployments SET status=?2, finished_at=unixepoch() WHERE id=?1",
params![id, status],
)?;
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
FROM deployments WHERE run_id=?1 ORDER BY id",
)?;
Ok(statement
.query_map([run_id], map_deployment)?
.collect::<rusqlite::Result<_>>()?)
}
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, heartbeat_at=unixepoch()",
[],
)?;
Ok(())
}
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,
status='starting', capacity=excluded.capacity, current_run_id=NULL,
started_at=excluded.started_at, heartbeat_at=excluded.heartbeat_at, completed_runs=0",
params![id, host, capabilities, i64::try_from(capacity)?],
)?;
Ok(())
}
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",
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
FROM workers ORDER BY id",
)?;
Ok(statement
.query_map([], map_worker)?
.collect::<rusqlite::Result<_>>()?)
}
pub fn recover_interrupted(&self) -> Result<usize> {
let connection = self.connection()?;
let deployments = connection.execute(
"UPDATE deployments SET status='failed', finished_at=unixepoch() WHERE status='running'",
[],
)?;
let jobs = connection.execute(
"UPDATE jobs SET status='interrupted', finished_at=unixepoch() WHERE status='running'",
[],
)?;
let runs = connection.execute(
"UPDATE runs SET status='interrupted', error='controller restarted during execution', finished_at=unixepoch() WHERE status='running'",
[],
)?;
Ok(deployments + jobs + runs)
}
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)?,
created_at: row.get(5)?,
})
}
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)?,
started_at: row.get(15)?,
finished_at: row.get(16)?,
})
}
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)?,
})
}
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)?,
})
}
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(())
}
trait OptionalRow<T> {
fn optional(self) -> rusqlite::Result<Option<T>>;
}
impl<T> OptionalRow<T> for rusqlite::Result<T> {
fn optional(self) -> rusqlite::Result<Option<T>> {
match self {
Ok(value) => Ok(Some(value)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(error) => Err(error),
}
}
}
pub fn default_paths(root: &Path) -> (PathBuf, PathBuf, PathBuf) {
(
root.join("akurai.db"),
root.join("workspaces"),
root.join("artifacts"),
)
}
#[cfg(test)]
mod tests {
use super::*;
const KEY: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
#[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(())
}
}