563bb99
use models::ci::ci_log_chunk;
use models::ci::workflow_job;
use sea_orm::{
ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set,
};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::errors::{CiError, CiResult};
use crate::service::CiService;
#[derive(Debug, Clone, serde::Serialize, utoipa::ToSchema)]
pub struct LogPageDto {
pub chunks: Vec<LogChunkDto>,
pub next_seq: i32,
pub complete: bool,
}
#[derive(Debug, Clone, serde::Serialize, utoipa::ToSchema)]
pub struct LogChunkDto {
pub seq: i32,
pub text: String,
}
impl CiService {
pub async fn append_log_chunks(
&self,
job_id: Uuid,
runner_id: Uuid,
chunks: &[ci_protocol::runner::LogChunk],
) -> CiResult<()> {
self.ensure_job_owned(job_id, runner_id).await?;
let job = workflow_job::Entity::find_by_id(job_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
let secrets = self.job_secret_values(job_id).await?;
let limit = self.config().ci.log_chunk_max_bytes;
for chunk in chunks {
let text = crate::secrets::redact(&chunk.text, &secrets);
if text.len() as u64 > limit {
return Err(CiError::InvalidArgument(
"log chunk exceeds the configured limit".to_string(),
));
}
let existing = ci_log_chunk::Entity::find()
.filter(ci_log_chunk::Column::JobId.eq(job_id))
.filter(ci_log_chunk::Column::Seq.eq(chunk.seq))
.one(self.db_read())
.await?;
if existing.is_some() {
continue;
}
let key = Self::log_object_key(job.run_id, job_id, chunk.seq);
self.storage()
.client()
.put_with_content_type(&key, text.as_bytes(), "text/plain; charset=utf-8")
.await
.map_err(CiError::from)?;
let now = OffsetDateTime::now_utc();
ci_log_chunk::ActiveModel {
id: Set(Uuid::now_v7()),
job_id: Set(job_id),
seq: Set(chunk.seq),
object_key: Set(key),
size: Set(text.len() as i64),
created_at: Set(now),
}
.insert(self.db_write())
.await?;
}
Ok(())
}
pub async fn read_job_logs(
&self,
job_id: Uuid,
after_seq: i32,
limit: u64,
) -> CiResult<LogPageDto> {
let limit = limit.clamp(1, 100) as u64;
let rows = ci_log_chunk::Entity::find()
.filter(ci_log_chunk::Column::JobId.eq(job_id))
.filter(ci_log_chunk::Column::Seq.gt(after_seq))
.order_by_asc(ci_log_chunk::Column::Seq)
.limit(limit)
.all(self.db_read())
.await?;
let mut chunks = Vec::with_capacity(rows.len());
let mut next_seq = after_seq;
for row in rows {
let data = self
.storage()
.client()
.get(&row.object_key)
.await
.map_err(CiError::from)?;
let text = String::from_utf8_lossy(&data).into_owned();
next_seq = row.seq;
chunks.push(LogChunkDto { seq: row.seq, text });
}
let job = workflow_job::Entity::find_by_id(job_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
let complete = job.state.is_terminal();
Ok(LogPageDto {
chunks,
next_seq,
complete,
})
}
pub async fn job_secret_values(&self, job_id: Uuid) -> CiResult<Vec<String>> {
let job = workflow_job::Entity::find_by_id(job_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
let mut out = Vec::new();
for secret in self
.resolved_secrets(job.repo_id, job.space_id, job.environment.as_deref())
.await?
{
out.push(secret.1);
}
Ok(out)
}
}