89109d2
use ci_protocol::events::{CiEvent, TriggerEvent};
use models::ci::workflow_run::WorkflowEventKind;
use models::ci::{actions_schedule, workflow_run};
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set, TransactionTrait,
};
use serde_json::{Value, json};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::{
CiService, LoadedWorkflow,
errors::{CiError, CiResult},
runs::RunSeed,
};
#[derive(Debug, Clone, serde::Deserialize, utoipa::ToSchema)]
pub struct DispatchCmd {
pub event_type: String,
#[serde(default)]
pub client_payload: Value,
}
impl CiService {
pub async fn repository_dispatch(
&self,
repo: &models::repo::repos::Model,
space: &models::space::spaces::Model,
caller: i64,
command: &DispatchCmd,
) -> CiResult<Vec<Uuid>> {
self.ensure_enabled()?;
if command.event_type.trim().is_empty()
|| command.event_type.len() > 100
|| serde_json::to_vec(&command.client_payload)?.len() > 65536
{
return Err(CiError::InvalidArgument(
"invalid repository_dispatch event type or payload".into(),
));
}
let event = TriggerEvent {
id: Uuid::now_v7(),
repo_id: repo.id,
space_id: space.id,
actor_id: Some(caller),
event: "repository_dispatch".into(),
payload: json!({"action":command.event_type,"client_payload":command.client_payload}),
trace: export::telemetry::traceparent(),
request_id: export::telemetry::request_id(),
};
self.handle_trigger_event(&event).await
}
pub async fn handle_trigger_event(&self, event: &TriggerEvent) -> CiResult<Vec<Uuid>> {
let repo = self.repo_by_id(event.repo_id).await?;
let space = self.space_of_repo(&repo).await?;
if space.id != event.space_id {
return Err(CiError::InvalidArgument(
"event space does not match repository".into(),
));
}
let git = git::service::GitService::new(self.state.clone());
let sha = git
.rev_parse(
space.lower_name.clone(),
repo.name.clone(),
repo.default_branch.clone(),
)
.await?;
let workflows = self.load_valid_workflows(&space, &repo.name, &sha).await?;
let kind = match event.event.as_str() {
"repository_dispatch" => WorkflowEventKind::RepositoryDispatch,
"workflow_run" => WorkflowEventKind::WorkflowRun,
_ => return Err(CiError::InvalidArgument("unsupported event source".into())),
};
let mut payload = event.payload.clone();
payload["event_id"] = json!(event.id);
let seed = RunSeed {
event: kind,
event_payload: payload,
commit_sha: sha,
git_ref: format!("refs/heads/{}", repo.default_branch),
branch: event.payload["workflow_run"]["head_branch"]
.as_str()
.map(str::to_owned)
.or_else(|| Some(repo.default_branch.clone())),
tag: None,
base_sha: None,
base_ref: None,
pr_number: None,
trigger_user_id: event.actor_id,
};
let mut created = Vec::new();
for workflow in workflows {
if self
.actions_event_matches(&workflow, &seed, Vec::new())
.await?
{
created.push(self.create_run(&repo, &space, &seed, &workflow).await?)
}
}
Ok(created)
}
pub async fn actions_event_matches(
&self,
workflow: &LoadedWorkflow,
seed: &RunSeed,
paths: Vec<String>,
) -> CiResult<bool> {
let Some(actions) = &workflow.actions else {
return Ok(false);
};
let action = seed.event_payload["action"].as_str().unwrap_or_default();
let result=self.actions_engine(json!({"op":"trigger","source":actions.source,"file":actions.file,"trigger_event":seed.event.as_str(),"branch":if seed.event==WorkflowEventKind::PullRequest{seed.base_ref.as_deref()}else{seed.branch.as_deref()},"tag":seed.tag,"action":action,"paths":paths,"expression":seed.event_payload["workflow_run"]["name"].as_str().unwrap_or_default()})).await?;
Ok(result.as_bool().unwrap_or(false))
}
pub async fn actions_changed_paths(
&self,
repo_id: Uuid,
seed: &RunSeed,
) -> CiResult<Vec<String>> {
let repo = models::repo::repos::Entity::find_by_id(repo_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
let space = self.space_of_repo(&repo).await?;
let service = git::service::GitService::new(self.state.clone());
let before = seed
.base_sha
.as_deref()
.or_else(|| seed.event_payload["old_oid"].as_str())
.filter(|oid| !oid.is_empty() && !oid.chars().all(|ch| ch == '0'));
let paths = match before {
Some(before) => {
service
.diff_files_changed_between(
space.lower_name,
repo.name,
before.to_string(),
seed.commit_sha.clone(),
)
.await?
}
None => service
.tree_walk(
space.lower_name,
repo.name,
seed.commit_sha.clone(),
true,
false,
)
.await?
.into_iter()
.map(|entry| entry.name)
.collect(),
};
Ok(paths)
}
pub async fn sync_actions_schedules(
&self,
repo_id: Uuid,
workflows: &[LoadedWorkflow],
) -> CiResult<()> {
let existing = actions_schedule::Entity::find()
.filter(actions_schedule::Column::RepoId.eq(repo_id))
.all(self.db_write())
.await?;
let now = OffsetDateTime::now_utc();
let mut wanted = std::collections::BTreeSet::new();
for workflow in workflows {
let Some(actions) = &workflow.actions else {
continue;
};
for schedule in actions.document["on"]["schedule"]
.as_array()
.into_iter()
.flatten()
{
let cron = schedule["cron"]
.as_str()
.ok_or_else(|| CiError::InvalidArgument("schedule cron is required".into()))?;
wanted.insert((workflow.file.clone(), cron.to_owned()));
if existing
.iter()
.any(|row| row.workflow_file == workflow.file && row.cron == cron)
{
continue;
}
let next = self
.actions_engine(
json!({"op":"cron","expression":cron,"value":now.unix_timestamp()}),
)
.await?
.as_i64()
.ok_or_else(|| CiError::InvalidArgument("invalid cron next time".into()))?;
actions_schedule::ActiveModel {
id: Set(Uuid::now_v7()),
repo_id: Set(repo_id),
workflow_file: Set(workflow.file.clone()),
cron: Set(cron.into()),
next_at: Set(OffsetDateTime::from_unix_timestamp(next)
.map_err(|error| CiError::InvalidArgument(error.to_string()))?),
updated_at: Set(now),
}
.insert(self.db_write())
.await?;
}
}
for row in existing {
if !wanted.contains(&(row.workflow_file, row.cron)) {
actions_schedule::Entity::delete_by_id(row.id)
.exec(self.db_write())
.await?;
}
}
Ok(())
}
#[tracing::instrument(name = "ci.discover_schedules", skip_all)]
pub async fn discover_actions_schedules(&self) -> CiResult<()> {
let repos = models::repo::repos::Entity::find()
.all(self.db_read())
.await?;
for repo in repos {
let result: CiResult<()> = async {
let space = self.space_of_repo(&repo).await?;
let workflows = self
.load_valid_workflows(&space, &repo.name, &repo.default_branch)
.await?;
self.sync_actions_schedules(repo.id, &workflows).await
}
.await;
if let Err(error) = result {
tracing::warn!(repo=%repo.id,%error,"schedule discovery failed");
}
}
Ok(())
}
#[tracing::instrument(name = "ci.run_schedules", skip_all)]
pub async fn run_due_actions_schedules(&self) -> CiResult<u64> {
let now = OffsetDateTime::now_utc();
let due = actions_schedule::Entity::find()
.filter(actions_schedule::Column::NextAt.lte(now))
.all(self.db_write())
.await?;
let mut count = 0;
for due in due {
let txn = self.db_write().begin().await?;
let Some(current) = actions_schedule::Entity::find_by_id(due.id)
.lock_exclusive()
.one(&txn)
.await?
else {
continue;
};
if current.next_at > now {
continue;
}
let repo = self.repo_by_id(current.repo_id).await?;
let space = self.space_of_repo(&repo).await?;
let git = git::service::GitService::new(self.state.clone());
let sha = git
.rev_parse(
space.lower_name.clone(),
repo.name.clone(),
repo.default_branch.clone(),
)
.await?;
let workflow = self
.load_workflow_by_file(&space, &repo.name, &sha, ¤t.workflow_file)
.await?;
let still_declared = workflow.actions.as_ref().is_some_and(|actions| {
actions.document["on"]["schedule"]
.as_array()
.is_some_and(|values| {
values
.iter()
.any(|value| value["cron"].as_str() == Some(current.cron.as_str()))
})
});
if !still_declared {
actions_schedule::Entity::delete_by_id(current.id)
.exec(&txn)
.await?;
txn.commit().await?;
continue;
}
let seed = RunSeed {
event: WorkflowEventKind::Schedule,
event_payload: json!({"schedule":current.cron,"event_id":format!("schedule:{}:{}",current.id,current.next_at.unix_timestamp())}),
commit_sha: sha,
git_ref: format!("refs/heads/{}", repo.default_branch),
branch: Some(repo.default_branch.clone()),
tag: None,
base_sha: None,
base_ref: None,
pr_number: None,
trigger_user_id: None,
};
self.create_run(&repo, &space, &seed, &workflow).await?;
let next = self
.actions_engine(
json!({"op":"cron","expression":current.cron,"value":now.unix_timestamp()}),
)
.await?
.as_i64()
.ok_or_else(|| CiError::InvalidArgument("invalid cron next time".into()))?;
let mut active: actions_schedule::ActiveModel = current.into();
active.next_at = Set(OffsetDateTime::from_unix_timestamp(next)
.map_err(|error| CiError::InvalidArgument(error.to_string()))?);
active.updated_at = Set(now);
active.update(&txn).await?;
txn.commit().await?;
count += 1;
}
Ok(count)
}
pub async fn emit_actions_workflow_run(
&self,
run: &workflow_run::Model,
action: &str,
) -> CiResult<()> {
if run.event_payload["_workflow_run_depth"]
.as_u64()
.unwrap_or(0)
>= 3
{
return Ok(());
}
let repo = self.repo_by_id(run.repo_id).await?;
let space = self.space_of_repo(&repo).await?;
use sha2::Digest;
let hash = sha2::Sha256::digest(
format!("workflow_run:{}:{action}:{}", run.id, run.attempt).as_bytes(),
);
let mut identity = [0u8; 16];
identity.copy_from_slice(&hash[..16]);
let event = CiEvent::Trigger(TriggerEvent {
id: Uuid::from_bytes(identity),
repo_id: repo.id,
space_id: space.id,
actor_id: run.trigger_user_id,
event: "workflow_run".into(),
payload: json!({"action":action,"_workflow_run_depth":run.event_payload["_workflow_run_depth"].as_u64().unwrap_or(0)+1,"workflow_run":{"id":run.actions_id,"name":run.workflow_name,"head_branch":run.branch,"head_sha":run.commit_sha,"status":if action=="completed"{"completed"}else if action=="requested"{"queued"}else{"in_progress"},"conclusion":if action=="completed"{Some(crate::share::states::run_state_str(run.state))}else{None},"event":run.event.as_str()}}),
trace: export::telemetry::traceparent(),
request_id: export::telemetry::request_id(),
});
crate::events::publish_event(
&self.state.redis,
&self.config().ci.event_queue_topic,
&event,
)
.await
}
}