fad06fd
use std::collections::{BTreeMap, BTreeSet};
use models::ci::runner::{self, RunnerState};
use models::ci::workflow_job::{self, WorkflowJobState};
use models::ci::workflow_run::{self, WorkflowRunState};
use models::dir::dirs;
use models::repo::repos;
use models::space::spaces;
use sea_orm::{
ColumnTrait, Condition, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect,
};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use crate::errors::AdminError;
use crate::services::runners::map_ci_error;
use crate::services::service::AdminService;
#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct ListActRunsQuery {
pub q: Option<String>,
pub state: Option<WorkflowRunState>,
pub repo_id: Option<Uuid>,
pub limit: Option<u64>,
pub offset: Option<u64>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminActSummary {
pub runners_total: u64,
pub runners_active: u64,
pub runners_busy: u64,
pub jobs_queued: u64,
pub jobs_running: u64,
pub runs_queued: u64,
pub runs_running: u64,
pub runs_success: u64,
pub runs_failure: u64,
pub generated_at: OffsetDateTime,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminActRunDto {
pub id: Uuid,
pub repo_id: Uuid,
pub repo_name: String,
pub repo_full_name: String,
pub space_id: Option<Uuid>,
pub space_name: Option<String>,
pub workflow_file: String,
pub workflow_name: String,
pub display_name: Option<String>,
pub actions_id: Option<i64>,
pub run_number: Option<i64>,
pub event: String,
pub state: String,
pub commit_sha: String,
pub branch: Option<String>,
pub tag: Option<String>,
pub pr_number: Option<i64>,
pub jobs_total: u64,
pub jobs_active: u64,
pub jobs_failed: u64,
pub created_at: OffsetDateTime,
pub started_at: Option<OffsetDateTime>,
pub finished_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminActRunPage {
pub items: Vec<AdminActRunDto>,
pub total: u64,
pub limit: u64,
pub offset: u64,
}
#[derive(Default)]
struct RunJobCounts {
total: u64,
active: u64,
failed: u64,
}
struct RunContext {
repo: Option<repos::Model>,
space: Option<spaces::Model>,
}
impl AdminService {
pub async fn act_summary(&self) -> Result<AdminActSummary, AdminError> {
let runners_total = runner::Entity::find().count(self.db().read()).await?;
let runners_active = runner::Entity::find()
.filter(runner::Column::State.eq(RunnerState::Active))
.count(self.db().read())
.await?;
let active_jobs = workflow_job::Entity::find()
.filter(
workflow_job::Column::State
.is_in([WorkflowJobState::Claimed, WorkflowJobState::Running]),
)
.all(self.db().read())
.await?;
let runners_busy = active_jobs
.iter()
.filter_map(|job| job.runner_id)
.collect::<BTreeSet<_>>()
.len() as u64;
let jobs_queued = workflow_job::Entity::find()
.filter(workflow_job::Column::State.eq(WorkflowJobState::Queued))
.count(self.db().read())
.await?;
let jobs_running = active_jobs.len() as u64;
let runs_queued = workflow_run::Entity::find()
.filter(workflow_run::Column::State.eq(WorkflowRunState::Queued))
.count(self.db().read())
.await?;
let runs_running = workflow_run::Entity::find()
.filter(
workflow_run::Column::State
.is_in([WorkflowRunState::Running, WorkflowRunState::CancelRequested]),
)
.count(self.db().read())
.await?;
let runs_success = workflow_run::Entity::find()
.filter(workflow_run::Column::State.eq(WorkflowRunState::Success))
.count(self.db().read())
.await?;
let runs_failure = workflow_run::Entity::find()
.filter(workflow_run::Column::State.eq(WorkflowRunState::Failure))
.count(self.db().read())
.await?;
Ok(AdminActSummary {
runners_total,
runners_active,
runners_busy,
jobs_queued,
jobs_running,
runs_queued,
runs_running,
runs_success,
runs_failure,
generated_at: OffsetDateTime::now_utc(),
})
}
pub async fn list_act_runs(
&self,
query: &ListActRunsQuery,
) -> Result<AdminActRunPage, AdminError> {
let limit = query.limit.unwrap_or(20).clamp(1, 100);
let offset = query.offset.unwrap_or(0).min(50_000);
let mut q = workflow_run::Entity::find().order_by_desc(workflow_run::Column::CreatedAt);
if let Some(term) = query
.q
.as_deref()
.map(str::trim)
.filter(|term| !term.is_empty())
{
q = q.filter(
Condition::any()
.add(workflow_run::Column::WorkflowName.contains(term))
.add(workflow_run::Column::WorkflowFile.contains(term))
.add(workflow_run::Column::CommitSha.contains(term)),
);
}
if let Some(state) = query.state {
q = q.filter(workflow_run::Column::State.eq(state));
}
if let Some(repo_id) = query.repo_id {
q = q.filter(workflow_run::Column::RepoId.eq(repo_id));
}
let total = q.clone().count(self.db().read()).await?;
let rows = q.limit(limit).offset(offset).all(self.db().read()).await?;
let run_ids: Vec<Uuid> = rows.iter().map(|run| run.id).collect();
let jobs = if run_ids.is_empty() {
Vec::new()
} else {
workflow_job::Entity::find()
.filter(workflow_job::Column::RunId.is_in(run_ids))
.all(self.db().read())
.await?
};
let mut job_counts: BTreeMap<Uuid, RunJobCounts> = BTreeMap::new();
for job in jobs {
let entry = job_counts.entry(job.run_id).or_default();
entry.total += 1;
if job.state.is_active() {
entry.active += 1;
}
if matches!(
job.state,
WorkflowJobState::Failure | WorkflowJobState::TimedOut
) {
entry.failed += 1;
}
}
let contexts = self.run_contexts(&rows).await?;
Ok(AdminActRunPage {
items: rows
.into_iter()
.map(|run| {
let counts = job_counts.remove(&run.id).unwrap_or_default();
run_summary(run, &contexts, counts)
})
.collect(),
total,
limit,
offset,
})
}
pub async fn cancel_act_run(&self, id: Uuid) -> Result<(), AdminError> {
self.require_ci_enabled()?;
let service = ci::CiService::new(self.state.clone());
service
.mark_run_cancel_requested(id)
.await
.map_err(map_ci_error)?;
service.apply_cancellations().await.map_err(map_ci_error)?;
Ok(())
}
pub async fn rerun_act_run(&self, id: Uuid) -> Result<Uuid, AdminError> {
self.require_ci_enabled()?;
let run = workflow_run::Entity::find_by_id(id)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let caller = if let Some(caller) = run.trigger_user_id {
caller
} else {
self.run_space_owner(run.repo_id).await?
};
ci::CiService::new(self.state.clone())
.rerun(id, caller)
.await
.map_err(map_ci_error)
}
fn require_ci_enabled(&self) -> Result<(), AdminError> {
if self.config().ci.enabled {
Ok(())
} else {
Err(AdminError::BadRequest("ci is disabled".to_string()))
}
}
async fn run_space_owner(&self, repo_id: Uuid) -> Result<Uuid, AdminError> {
let repo = repos::Entity::find_by_id(repo_id)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let dir = dirs::Entity::find_by_id(repo.dir_id)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let space_id = dir.space_id.ok_or(AdminError::NotFound)?;
let space = spaces::Entity::find_by_id(space_id)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
Ok(space.owner_id)
}
async fn run_contexts(
&self,
rows: &[workflow_run::Model],
) -> Result<BTreeMap<Uuid, RunContext>, AdminError> {
let repo_ids: BTreeSet<Uuid> = rows.iter().map(|run| run.repo_id).collect();
let repo_rows = if repo_ids.is_empty() {
Vec::new()
} else {
repos::Entity::find()
.filter(repos::Column::Id.is_in(repo_ids))
.all(self.db().read())
.await?
};
let repo_by_id: BTreeMap<Uuid, repos::Model> =
repo_rows.into_iter().map(|repo| (repo.id, repo)).collect();
let dir_ids: BTreeSet<Uuid> = repo_by_id.values().map(|repo| repo.dir_id).collect();
let dir_rows = if dir_ids.is_empty() {
Vec::new()
} else {
dirs::Entity::find()
.filter(dirs::Column::Id.is_in(dir_ids))
.all(self.db().read())
.await?
};
let dir_by_id: BTreeMap<Uuid, dirs::Model> =
dir_rows.into_iter().map(|dir| (dir.id, dir)).collect();
let space_ids: BTreeSet<Uuid> = dir_by_id.values().filter_map(|dir| dir.space_id).collect();
let space_rows = if space_ids.is_empty() {
Vec::new()
} else {
spaces::Entity::find()
.filter(spaces::Column::Id.is_in(space_ids))
.all(self.db().read())
.await?
};
let space_by_id: BTreeMap<Uuid, spaces::Model> = space_rows
.into_iter()
.map(|space| (space.id, space))
.collect();
Ok(repo_by_id
.into_iter()
.map(|(repo_id, repo)| {
let space = dir_by_id
.get(&repo.dir_id)
.and_then(|dir| dir.space_id)
.and_then(|space_id| space_by_id.get(&space_id).cloned());
(
repo_id,
RunContext {
repo: Some(repo),
space,
},
)
})
.collect())
}
}
fn run_summary(
run: workflow_run::Model,
contexts: &BTreeMap<Uuid, RunContext>,
counts: RunJobCounts,
) -> AdminActRunDto {
let context = contexts.get(&run.repo_id);
let repo = context.and_then(|context| context.repo.as_ref());
let space = context.and_then(|context| context.space.as_ref());
let repo_name = repo
.map(|repo| repo.name.clone())
.unwrap_or_else(|| run.repo_id.to_string());
let repo_full_name = match (space, repo) {
(Some(space), Some(repo)) => format!("{}/{}", space.lower_name, repo.lower_name),
(_, Some(repo)) => repo.lower_name.clone(),
_ => repo_name.clone(),
};
AdminActRunDto {
id: run.id,
repo_id: run.repo_id,
repo_name,
repo_full_name,
space_id: space.map(|space| space.id),
space_name: space.map(|space| space.name.clone()),
workflow_file: run.workflow_file,
workflow_name: run.workflow_name,
display_name: run
.definition
.get("_display_name")
.and_then(|value| value.as_str())
.map(str::to_owned),
actions_id: run.actions_id,
run_number: run.run_number,
event: run.event.as_str().to_string(),
state: ci::share::states::run_state_str(run.state).to_string(),
commit_sha: run.commit_sha,
branch: run.branch,
tag: run.tag,
pr_number: run.pr_number,
jobs_total: counts.total,
jobs_active: counts.active,
jobs_failed: counts.failed,
created_at: run.created_at,
started_at: run.started_at,
finished_at: run.finished_at,
}
}