80f9e01
use models::ci::workflow_artifact::{self, ArtifactState};
use models::ci::workflow_run;
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, Set};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::errors::{CiError, CiResult};
use crate::service::CiService;
impl CiService {
pub fn log_object_key(run_id: Uuid, job_id: Uuid, seq: i32) -> String {
format!("ci/logs/{run_id}/{job_id}/{seq:06}.log")
}
pub fn artifact_object_key(run_id: Uuid, job_id: Uuid, artifact_id: Uuid) -> String {
format!("ci/artifacts/{run_id}/{job_id}/{artifact_id}")
}
pub async fn init_artifact(
&self,
job_id: Uuid,
runner_id: Uuid,
name: &str,
size: i64,
sha256: Option<String>,
) -> CiResult<(Uuid, String)> {
self.ensure_job_owned(job_id, runner_id).await?;
if size < 0 {
return Err(CiError::InvalidArgument(
"artifact size must not be negative".to_string(),
));
}
if size as u64 > self.config().ci.artifact_max_bytes {
return Err(CiError::InvalidArgument(
"artifact exceeds the configured size limit".to_string(),
));
}
let job = models::ci::workflow_job::Entity::find_by_id(job_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
let artifact_id = Uuid::now_v7();
let key = Self::artifact_object_key(job.run_id, job_id, artifact_id);
let now = OffsetDateTime::now_utc();
let expires_at =
now + time::Duration::days(self.config().ci.artifact_retention_days.max(1));
workflow_artifact::ActiveModel {
id: Set(artifact_id),
run_id: Set(job.run_id),
job_id: Set(job_id),
name: Set(name.to_string()),
object_key: Set(key.clone()),
size: Set(size),
sha256: Set(sha256),
mime_type: Set(None),
state: Set(ArtifactState::Uploading),
expires_at: Set(Some(expires_at)),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db_write())
.await?;
let url = self
.storage()
.client()
.presign_put(&key, 3600)
.map_err(CiError::from)?;
Ok((artifact_id, url))
}
pub async fn complete_artifact(
&self,
job_id: Uuid,
runner_id: Uuid,
artifact_id: Uuid,
size: i64,
sha256: Option<String>,
) -> CiResult<()> {
self.ensure_job_owned(job_id, runner_id).await?;
let row = workflow_artifact::Entity::find_by_id(artifact_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
if row.job_id != job_id {
return Err(CiError::NotFound);
}
let exists = self
.storage()
.client()
.exists(&row.object_key)
.await
.map_err(CiError::from)?;
if !exists {
return Err(CiError::InvalidArgument(
"artifact object was not uploaded".to_string(),
));
}
let mut am: workflow_artifact::ActiveModel = row.into();
am.size = Set(size);
if sha256.is_some() {
am.sha256 = Set(sha256);
}
am.state = Set(ArtifactState::Ready);
am.updated_at = Set(OffsetDateTime::now_utc());
am.update(self.db_write()).await?;
Ok(())
}
pub async fn artifact_download_url(&self, artifact_id: Uuid) -> CiResult<String> {
let row = workflow_artifact::Entity::find_by_id(artifact_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
if row.state != ArtifactState::Ready {
return Err(CiError::InvalidArgument(
"artifact is not ready".to_string(),
));
}
self.storage()
.client()
.presign_get(&row.object_key, 900)
.map_err(CiError::from)
}
pub async fn artifacts_for_run(&self, run_id: Uuid) -> CiResult<Vec<workflow_artifact::Model>> {
Ok(workflow_artifact::Entity::find()
.filter(workflow_artifact::Column::RunId.eq(run_id))
.all(self.db_read())
.await?)
}
pub async fn needs_artifacts(
&self,
run_id: Uuid,
job_id: Uuid,
) -> CiResult<Vec<ci_protocol::runner::NeedsArtifactDto>> {
let job = models::ci::workflow_job::Entity::find_by_id(job_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
let needs: Vec<String> = serde_json::from_value(job.needs.clone()).unwrap_or_default();
if needs.is_empty() {
return Ok(Vec::new());
}
let jobs = models::ci::workflow_job::Entity::find()
.filter(models::ci::workflow_job::Column::RunId.eq(run_id))
.all(self.db_read())
.await?;
let mut out = Vec::new();
for need in needs {
let Some(need_job) = jobs.iter().find(|j| j.job_key == need) else {
continue;
};
let artifacts = self.artifacts_for_run(run_id).await?;
for artifact in artifacts
.into_iter()
.filter(|a| a.job_id == need_job.id && a.state == ArtifactState::Ready)
{
let download_url = self
.storage()
.client()
.presign_get(&artifact.object_key, 3600)
.map_err(CiError::from)?;
out.push(ci_protocol::runner::NeedsArtifactDto {
run_id,
job_id: need_job.id,
name: artifact.name,
download_url,
size: artifact.size,
sha256: artifact.sha256,
});
}
}
Ok(out)
}
#[tracing::instrument(name = "ci.prune", skip_all)]
pub async fn prune_expired(&self) -> CiResult<u64> {
use models::ci::ci_log_chunk;
use models::ci::workflow_artifact as artifact_entity;
let now = OffsetDateTime::now_utc();
let mut removed = 0u64;
let log_cutoff = now - time::Duration::days(self.config().ci.log_retention_days.max(0));
let logs = ci_log_chunk::Entity::find()
.filter(ci_log_chunk::Column::CreatedAt.lt(log_cutoff))
.limit(500)
.all(self.db_read())
.await?;
for chunk in logs {
let _ = self.storage().client().delete(&chunk.object_key).await;
let _ = ci_log_chunk::Entity::delete_by_id(chunk.id)
.exec(self.db_write())
.await;
removed += 1;
}
let artifact_cutoff =
now - time::Duration::days(self.config().ci.artifact_retention_days.max(0));
let artifacts = artifact_entity::Entity::find()
.filter(artifact_entity::Column::CreatedAt.lt(artifact_cutoff))
.limit(500)
.all(self.db_read())
.await?;
for artifact in artifacts {
let _ = self.storage().client().delete(&artifact.object_key).await;
let _ = artifact_entity::Entity::delete_by_id(artifact.id)
.exec(self.db_write())
.await;
removed += 1;
}
let run_cutoff = now - time::Duration::days(self.config().ci.run_retention_days.max(0));
let runs = workflow_run::Entity::find()
.filter(workflow_run::Column::CreatedAt.lt(run_cutoff))
.limit(200)
.all(self.db_read())
.await?;
for run in runs {
let job_ids: Vec<Uuid> = models::ci::workflow_job::Entity::find()
.filter(models::ci::workflow_job::Column::RunId.eq(run.id))
.all(self.db_read())
.await
.unwrap_or_default()
.into_iter()
.map(|j| j.id)
.collect();
for job_id in &job_ids {
let _ = models::ci::workflow_step::Entity::delete_many()
.filter(models::ci::workflow_step::Column::JobId.eq(*job_id))
.exec(self.db_write())
.await;
let _ = models::ci::ci_log_chunk::Entity::delete_many()
.filter(models::ci::ci_log_chunk::Column::JobId.eq(*job_id))
.exec(self.db_write())
.await;
}
let _ = models::ci::workflow_job::Entity::delete_many()
.filter(models::ci::workflow_job::Column::RunId.eq(run.id))
.exec(self.db_write())
.await;
let _ = models::ci::ci_deployment::Entity::delete_many()
.filter(models::ci::ci_deployment::Column::RunId.eq(run.id))
.exec(self.db_write())
.await;
let _ = workflow_run::Entity::delete_by_id(run.id)
.exec(self.db_write())
.await;
removed += 1;
}
Ok(removed)
}
}