eba4fd1
use models::ci::ci_environment;
use models::ci::ci_secret::{self, CiSecretScope};
use models::ci::runner::{self, RunnerState};
use models::ci::runner_token;
use models::ci::workflow_job;
use models::ci::workflow_run::{self, WorkflowRunState};
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
QuerySelect, Set,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;
use crate::errors::AdminError;
use crate::services::service::AdminService;
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct RunnerDto {
pub id: Uuid,
pub name: String,
pub labels: Vec<String>,
pub capacity: i32,
pub version: Option<String>,
pub protocol_version: Option<i32>,
pub capabilities: Vec<String>,
pub state: String,
pub space_id: Option<Uuid>,
pub active_jobs: i64,
pub last_seen_at: Option<OffsetDateTime>,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct RunnerPageDto {
pub items: Vec<RunnerDto>,
pub total: u64,
pub limit: u64,
pub offset: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct UpdateRunnerCmd {
pub labels: Option<Vec<String>>,
pub capacity: Option<i32>,
pub enabled: Option<bool>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct RunnerTokenDto {
pub id: Uuid,
pub name: String,
pub space_id: Option<Uuid>,
pub single_use: bool,
pub used_at: Option<OffsetDateTime>,
pub revoked_at: Option<OffsetDateTime>,
pub expires_at: Option<OffsetDateTime>,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct CreatedRunnerTokenDto {
pub token: RunnerTokenDto,
pub plaintext: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct CreateRunnerTokenCmd {
pub name: String,
pub space_id: Option<Uuid>,
#[serde(default)]
pub expires_in_days: Option<i64>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct EnvironmentAdminDto {
pub id: Uuid,
pub space_id: Uuid,
pub name: String,
pub description: Option<String>,
pub required_labels: Vec<String>,
pub secrets: Vec<String>,
pub created_at: OffsetDateTime,
pub updated_at: OffsetDateTime,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct UpsertEnvironmentAdminCmd {
pub name: String,
#[serde(default)]
pub description: Option<String>,
#[serde(default)]
pub required_labels: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct UpsertEnvironmentSecretCmd {
pub name: String,
pub value: String,
}
pub(crate) fn map_ci_error(error: ci::CiError) -> AdminError {
match error {
ci::CiError::NotFound => AdminError::NotFound,
ci::CiError::InvalidArgument(message) => AdminError::BadRequest(message),
ci::CiError::Conflict(message) => AdminError::Conflict(message),
ci::CiError::Unauthorized | ci::CiError::Forbidden => AdminError::Forbidden,
other => AdminError::Internal(other.to_string()),
}
}
impl RunnerDto {
fn from_model(model: runner::Model, active_jobs: i64) -> Self {
Self {
id: model.id,
name: model.name,
labels: serde_json::from_value(model.labels).unwrap_or_default(),
capacity: model.capacity,
version: model.version,
protocol_version: model.protocol_version,
capabilities: model
.capabilities
.and_then(|value| serde_json::from_value(value).ok())
.unwrap_or_default(),
state: match model.state {
RunnerState::Active => "active",
RunnerState::Disabled => "disabled",
}
.to_string(),
space_id: model.space_id,
active_jobs,
last_seen_at: model.last_seen_at,
created_at: model.created_at,
}
}
}
fn token_dto(model: runner_token::Model) -> RunnerTokenDto {
RunnerTokenDto {
id: model.id,
name: model.name,
space_id: model.space_id,
single_use: model.single_use,
used_at: model.used_at,
revoked_at: model.revoked_at,
expires_at: model.expires_at,
created_at: model.created_at,
}
}
impl AdminService {
pub async fn list_runners(&self, limit: u64, offset: u64) -> Result<RunnerPageDto, AdminError> {
let limit = limit.clamp(1, 100);
let offset = offset.min(50_000);
let total = runner::Entity::find().count(self.db().read()).await?;
let rows = runner::Entity::find()
.order_by_desc(runner::Column::CreatedAt)
.limit(limit)
.offset(offset)
.all(self.db().read())
.await?;
let mut items = Vec::with_capacity(rows.len());
for row in rows {
let active = workflow_job::Entity::find()
.filter(workflow_job::Column::RunnerId.eq(row.id))
.filter(workflow_job::Column::State.is_in([
models::ci::workflow_job::WorkflowJobState::Claimed,
models::ci::workflow_job::WorkflowJobState::Running,
]))
.count(self.db().read())
.await? as i64;
items.push(RunnerDto::from_model(row, active));
}
Ok(RunnerPageDto {
items,
total,
limit,
offset,
})
}
pub async fn update_runner(
&self,
id: Uuid,
cmd: &UpdateRunnerCmd,
) -> Result<RunnerDto, AdminError> {
let row = runner::Entity::find_by_id(id)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let mut am: runner::ActiveModel = row.into();
if let Some(labels) = &cmd.labels {
let labels: Vec<String> = labels
.iter()
.map(|l| l.trim().to_string())
.filter(|l| !l.is_empty())
.collect();
if labels.is_empty() {
return Err(AdminError::BadRequest(
"runner labels must not be empty".to_string(),
));
}
am.labels = Set(serde_json::to_value(labels)
.map_err(|e| AdminError::Internal(format!("encoding labels failed: {e}")))?);
}
if let Some(capacity) = cmd.capacity {
if capacity <= 0 || capacity > 256 {
return Err(AdminError::BadRequest(
"runner capacity must be between 1 and 256".to_string(),
));
}
am.capacity = Set(capacity);
}
if let Some(enabled) = cmd.enabled {
am.state = Set(if enabled {
RunnerState::Active
} else {
RunnerState::Disabled
});
}
am.updated_at = Set(OffsetDateTime::now_utc());
let updated = am.update(self.db().write()).await?;
Ok(RunnerDto::from_model(updated, 0))
}
pub async fn delete_runner(&self, id: Uuid) -> Result<(), AdminError> {
let _ = runner_token::Entity::delete_many()
.filter(runner_token::Column::RunnerId.eq(id))
.exec(self.db().write())
.await;
let deleted = runner::Entity::delete_by_id(id)
.exec(self.db().write())
.await?;
if deleted.rows_affected == 0 {
return Err(AdminError::NotFound);
}
Ok(())
}
pub async fn list_runner_tokens(&self) -> Result<Vec<RunnerTokenDto>, AdminError> {
let rows = runner_token::Entity::find()
.filter(runner_token::Column::RunnerId.is_null())
.order_by_desc(runner_token::Column::CreatedAt)
.all(self.db().read())
.await?;
Ok(rows.into_iter().map(token_dto).collect())
}
pub async fn create_runner_token(
&self,
cmd: &CreateRunnerTokenCmd,
) -> Result<CreatedRunnerTokenDto, AdminError> {
let name = cmd.name.trim();
if name.is_empty() || name.len() > 200 {
return Err(AdminError::BadRequest(
"token name must be 1-200 characters".to_string(),
));
}
let id = Uuid::now_v7();
let (credential, hash, salt, last_eight) = ci::runners::generate_credential(id);
let secret = credential
.split_once('.')
.map(|(_, secret)| secret.to_string())
.unwrap_or_default();
let token = format!("{id}.{secret}");
let now = OffsetDateTime::now_utc();
let expires_at = cmd
.expires_in_days
.filter(|days| *days > 0)
.map(|days| now + time::Duration::days(days));
let model = runner_token::ActiveModel {
id: Set(id),
runner_id: Set(None),
name: Set(name.to_string()),
token_hash: Set(hash),
token_salt: Set(salt),
token_last_eight: Set(last_eight),
space_id: Set(cmd.space_id),
single_use: Set(true),
used_at: Set(None),
revoked_at: Set(None),
expires_at: Set(expires_at),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db().write())
.await?;
Ok(CreatedRunnerTokenDto {
token: token_dto(model),
plaintext: token,
})
}
pub async fn revoke_runner_token(&self, id: Uuid) -> Result<(), AdminError> {
let row = runner_token::Entity::find_by_id(id)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let mut am: runner_token::ActiveModel = row.into();
am.revoked_at = Set(Some(OffsetDateTime::now_utc()));
am.updated_at = Set(OffsetDateTime::now_utc());
am.update(self.db().write()).await?;
Ok(())
}
pub async fn list_environments(
&self,
space_id: Uuid,
) -> Result<Vec<EnvironmentAdminDto>, AdminError> {
let rows = ci_environment::Entity::find()
.filter(ci_environment::Column::SpaceId.eq(space_id))
.order_by_asc(ci_environment::Column::Name)
.all(self.db().read())
.await?;
let mut out = Vec::with_capacity(rows.len());
for row in rows {
let secrets = ci_secret::Entity::find()
.filter(ci_secret::Column::EnvironmentId.eq(row.id))
.filter(ci_secret::Column::Scope.eq(CiSecretScope::Environment))
.all(self.db().read())
.await?
.into_iter()
.map(|s| s.name)
.collect();
out.push(EnvironmentAdminDto {
id: row.id,
space_id: row.space_id,
name: row.name,
description: row.description,
required_labels: serde_json::from_value(row.required_labels).unwrap_or_default(),
secrets,
created_at: row.created_at,
updated_at: row.updated_at,
});
}
Ok(out)
}
pub async fn upsert_environment(
&self,
space_id: Uuid,
cmd: &UpsertEnvironmentAdminCmd,
) -> Result<EnvironmentAdminDto, AdminError> {
let service = ci::CiService::new(self.state.clone());
service
.upsert_environment(
space_id,
&cmd.name,
cmd.description.clone(),
cmd.required_labels.clone(),
)
.await
.map_err(map_ci_error)?;
let list = self.list_environments(space_id).await?;
list.into_iter()
.find(|e| e.name == cmd.name.trim())
.ok_or(AdminError::NotFound)
}
pub async fn delete_environment(&self, id: Uuid) -> Result<(), AdminError> {
let service = ci::CiService::new(self.state.clone());
service.delete_environment(id).await.map_err(map_ci_error)?;
Ok(())
}
pub async fn upsert_environment_secret(
&self,
environment_id: Uuid,
cmd: &UpsertEnvironmentSecretCmd,
actor: Option<Uuid>,
) -> Result<(), AdminError> {
let service = ci::CiService::new(self.state.clone());
service
.upsert_environment_secret(environment_id, &cmd.name, &cmd.value, actor)
.await
.map_err(map_ci_error)?;
Ok(())
}
pub async fn delete_environment_secret(
&self,
environment_id: Uuid,
name: &str,
) -> Result<(), AdminError> {
let row = ci_secret::Entity::find()
.filter(ci_secret::Column::EnvironmentId.eq(environment_id))
.filter(ci_secret::Column::Name.eq(name.to_string()))
.one(self.db().read())
.await?;
let Some(row) = row else {
return Err(AdminError::NotFound);
};
ci_secret::Entity::delete_by_id(row.id)
.exec(self.db().write())
.await?;
Ok(())
}
pub async fn pipeline_run_counts(&self) -> Result<(i64, i64, i64, i64), AdminError> {
let queued = workflow_run::Entity::find()
.filter(workflow_run::Column::State.eq(WorkflowRunState::Queued))
.count(self.db().read())
.await? as i64;
let running = workflow_run::Entity::find()
.filter(workflow_run::Column::State.eq(WorkflowRunState::Running))
.count(self.db().read())
.await? as i64;
let failed = workflow_run::Entity::find()
.filter(workflow_run::Column::State.eq(WorkflowRunState::Failure))
.count(self.db().read())
.await? as i64;
let succeeded = workflow_run::Entity::find()
.filter(workflow_run::Column::State.eq(WorkflowRunState::Success))
.count(self.db().read())
.await? as i64;
Ok((queued, running, failed, succeeded))
}
#[allow(dead_code)]
fn _touch(_expr: sea_orm::sea_query::SimpleExpr) {
let _ = Expr::value(0);
}
}
impl AdminService {
pub async fn environment_space(&self, id: Uuid) -> Result<Uuid, AdminError> {
let row = ci_environment::Entity::find_by_id(id)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
Ok(row.space_id)
}
}