291 lines
8.8 KiB
Rust
291 lines
8.8 KiB
Rust
|
use std::{collections::HashMap, fs::OpenOptions, path::PathBuf, process::Stdio, sync::Arc};
|
||
|
|
|
||
|
|
use axum::{extract::{Query, State, Json}, http::{HeaderMap, StatusCode}, response::{IntoResponse, Response}};
|
||
|
|
use chrono::NaiveDateTime;
|
||
|
|
use libc::SIGTERM;
|
||
|
|
use tokio::{process::Command, sync::RwLock};
|
||
|
|
|
||
|
|
pub struct App;
|
||
|
|
|
||
|
|
#[derive(Clone)]
|
||
|
|
pub struct AppState {
|
||
|
|
pub bearer_secret: String,
|
||
|
|
pub refs: HashMap<String, Arc<RefState>>,
|
||
|
|
}
|
||
|
|
|
||
|
|
pub struct RefState {
|
||
|
|
pub config: RefConfig,
|
||
|
|
pub status: RwLock<RefStatus>,
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
|
||
|
|
pub struct RefStatus {
|
||
|
|
pub status: Status,
|
||
|
|
pub triggered: Option<NaiveDateTime>,
|
||
|
|
pub succeeded: Option<NaiveDateTime>,
|
||
|
|
pub last_payload: Option<WebhookPayload>,
|
||
|
|
pub child_pid: u32,
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
|
||
|
|
#[serde(rename_all = "snake_case")]
|
||
|
|
pub enum Status {
|
||
|
|
Never,
|
||
|
|
Deploying,
|
||
|
|
Validating,
|
||
|
|
Success,
|
||
|
|
Fatal(String),
|
||
|
|
Error(String),
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, serde::Deserialize, serde::Serialize)]
|
||
|
|
pub struct RefConfig {
|
||
|
|
#[serde(rename = "ref")]
|
||
|
|
pub ref_: String,
|
||
|
|
pub stdout: Option<PathBuf>,
|
||
|
|
pub stderr: Option<PathBuf>,
|
||
|
|
pub working_dir: PathBuf,
|
||
|
|
pub compose_flags: Option<Vec<String>>,
|
||
|
|
pub env: Option<HashMap<String, String>>,
|
||
|
|
pub uid: u32,
|
||
|
|
pub gid: u32,
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
|
||
|
|
pub struct ApiError {
|
||
|
|
pub code: u16,
|
||
|
|
pub message: String,
|
||
|
|
}
|
||
|
|
|
||
|
|
impl IntoResponse for ApiError {
|
||
|
|
fn into_response(self) -> Response {
|
||
|
|
(StatusCode::from_u16(self.code).unwrap(), Json(&self)).into_response()
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
|
||
|
|
pub struct WebhookPayload {
|
||
|
|
#[serde(rename = "ref")]
|
||
|
|
pub ref_: String,
|
||
|
|
|
||
|
|
pub before: String,
|
||
|
|
pub after: String,
|
||
|
|
}
|
||
|
|
|
||
|
|
#[derive(Debug, serde::Deserialize)]
|
||
|
|
pub struct GetStatusQuery {
|
||
|
|
#[serde(rename = "ref")]
|
||
|
|
pub ref_: Option<String>,
|
||
|
|
|
||
|
|
pub branch: Option<String>,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
fn require_bearer(
|
||
|
|
bcrypted: &str,
|
||
|
|
headers: &HeaderMap) -> Result<(), ApiError> {
|
||
|
|
let bearer = headers
|
||
|
|
.get("authorization")
|
||
|
|
.ok_or_else(|| ApiError {
|
||
|
|
code: 401,
|
||
|
|
message: "Unauthorized".to_string(),
|
||
|
|
})?
|
||
|
|
.to_str()
|
||
|
|
.map_err(|_| ApiError {
|
||
|
|
code: 400,
|
||
|
|
message: "Invalid Authorization header".to_string(),
|
||
|
|
})?;
|
||
|
|
let bearer = bearer.trim_start_matches("Bearer ").trim_start_matches("bearer ");
|
||
|
|
|
||
|
|
if !bcrypt::verify(bearer, bcrypted).map_err(|_| ApiError {
|
||
|
|
code: 401,
|
||
|
|
message: "Unauthorized".to_string(),
|
||
|
|
})? {
|
||
|
|
return Err(ApiError {
|
||
|
|
code: 401,
|
||
|
|
message: "Unauthorized".to_string(),
|
||
|
|
});
|
||
|
|
}
|
||
|
|
|
||
|
|
Ok(())
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
impl App {
|
||
|
|
pub async fn post_deploy(
|
||
|
|
State(state): State<AppState>,
|
||
|
|
headers: HeaderMap,
|
||
|
|
Json(payload): Json<WebhookPayload>,
|
||
|
|
) -> Result<StatusCode, ApiError> {
|
||
|
|
require_bearer(&state.bearer_secret, &headers)?;
|
||
|
|
|
||
|
|
let matched_ref = state.refs.get(&payload.ref_).ok_or_else(|| ApiError {
|
||
|
|
code: 404,
|
||
|
|
message: "Ref not found".to_string(),
|
||
|
|
})?.clone();
|
||
|
|
|
||
|
|
let mut status = matched_ref.status.write().await;
|
||
|
|
|
||
|
|
status.last_payload = Some(payload.clone());
|
||
|
|
|
||
|
|
if status.status == Status::Deploying {
|
||
|
|
if status.child_pid == 0 {
|
||
|
|
return Err(ApiError {
|
||
|
|
code: 500,
|
||
|
|
message: "Child PID is 0 while status is InProgress".to_string(),
|
||
|
|
});
|
||
|
|
}
|
||
|
|
if unsafe { libc::kill(status.child_pid as i32, SIGTERM) } != 0 {
|
||
|
|
return Err(ApiError {
|
||
|
|
code: 500,
|
||
|
|
message: "Failed to send SIGTERM to existing process".to_string(),
|
||
|
|
});
|
||
|
|
}
|
||
|
|
};
|
||
|
|
|
||
|
|
if !Command::new("git")
|
||
|
|
.arg("fetch")
|
||
|
|
.arg("origin")
|
||
|
|
.uid(matched_ref.config.uid)
|
||
|
|
.gid(matched_ref.config.gid)
|
||
|
|
.current_dir(matched_ref.config.working_dir.clone())
|
||
|
|
.spawn()
|
||
|
|
.map_err(|_| ApiError {
|
||
|
|
code: 500,
|
||
|
|
message: "Failed to spawn command".to_string(),
|
||
|
|
})?
|
||
|
|
.wait()
|
||
|
|
.await
|
||
|
|
.map_err(|_| ApiError {
|
||
|
|
code: 500,
|
||
|
|
message: "Failed to wait for command".to_string(),
|
||
|
|
})?.success() {
|
||
|
|
status.status = Status::Error("Failed to fetch".to_string());
|
||
|
|
return Ok(StatusCode::OK);
|
||
|
|
}
|
||
|
|
|
||
|
|
if !Command::new("git")
|
||
|
|
.arg("reset")
|
||
|
|
.arg("--hard")
|
||
|
|
.arg(payload.after)
|
||
|
|
.uid(matched_ref.config.uid)
|
||
|
|
.gid(matched_ref.config.gid)
|
||
|
|
.current_dir(matched_ref.config.working_dir.clone())
|
||
|
|
.spawn()
|
||
|
|
.map_err(|_| ApiError {
|
||
|
|
code: 500,
|
||
|
|
message: "Failed to spawn command".to_string(),
|
||
|
|
})?
|
||
|
|
.wait()
|
||
|
|
.await
|
||
|
|
.map_err(|_| ApiError {
|
||
|
|
code: 500,
|
||
|
|
message: "Failed to wait for command".to_string(),
|
||
|
|
})?.success() {
|
||
|
|
status.status = Status::Error("Failed to reset".to_string());
|
||
|
|
return Ok(StatusCode::OK);
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
let mut command = Command::new("docker")
|
||
|
|
.arg("compose")
|
||
|
|
.arg("up")
|
||
|
|
.arg("--detach")
|
||
|
|
.arg("--build")
|
||
|
|
.args(matched_ref.config.compose_flags.as_ref().unwrap_or(&Vec::new()))
|
||
|
|
.current_dir(matched_ref.config.working_dir.clone())
|
||
|
|
.kill_on_drop(true)
|
||
|
|
.envs(matched_ref.config.env.as_ref().unwrap_or(&HashMap::new()).iter())
|
||
|
|
.stdout(
|
||
|
|
match matched_ref.config.stdout.as_ref() {
|
||
|
|
Some(path) => {
|
||
|
|
let file = OpenOptions::new()
|
||
|
|
.create(true)
|
||
|
|
.append(true)
|
||
|
|
.open(path)
|
||
|
|
.map_err(|_| ApiError {
|
||
|
|
code: 500,
|
||
|
|
message: "Failed to open stdout file".to_string(),
|
||
|
|
})?;
|
||
|
|
|
||
|
|
Stdio::from(file)
|
||
|
|
}
|
||
|
|
None => Stdio::null(),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
.spawn()
|
||
|
|
.map_err(|_| ApiError {
|
||
|
|
code: 500,
|
||
|
|
message: "Failed to spawn command".to_string(),
|
||
|
|
})?;
|
||
|
|
|
||
|
|
status.status = Status::Deploying;
|
||
|
|
status.triggered = Some(chrono::Utc::now().naive_utc());
|
||
|
|
|
||
|
|
status.child_pid = command.id().ok_or_else( ||
|
||
|
|
ApiError {
|
||
|
|
code: 500,
|
||
|
|
message: "Failed to get child PID".to_string(),
|
||
|
|
}
|
||
|
|
)?;
|
||
|
|
|
||
|
|
let matched_ref = matched_ref.clone();
|
||
|
|
tokio::spawn(async move {
|
||
|
|
match command.wait().await {
|
||
|
|
Ok(exit) => {
|
||
|
|
let mut status = matched_ref.status.write().await;
|
||
|
|
status.status = if exit.success() {
|
||
|
|
status.succeeded = Some(chrono::Utc::now().naive_utc());
|
||
|
|
Status::Success
|
||
|
|
} else {
|
||
|
|
Status::Error(format!("Exit code: {}", exit.code().unwrap_or(-1)))
|
||
|
|
};
|
||
|
|
status.child_pid = 0;
|
||
|
|
}
|
||
|
|
Err(e) => {
|
||
|
|
let mut status = matched_ref.status.write().await;
|
||
|
|
status.status = Status::Fatal(format!("Subprocess error: {}", e));
|
||
|
|
status.child_pid = 0;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
});
|
||
|
|
|
||
|
|
|
||
|
|
Ok(StatusCode::OK)
|
||
|
|
}
|
||
|
|
|
||
|
|
pub async fn summary(
|
||
|
|
State(state): State<AppState>,
|
||
|
|
) -> Result<Json<HashMap<String, RefStatus>>, ApiError> {
|
||
|
|
let mut statuses = HashMap::new();
|
||
|
|
for (ref_, ref_state) in state.refs.iter() {
|
||
|
|
let status = ref_state.status.read().await;
|
||
|
|
statuses.insert(ref_.clone(), status.clone());
|
||
|
|
}
|
||
|
|
|
||
|
|
Ok(Json(statuses))
|
||
|
|
}
|
||
|
|
|
||
|
|
pub async fn get_status(
|
||
|
|
State(state): State<AppState>,
|
||
|
|
Query(query): Query<GetStatusQuery>,
|
||
|
|
) -> Result<Json<RefStatus>, ApiError> {
|
||
|
|
let ref_ = match (&query.ref_, &query.branch) {
|
||
|
|
(Some(ref_), None) => ref_.clone(),
|
||
|
|
(None, Some(branch)) => format!("refs/heads/{}", branch),
|
||
|
|
_ => return Err(ApiError {
|
||
|
|
code: 400,
|
||
|
|
message: "Either ref or branch must be specified".to_string(),
|
||
|
|
}),
|
||
|
|
};
|
||
|
|
|
||
|
|
let matched_ref = state.refs.get(&ref_).ok_or_else(|| ApiError {
|
||
|
|
code: 404,
|
||
|
|
message: "Ref not found".to_string(),
|
||
|
|
})?.clone();
|
||
|
|
|
||
|
|
let status = matched_ref.status.read().await;
|
||
|
|
|
||
|
|
Ok(Json(status.clone()))
|
||
|
|
}
|
||
|
|
}
|