d58a911
use ci_protocol::actions::{ActionsWorkflow, EngineEvent};
use serde_json::{Value, json};
use tokio::io::AsyncWriteExt;
use tokio::process::Command;
use crate::errors::{CiError, CiResult};
use crate::service::CiService;
pub async fn identity<C: sea_orm::ConnectionTrait>(
db: &C,
entity_id: uuid::Uuid,
kind: &str,
) -> CiResult<i64> {
use sea_orm::{ActiveModelTrait, Set};
let row = models::ci::actions_identity::ActiveModel {
entity_id: Set(entity_id),
kind: Set(kind.to_owned()),
..Default::default()
}
.insert(db)
.await?;
Ok(row.id)
}
impl CiService {
pub async fn report_actions_event(
&self,
job_id: uuid::Uuid,
runner_id: uuid::Uuid,
event: &EngineEvent,
) -> CiResult<()> {
use models::ci::workflow_step::WorkflowStepState as State;
use models::ci::{workflow_job, workflow_step};
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, Set,
};
let job = workflow_job::Entity::find_by_id(job_id)
.one(self.db_write())
.await?
.ok_or(CiError::NotFound)?;
if job.runner_id != Some(runner_id) || job.state.is_terminal() {
return Err(CiError::Forbidden);
}
if event.lease_token != job.lease_token {
return Err(CiError::Conflict("job lease was superseded".into()));
}
let key = event.data["step_id"].as_str().unwrap_or("--job");
let stage = event.data["stage"].as_str().unwrap_or("main");
let existing = workflow_step::Entity::find()
.filter(workflow_step::Column::JobId.eq(job_id))
.filter(workflow_step::Column::StepKey.eq(key))
.filter(workflow_step::Column::Stage.eq(stage))
.one(self.db_write())
.await?;
let is_new = existing.is_none();
let now = time::OffsetDateTime::now_utc();
let mut active = match existing {
Some(row) => row.into(),
None => workflow_step::ActiveModel {
id: Set(uuid::Uuid::now_v7()),
job_id: Set(job_id),
step_index: Set(workflow_step::Entity::find()
.filter(workflow_step::Column::JobId.eq(job_id))
.count(self.db_write())
.await? as i32),
step_key: Set(Some(key.into())),
stage: Set(Some(stage.into())),
name: Set(event.data["name"].as_str().unwrap_or(key).to_owned()),
script: Set(String::new()),
state: Set(State::Pending),
created_at: Set(now),
updated_at: Set(now),
..Default::default()
},
};
match event.event.as_str() {
"step_started" => {
active.state = Set(State::Running);
active.started_at = Set(Some(now));
}
"step_finished" => {
let conclusion = event.data["conclusion"].as_str().unwrap_or("failure");
active.state = Set(match conclusion {
"success" => State::Success,
"skipped" => State::Skipped,
"cancelled" => State::Cancelled,
_ => State::Failure,
});
active.outcome = Set(event.data["outcome"].as_str().map(str::to_owned));
active.conclusion = Set(Some(conclusion.into()));
active.outputs = Set(Some(
event.data.get("outputs").cloned().unwrap_or(json!({})),
));
active.finished_at = Set(Some(now));
active.exit_code = Set((conclusion != "skipped")
.then_some(if conclusion == "success" { 0 } else { 1 }));
}
"summary" => {
let mut summary = active
.summary
.clone()
.take()
.unwrap_or(None)
.unwrap_or_default();
summary.push_str(event.data["text"].as_str().unwrap_or_default());
if summary.len() > 1024 * 1024 {
return Err(CiError::InvalidArgument(
"step summary exceeds 1 MiB".into(),
));
}
active.summary = Set(Some(summary));
}
"annotation" => {
let mut annotations: Vec<Value> = active
.annotations
.clone()
.take()
.unwrap_or(None)
.and_then(|v| serde_json::from_value(v).ok())
.unwrap_or_default();
if annotations.len() < 1000 {
annotations.push(event.data.clone());
}
active.annotations = Set(Some(json!(annotations)));
}
_ => {
return Err(CiError::InvalidArgument(format!(
"unknown Actions event `{}`",
event.event
)));
}
}
if let Some(name) = event.data["name"].as_str() {
active.name = Set(name.into());
}
active.updated_at = Set(now);
if is_new {
active.insert(self.db_write()).await?;
} else {
active.update(self.db_write()).await?;
}
Ok(())
}
/// One `gitdata-actions-engine` process per call: one JSON line in, one out. The
/// engine is built by the separate `git-data-actions` repository; its ops are
/// `parse`, `match`, `trigger`, `cron`, `evaluate`, `expand`, `execute`,
/// `oidc_sign`, `oidc_jwks`. Any request/response field change has to land in both
/// repositories together.
pub async fn actions_engine(&self, request: Value) -> CiResult<Value> {
let config = &self.config().ci;
let mut child = Command::new(&config.actions_engine_binary)
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.kill_on_drop(true)
.spawn()
.map_err(|error| {
CiError::InvalidArgument(format!(
"Actions engine `{}`: {error}",
config.actions_engine_binary
))
})?;
let mut input = child
.stdin
.take()
.ok_or_else(|| CiError::Internal("engine stdin missing".into()))?;
input
.write_all(&serde_json::to_vec(&request)?)
.await
.map_err(|error| CiError::Internal(error.to_string()))?;
input
.write_all(b"\n")
.await
.map_err(|error| CiError::Internal(error.to_string()))?;
drop(input);
let output = tokio::time::timeout(
std::time::Duration::from_secs(config.actions_engine_timeout_secs.max(1)),
child.wait_with_output(),
)
.await
.map_err(|_| CiError::InvalidArgument("Actions engine request timed out".into()))?
.map_err(|error| CiError::Internal(error.to_string()))?;
let result = output
.stdout
.split(|byte| *byte == b'\n')
.filter_map(|line| serde_json::from_slice::<EngineEvent>(line).ok())
.find(|message| message.event == "result")
.ok_or_else(|| {
CiError::InvalidArgument(format!(
"Actions engine exited {}: {}",
output.status,
String::from_utf8_lossy(&output.stderr)
))
})?;
if let Some(error) = result.error {
return Err(CiError::InvalidArgument(error));
}
Ok(result.data)
}
pub async fn parse_actions_workflow(
&self,
file: &str,
source: &str,
) -> CiResult<ActionsWorkflow> {
let document = self
.actions_engine(json!({"op":"parse", "file":file, "source":source}))
.await?;
Ok(ActionsWorkflow {
file: file.to_owned(),
source: source.to_owned(),
document,
})
}
}