f0a1552
ZhenYi
feat(chat): add chat module with room management, message handling, and session revocation
f0a1552
use models::repo::pipeline_run::{self, PipelineRunState};
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::service::GitService;
use crate::share::errors::GitError;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)]
#[serde(rename_all = "lowercase")]
pub enum CommitStatusState {
Pending,
Success,
Failure,
Error,
Warning,
Skipped,
}
#[derive(Debug, Clone, Serialize)]
pub struct CommitStatusDto {
pub state: CommitStatusState,
pub context: String,
pub description: Option<String>,
pub target_url: Option<String>,
pub creator: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
pub struct JobDto {
pub id: String,
pub name: String,
pub state: CommitStatusState,
pub started_at: Option<OffsetDateTime>,
pub finished_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Serialize, utoipa::ToSchema)]
pub struct TriggerResultDto {
pub job_id: String,
pub queued: bool,
}
#[derive(Debug, Clone, Serialize, utoipa::ToSchema)]
pub struct PipelineRunDto {
pub id: Uuid,
pub repo_id: Uuid,
pub trigger_user_id: Uuid,
pub commit_sha: String,
pub branch: String,
pub state: PipelineRunState,
pub started_at: Option<OffsetDateTime>,
pub finished_at: Option<OffsetDateTime>,
pub created_at: OffsetDateTime,
}
impl From<pipeline_run::Model> for PipelineRunDto {
fn from(m: pipeline_run::Model) -> Self {
Self {
id: m.id,
repo_id: m.repo_id,
trigger_user_id: m.trigger_user_id,
commit_sha: m.commit_sha,
branch: m.branch,
state: m.state,
started_at: m.started_at,
finished_at: m.finished_at,
created_at: m.created_at,
}
}
}
#[derive(Debug, Clone, Serialize)]
pub struct StatusListDto {
pub statuses: Vec<CommitStatusDto>,
pub total: u64,
}
impl GitService {
async fn insert_run(
&self,
repo_id: Uuid,
caller: Uuid,
commit_sha: String,
branch: String,
) -> Result<Uuid, GitError> {
let id = Uuid::now_v7();
let now = OffsetDateTime::now_utc();
pipeline_run::ActiveModel {
id: Set(id),
repo_id: Set(repo_id),
trigger_user_id: Set(caller),
commit_sha: Set(commit_sha),
branch: Set(branch),
state: Set(PipelineRunState::Queued),
started_at: Set(None),
finished_at: Set(None),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db_write())
.await?;
Ok(id)
}
pub async fn pipeline_trigger(
&self,
caller: Uuid,
space_name: String,
repo_name: String,
commit: String,
branch: String,
) -> Result<TriggerResultDto, GitError> {
let ctx = self.context(space_name, repo_name).await?;
let job_id = self
.insert_run(ctx.model.id, caller, commit, branch)
.await?;
Ok(TriggerResultDto {
job_id: job_id.to_string(),
queued: true,
})
}
pub async fn pipeline_schedule(
&self,
caller: Uuid,
space_name: String,
repo_name: String,
commit: String,
job: JobDto,
) -> Result<TriggerResultDto, GitError> {
let ctx = self.context(space_name, repo_name).await?;
let job_id = self
.insert_run(ctx.model.id, caller, commit, job.name)
.await?;
Ok(TriggerResultDto {
job_id: job_id.to_string(),
queued: true,
})
}
pub async fn pipeline_cancel(&self, job_id: String) -> Result<(), GitError> {
let id = Uuid::parse_str(&job_id).map_err(|_| GitError::NotFound)?;
let row = pipeline_run::Entity::find_by_id(id)
.one(self.db_read())
.await?
.ok_or(GitError::NotFound)?;
match row.state {
PipelineRunState::Queued | PipelineRunState::Running => {}
_ => {
return Err(GitError::Conflict(
"pipeline run already finished".to_string(),
))
}
}
let mut am: pipeline_run::ActiveModel = row.into();
am.state = Set(PipelineRunState::CancelRequested);
am.updated_at = Set(OffsetDateTime::now_utc());
am.update(self.db_write()).await?;
Ok(())
}
pub async fn list_pipeline_runs(
&self,
space_name: String,
repo_name: String,
commit: Option<String>,
branch: Option<String>,
state: Option<PipelineRunState>,
limit: Option<u64>,
offset: Option<u64>,
) -> Result<Vec<PipelineRunDto>, GitError> {
let ctx = self.context(space_name, repo_name).await?;
let mut q = pipeline_run::Entity::find()
.filter(pipeline_run::Column::RepoId.eq(ctx.model.id))
.order_by_desc(pipeline_run::Column::CreatedAt);
if let Some(commit) = commit.filter(|c| !c.is_empty()) {
q = q.filter(pipeline_run::Column::CommitSha.eq(commit));
}
if let Some(branch) = branch.filter(|b| !b.is_empty()) {
q = q.filter(pipeline_run::Column::Branch.eq(branch));
}
if let Some(state) = state {
q = q.filter(pipeline_run::Column::State.eq(state));
}
let limit = limit.unwrap_or(50).clamp(1, 100);
let offset = offset.unwrap_or(0).min(5000);
let rows = q.limit(limit).offset(offset).all(self.db_read()).await?;
Ok(rows.into_iter().map(PipelineRunDto::from).collect())
}
pub async fn get_pipeline_run(&self, job_id: String) -> Result<PipelineRunDto, GitError> {
let id = Uuid::parse_str(&job_id).map_err(|_| GitError::NotFound)?;
let row = pipeline_run::Entity::find_by_id(id)
.one(self.db_read())
.await?
.ok_or(GitError::NotFound)?;
Ok(PipelineRunDto::from(row))
}
pub async fn report_pipeline_run(
&self,
job_id: String,
state: PipelineRunState,
started_at: Option<OffsetDateTime>,
finished_at: Option<OffsetDateTime>,
) -> Result<PipelineRunDto, GitError> {
let id = Uuid::parse_str(&job_id).map_err(|_| GitError::NotFound)?;
let row = pipeline_run::Entity::find_by_id(id)
.one(self.db_read())
.await?
.ok_or(GitError::NotFound)?;
let mut am: pipeline_run::ActiveModel = row.into();
am.state = Set(state);
if let Some(t) = started_at {
am.started_at = Set(Some(t));
}
if let Some(t) = finished_at {
am.finished_at = Set(Some(t));
}
am.updated_at = Set(OffsetDateTime::now_utc());
let updated = am.update(self.db_write()).await?;
Ok(PipelineRunDto::from(updated))
}
}