Menu
AkurAI-Build
publicLatest change ab672e9cd796b2ae3b9efc364a14d20dec03a31d - Add bounded agent wait and readiness recovery by Ólafur Búi Ólafsson
use std::{
env, fs,
net::SocketAddr,
path::{Path, PathBuf},
process::Command,
thread,
time::{Duration, Instant},
};
use akurai_build::{
DEFAULT_ADDRESS, config,
db::{Database, generate_secret, load_secret},
runner::Runner,
server::{self, ServerOptions},
};
use anyhow::{Context, Result, ensure};
use clap::{Args, Parser, Subcommand};
use serde::Serialize;
use serde_json::json;
use tracing_subscriber::EnvFilter;
#[derive(Parser)]
#[command(
name = "akurai",
version,
about = "Lean Git-native CI/CD on the Bunfork foundation"
)]
struct Cli {
#[arg(long, env = "AKURAI_DATA")]
data: Option<PathBuf>,
#[arg(long, env = "AKURAI_KEY_FILE")]
key_file: Option<PathBuf>,
#[command(subcommand)]
command: Action,
}
#[derive(Subcommand)]
enum Action {
/// Generate a private 256-bit key or token.
Keygen {
#[arg(long)]
out: PathBuf,
},
/// Create or update the encrypted database schema.
Migrate,
/// Write a minimal .akurai.yml into a repository.
Init {
#[arg(long, default_value = ".akurai.yml")]
out: PathBuf,
#[arg(long)]
force: bool,
},
/// Register and inspect Git repositories.
Repo {
#[command(subcommand)]
command: RepoAction,
},
/// Queue a pipeline, optionally waiting in this process.
Run(RunArgs),
/// List recent pipeline runs.
Runs {
#[arg(long)]
repo: Option<String>,
#[arg(long, default_value_t = 20)]
limit: usize,
},
/// Show a run, jobs, artifacts, and deployments.
Show { id: i64 },
/// Wait for a persisted terminal or approval-required state.
Wait {
id: i64,
#[arg(long, default_value_t = 3600)]
timeout: u64,
#[arg(long, default_value_t = 2)]
interval: u64,
},
/// Return job logs for a run.
Logs {
id: i64,
#[arg(long)]
failed: bool,
},
/// Queue the same immutable revision as a prior run.
Retry {
id: i64,
#[arg(long)]
wait: bool,
},
/// Approve and resume a protected environment.
Promote {
id: i64,
environment: String,
#[arg(long)]
wait: bool,
},
/// Retrieve build artifacts.
Artifact {
#[command(subcommand)]
command: ArtifactAction,
},
/// Run controller diagnostics without changing state.
Doctor,
/// Serve the MiniJinja dashboard, API, webhooks, and worker.
Serve(ServeArgs),
}
#[derive(Subcommand)]
enum RepoAction {
Add {
name: String,
url: String,
#[arg(long, default_value = "main")]
branch: String,
},
List,
}
#[derive(Args)]
struct RunArgs {
repository: String,
#[arg(long)]
git_ref: Option<String>,
#[arg(long)]
commit: Option<String>,
#[arg(long)]
wait: bool,
}
#[derive(Subcommand)]
enum ArtifactAction {
Get {
id: i64,
output: PathBuf,
#[arg(long)]
force: bool,
},
}
#[derive(Args)]
struct ServeArgs {
#[arg(long, env = "AKURAI_ADDRESS", default_value = DEFAULT_ADDRESS)]
address: SocketAddr,
#[arg(long, env = "AKURAI_TOKEN_FILE")]
token_file: Option<PathBuf>,
#[arg(long, env = "AKURAI_WEBHOOK_SECRET_FILE")]
webhook_secret_file: Option<PathBuf>,
#[arg(
long,
env = "AKURAI_PUBLIC_ORIGIN",
default_value = "https://akurai-builds.olibuijr.com"
)]
public_origin: String,
#[arg(long, env = "AKURAI_ALLOW_NATIVE", default_value_t = false)]
allow_native: bool,
}
#[derive(Serialize)]
struct Envelope<T: Serialize> {
ok: bool,
data: T,
}
#[tokio::main]
async fn main() {
init_logging();
let cli = Cli::parse();
if let Err(error) = execute(cli).await {
let body = json!({"ok":false,"error":{"code":"failed","message":format!("{error:#}")}});
println!(
"{}",
serde_json::to_string(&body).unwrap_or_else(|_| "{\"ok\":false}".into())
);
std::process::exit(1);
}
}
async fn execute(cli: Cli) -> Result<()> {
let paths = Paths::new(cli.data, cli.key_file)?;
match cli.command {
Action::Keygen { out } => {
generate_secret(&out)?;
print(json!({"path":out,"created":true}))?;
}
Action::Init { out, force } => {
ensure!(
force || !out.exists(),
"{} already exists; use --force",
out.display()
);
fs::write(&out, include_str!("../akurai.example.yml"))?;
print(json!({"path":out,"created":true}))?;
}
Action::Migrate => {
let key = load_secret(&paths.key_file, "AKURAI_DB_KEY")?;
let database = Database::open(&paths.database, &key, true)?;
let applied = database.migrate()?;
print(json!({"database":paths.database,"migrations_applied":applied}))?;
}
Action::Repo { command } => {
let database = open(&paths)?;
match command {
RepoAction::Add { name, url, branch } => {
validate_name(&name)?;
validate_repository_url(&url)?;
config::validate_ref(&branch)?;
print(database.add_repository(&name, &url, &branch)?)?;
}
RepoAction::List => print(database.repositories()?)?,
}
}
Action::Run(args) => {
let database = open(&paths)?;
let runner = Runner::new(database.clone(), paths.root.clone(), native_allowed())?;
let id = runner.queue(
&args.repository,
args.git_ref.as_deref(),
args.commit.as_deref(),
"manual",
)?;
if args.wait {
print(runner.process(id)?)?;
} else {
print(json!({"id":id,"status":"queued"}))?;
}
}
Action::Runs { repo, limit } => print(open(&paths)?.runs(repo.as_deref(), limit)?)?,
Action::Show { id } => print(open(&paths)?.detail(id)?)?,
Action::Wait {
id,
timeout,
interval,
} => {
ensure!(
(1..=14_400).contains(&timeout),
"timeout must be 1..=14400 seconds"
);
ensure!(
(1..=60).contains(&interval),
"interval must be 1..=60 seconds"
);
let database = open(&paths)?;
let deadline = Instant::now() + Duration::from_secs(timeout);
loop {
let detail = database.detail(id)?;
if matches!(
detail.run.status.as_str(),
"waiting" | "succeeded" | "failed" | "canceled" | "interrupted"
) {
print(detail)?;
break;
}
ensure!(Instant::now() < deadline, "timed out waiting for run {id}");
thread::sleep(Duration::from_secs(interval));
}
}
Action::Logs { id, failed } => {
let mut jobs = open(&paths)?.jobs(id)?;
if failed {
jobs.retain(|job| {
matches!(job.status.as_str(), "failed" | "interrupted" | "canceled")
});
}
print(jobs.into_iter().map(|job| json!({"id":job.id,"name":job.name,"status":job.status,"logs":job.logs})).collect::<Vec<_>>())?;
}
Action::Retry { id, wait } => {
let database = open(&paths)?;
let id = database.retry(id)?;
if wait {
let runner = Runner::new(database, paths.root.clone(), native_allowed())?;
print(runner.process(id)?)?;
} else {
print(json!({"id":id,"status":"queued"}))?;
}
}
Action::Promote {
id,
environment,
wait,
} => {
let database = open(&paths)?;
let jobs = database.approve_environment(id, &environment)?;
if wait {
let runner = Runner::new(database, paths.root.clone(), native_allowed())?;
print(runner.process(id)?)?;
} else {
print(
json!({"id":id,"environment":environment,"approved_jobs":jobs,"status":"queued"}),
)?;
}
}
Action::Artifact { command } => match command {
ArtifactAction::Get { id, output, force } => {
ensure!(
force || !output.exists(),
"{} exists; use --force",
output.display()
);
let database = open(&paths)?;
let runner = Runner::new(database.clone(), paths.root.clone(), native_allowed())?;
let artifact = database.artifact(id)?;
let source = runner.artifact_path(&artifact)?;
if let Some(parent) = output.parent().filter(|path| !path.as_os_str().is_empty()) {
fs::create_dir_all(parent)?;
}
fs::copy(source, &output)?;
print(
json!({"id":id,"output":output,"sha256":artifact.sha256,"bytes":artifact.bytes}),
)?;
}
},
Action::Doctor => print(doctor(&paths)?)?,
Action::Serve(args) => {
let key = load_secret(&paths.key_file, "AKURAI_DB_KEY")?;
let database = Database::open(&paths.database, &key, false)?;
let token_file = args
.token_file
.unwrap_or_else(|| paths.config.join("api.token"));
let webhook_file = args
.webhook_secret_file
.unwrap_or_else(|| paths.config.join("webhook.token"));
let token = load_secret(&token_file, "AKURAI_API_TOKEN")?;
let webhook_secret = load_secret(&webhook_file, "AKURAI_WEBHOOK_SECRET")?;
server::serve(
database,
ServerOptions {
address: args.address,
token: token.to_string(),
webhook_secret: webhook_secret.to_string(),
public_origin: args.public_origin,
data_root: paths.root,
allow_native: args.allow_native,
},
)
.await?;
}
}
Ok(())
}
struct Paths {
root: PathBuf,
database: PathBuf,
config: PathBuf,
key_file: PathBuf,
}
impl Paths {
fn new(data: Option<PathBuf>, key_file: Option<PathBuf>) -> Result<Self> {
let home = env::var_os("HOME")
.map(PathBuf::from)
.context("HOME is not set")?;
let root = data.unwrap_or_else(|| home.join(".local/share/akurai-build"));
let config = home.join(".config/akurai-build");
let database = root.join("akurai.db");
let key_file = key_file.unwrap_or_else(|| config.join("database.key"));
Ok(Self {
root,
database,
config,
key_file,
})
}
}
fn open(paths: &Paths) -> Result<Database> {
let key = load_secret(&paths.key_file, "AKURAI_DB_KEY")?;
let database = Database::open(&paths.database, &key, false)?;
database.validate_schema()?;
Ok(database)
}
fn print<T: Serialize>(data: T) -> Result<()> {
println!(
"{}",
serde_json::to_string_pretty(&Envelope { ok: true, data })?
);
Ok(())
}
fn doctor(paths: &Paths) -> Result<serde_json::Value> {
let git = version("git", &["--version"]);
let docker = version("docker", &["info", "--format", "{{.ServerVersion}}"]);
let database = open(paths)
.and_then(|database| database.check_ready())
.map(|()| "ready".to_owned())
.unwrap_or_else(|error| format!("unavailable: {error}"));
Ok(json!({"database":database,"git":git,"docker":docker,"data":paths.root}))
}
fn version(program: &str, arguments: &[&str]) -> serde_json::Value {
match Command::new(program).args(arguments).output() {
Ok(output) => {
json!({"ok":output.status.success(),"output":String::from_utf8_lossy(if output.status.success(){&output.stdout}else{&output.stderr}).trim()})
}
Err(error) => json!({"ok":false,"output":error.to_string()}),
}
}
fn validate_name(value: &str) -> Result<()> {
ensure!(
!value.is_empty() && value.len() <= 64,
"invalid repository name"
);
ensure!(
value
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || b"._-".contains(&byte)),
"invalid repository name"
);
Ok(())
}
fn validate_repository_url(value: &str) -> Result<()> {
ensure!(
!value.starts_with('-') && value.len() <= 2048,
"invalid repository URL"
);
let local = Path::new(value).is_absolute();
ensure!(
local
|| value.starts_with("https://")
|| value.starts_with("ssh://")
|| value.starts_with("git@"),
"repository must use HTTPS, SSH, or an absolute local path"
);
Ok(())
}
fn native_allowed() -> bool {
env::var("AKURAI_ALLOW_NATIVE")
.is_ok_and(|value| matches!(value.as_str(), "1" | "true" | "yes"))
}
fn init_logging() {
let filter = EnvFilter::try_from_default_env()
.unwrap_or_else(|_| EnvFilter::new("akurai_build=info,tower_http=info"));
tracing_subscriber::fmt()
.with_env_filter(filter)
.with_writer(std::io::stderr)
.init();
}