f0a1552
ZhenYi
feat(chat): add chat module with room management, message handling, and session revocation
f0a1552
use models::admin::admin_user::AdminRole;
use models::chat::webhook_deliveries::{self, WebhookDeliveryStatus};
use models::import::job;
use models::repo::pipeline_run::{self, PipelineRunState};
use models::search::index_job;
use models::system::system_tasks::{self, TaskStatus};
use models::users::user_deletion_requests::{self, UserDeletionRequestStatus};
use sea_orm::sea_query::Expr;
use sea_orm::{ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter};
use serde::{Deserialize, Serialize};
use serde_json::json;
use utoipa::ToSchema;
use crate::errors::AdminError;
use crate::services::audit::{AuditCtx, AuditRecord};
use crate::services::service::AdminService;
use crate::share::roles::AdminActor;
pub const QUEUE_GIT_SYNC: &str = "git_sync";
pub const QUEUE_EMAIL: &str = "email";
pub const QUEUE_SEARCH_INDEX: &str = "search_index";
pub const QUEUE_IMPORT: &str = "import";
pub const QUEUE_CHAT_WEBHOOK: &str = "chat_webhook";
pub const QUEUE_PIPELINE: &str = "pipeline";
pub const QUEUE_DELETION: &str = "deletion";
pub const QUEUE_SYSTEM_TASKS: &str = "system_tasks";
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct QueueSnapshotDto {
pub name: String,
pub kind: String,
pub enabled: bool,
pub depth: Option<i64>,
pub queued: i64,
pub running: i64,
pub failed: i64,
pub succeeded: i64,
pub note: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct QueueActionResp {
pub queue: String,
pub affected: u64,
}
impl AdminService {
async fn git_sync_depth(&self) -> Result<i64, AdminError> {
let topic = self.config().git.sync.queue_topic.clone();
let mut conn = self.redis().connection_clone();
let depth: i64 = match &mut conn {
export::RedisConnection::Standalone(connection) => {
redis::cmd("LLEN").arg(&topic).query_async(connection).await?
}
export::RedisConnection::Cluster(connection) => {
redis::cmd("LLEN").arg(&topic).query_async(connection).await?
}
};
Ok(depth)
}
pub async fn queue_snapshots(&self) -> Result<Vec<QueueSnapshotDto>, AdminError> {
let mut snapshots = Vec::new();
let depth = self.git_sync_depth().await?;
snapshots.push(QueueSnapshotDto {
name: QUEUE_GIT_SYNC.to_string(),
kind: "redis_list".to_string(),
enabled: true,
depth: Some(depth),
queued: depth,
running: 0,
failed: 0,
succeeded: 0,
note: Some(format!(
"topic={} concurrency={}",
self.config().git.sync.queue_topic,
self.config().git.sync.concurrency.max(1)
)),
});
snapshots.push(QueueSnapshotDto {
name: QUEUE_EMAIL.to_string(),
kind: "redis_pubsub".to_string(),
enabled: self.config().email_queue.enabled(),
depth: None,
queued: 0,
running: 0,
failed: 0,
succeeded: 0,
note: Some("redis pub/sub fan-out, no persistent backlog".to_string()),
});
let search_queued = index_job::Entity::find()
.filter(index_job::Column::State.eq("queued"))
.count(self.db().read())
.await? as i64;
let search_processing = index_job::Entity::find()
.filter(index_job::Column::State.eq("processing"))
.count(self.db().read())
.await? as i64;
let search_succeeded = index_job::Entity::find()
.filter(index_job::Column::State.eq("succeeded"))
.count(self.db().read())
.await? as i64;
let search_failed = index_job::Entity::find()
.filter(index_job::Column::LastError.is_not_null())
.count(self.db().read())
.await? as i64;
snapshots.push(QueueSnapshotDto {
name: QUEUE_SEARCH_INDEX.to_string(),
kind: "db".to_string(),
enabled: self.config().search.enabled,
depth: Some(search_queued + search_processing),
queued: search_queued,
running: search_processing,
failed: search_failed,
succeeded: search_succeeded,
note: None,
});
let import_queued = job::Entity::find()
.filter(job::Column::State.eq("queued"))
.count(self.db().read())
.await? as i64;
let import_processing = job::Entity::find()
.filter(job::Column::State.eq("processing"))
.count(self.db().read())
.await? as i64;
let import_failed = job::Entity::find()
.filter(job::Column::State.eq("failed"))
.count(self.db().read())
.await? as i64;
let import_succeeded = job::Entity::find()
.filter(job::Column::State.eq("succeeded"))
.count(self.db().read())
.await? as i64;
snapshots.push(QueueSnapshotDto {
name: QUEUE_IMPORT.to_string(),
kind: "db".to_string(),
enabled: self.config().import.enabled,
depth: Some(import_queued + import_processing),
queued: import_queued,
running: import_processing,
failed: import_failed,
succeeded: import_succeeded,
note: Some(format!(
"max_concurrent_jobs={}",
self.config().import.max_concurrent_jobs.max(1)
)),
});
let webhook_pending = webhook_deliveries::Entity::find()
.filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
.count(self.db().read())
.await? as i64;
let webhook_running = webhook_deliveries::Entity::find()
.filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Delivering))
.count(self.db().read())
.await? as i64;
let webhook_failed = webhook_deliveries::Entity::find()
.filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Failed))
.count(self.db().read())
.await? as i64;
let webhook_done = webhook_deliveries::Entity::find()
.filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Done))
.count(self.db().read())
.await? as i64;
snapshots.push(QueueSnapshotDto {
name: QUEUE_CHAT_WEBHOOK.to_string(),
kind: "db".to_string(),
enabled: true,
depth: Some(webhook_pending + webhook_running),
queued: webhook_pending,
running: webhook_running,
failed: webhook_failed,
succeeded: webhook_done,
note: None,
});
let pipeline_queued = pipeline_run::Entity::find()
.filter(pipeline_run::Column::State.eq(PipelineRunState::Queued))
.count(self.db().read())
.await? as i64;
let pipeline_running = pipeline_run::Entity::find()
.filter(pipeline_run::Column::State.eq(PipelineRunState::Running))
.count(self.db().read())
.await? as i64;
let pipeline_failed = pipeline_run::Entity::find()
.filter(pipeline_run::Column::State.eq(PipelineRunState::Failure))
.count(self.db().read())
.await? as i64;
let pipeline_succeeded = pipeline_run::Entity::find()
.filter(pipeline_run::Column::State.eq(PipelineRunState::Success))
.count(self.db().read())
.await? as i64;
snapshots.push(QueueSnapshotDto {
name: QUEUE_PIPELINE.to_string(),
kind: "db".to_string(),
enabled: true,
depth: Some(pipeline_queued + pipeline_running),
queued: pipeline_queued,
running: pipeline_running,
failed: pipeline_failed,
succeeded: pipeline_succeeded,
note: Some("external runner driven".to_string()),
});
let deletion_pending = user_deletion_requests::Entity::find()
.filter(user_deletion_requests::Column::Status.eq(UserDeletionRequestStatus::Pending))
.count(self.db().read())
.await? as i64;
let deletion_scheduled = user_deletion_requests::Entity::find()
.filter(user_deletion_requests::Column::Status.eq(UserDeletionRequestStatus::Scheduled))
.count(self.db().read())
.await? as i64;
let deletion_completed = user_deletion_requests::Entity::find()
.filter(user_deletion_requests::Column::Status.eq(UserDeletionRequestStatus::Completed))
.count(self.db().read())
.await? as i64;
snapshots.push(QueueSnapshotDto {
name: QUEUE_DELETION.to_string(),
kind: "db".to_string(),
enabled: true,
depth: Some(deletion_pending + deletion_scheduled),
queued: deletion_scheduled,
running: deletion_pending,
failed: 0,
succeeded: deletion_completed,
note: None,
});
let task_pending = system_tasks::Entity::find()
.filter(system_tasks::Column::Status.eq(TaskStatus::Pending))
.count(self.db().read())
.await? as i64;
let task_running = system_tasks::Entity::find()
.filter(system_tasks::Column::Status.eq(TaskStatus::Running))
.count(self.db().read())
.await? as i64;
let task_failed = system_tasks::Entity::find()
.filter(system_tasks::Column::Status.eq(TaskStatus::Failed))
.count(self.db().read())
.await? as i64;
let task_succeeded = system_tasks::Entity::find()
.filter(system_tasks::Column::Status.eq(TaskStatus::Success))
.count(self.db().read())
.await? as i64;
snapshots.push(QueueSnapshotDto {
name: QUEUE_SYSTEM_TASKS.to_string(),
kind: "db".to_string(),
enabled: true,
depth: Some(task_pending + task_running),
queued: task_pending,
running: task_running,
failed: task_failed,
succeeded: task_succeeded,
note: Some("reserved scheduler table".to_string()),
});
Ok(snapshots)
}
pub async fn requeue_failed(
&self,
actor: &AdminActor,
queue: &str,
ctx: &AuditCtx,
) -> Result<QueueActionResp, AdminError> {
actor.require(AdminRole::Operator)?;
let now = time::OffsetDateTime::now_utc();
let affected = match queue {
QUEUE_SEARCH_INDEX => index_job::Entity::update_many()
.col_expr(index_job::Column::State, Expr::value("queued"))
.col_expr(index_job::Column::RunAfter, Expr::value(now))
.col_expr(
index_job::Column::LockedAt,
Expr::value(Option::<time::OffsetDateTime>::None),
)
.col_expr(index_job::Column::UpdatedAt, Expr::value(now))
.filter(index_job::Column::LastError.is_not_null())
.filter(index_job::Column::State.eq("queued"))
.exec(self.db().write())
.await?
.rows_affected,
QUEUE_IMPORT => job::Entity::update_many()
.col_expr(job::Column::State, Expr::value("queued"))
.col_expr(job::Column::RunAfter, Expr::value(now))
.col_expr(
job::Column::LockedAt,
Expr::value(Option::<time::OffsetDateTime>::None),
)
.col_expr(
job::Column::FinishedAt,
Expr::value(Option::<time::OffsetDateTime>::None),
)
.col_expr(job::Column::UpdatedAt, Expr::value(now))
.filter(job::Column::State.is_in(["failed", "canceled"]))
.exec(self.db().write())
.await?
.rows_affected,
QUEUE_CHAT_WEBHOOK => webhook_deliveries::Entity::update_many()
.col_expr(
webhook_deliveries::Column::Status,
Expr::value(WebhookDeliveryStatus::Pending),
)
.col_expr(webhook_deliveries::Column::NextRetryAt, Expr::value(now))
.col_expr(webhook_deliveries::Column::UpdatedAt, Expr::value(now))
.filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Failed))
.exec(self.db().write())
.await?
.rows_affected,
QUEUE_PIPELINE => pipeline_run::Entity::update_many()
.col_expr(
pipeline_run::Column::State,
Expr::value(PipelineRunState::Queued),
)
.col_expr(
pipeline_run::Column::StartedAt,
Expr::value(Option::<time::OffsetDateTime>::None),
)
.col_expr(
pipeline_run::Column::FinishedAt,
Expr::value(Option::<time::OffsetDateTime>::None),
)
.col_expr(pipeline_run::Column::UpdatedAt, Expr::value(now))
.filter(
pipeline_run::Column::State
.is_in([PipelineRunState::Failure, PipelineRunState::Cancelled]),
)
.exec(self.db().write())
.await?
.rows_affected,
QUEUE_DELETION => user_deletion_requests::Entity::update_many()
.col_expr(
user_deletion_requests::Column::Status,
Expr::value(UserDeletionRequestStatus::Scheduled),
)
.col_expr(
user_deletion_requests::Column::ScheduledFor,
Expr::value(Some(now)),
)
.col_expr(
user_deletion_requests::Column::CancelledAt,
Expr::value(Option::<time::OffsetDateTime>::None),
)
.col_expr(user_deletion_requests::Column::UpdatedAt, Expr::value(now))
.filter(
user_deletion_requests::Column::Status
.eq(UserDeletionRequestStatus::Cancelled),
)
.exec(self.db().write())
.await?
.rows_affected,
QUEUE_SYSTEM_TASKS => system_tasks::Entity::update_many()
.col_expr(system_tasks::Column::Status, Expr::value(TaskStatus::Pending))
.col_expr(system_tasks::Column::NextRunAt, Expr::value(Some(now)))
.col_expr(
system_tasks::Column::Error,
Expr::value(Option::<String>::None),
)
.col_expr(
system_tasks::Column::StartedAt,
Expr::value(Option::<time::OffsetDateTime>::None),
)
.col_expr(
system_tasks::Column::FinishedAt,
Expr::value(Option::<time::OffsetDateTime>::None),
)
.col_expr(system_tasks::Column::UpdatedAt, Expr::value(now))
.filter(system_tasks::Column::Status.eq(TaskStatus::Failed))
.exec(self.db().write())
.await?
.rows_affected,
_ => {
return Err(AdminError::BadRequest(format!(
"queue {queue} does not support requeue"
)));
}
};
self.audit(
Some(actor),
AuditRecord {
action: "queue.requeued",
target: Some(("queue", queue.to_string())),
detail: json!({ "affected": affected }),
},
ctx,
)
.await?;
Ok(QueueActionResp {
queue: queue.to_string(),
affected,
})
}
pub async fn purge_queue(
&self,
actor: &AdminActor,
queue: &str,
ctx: &AuditCtx,
) -> Result<QueueActionResp, AdminError> {
actor.require(AdminRole::SuperAdmin)?;
let affected = match queue {
QUEUE_GIT_SYNC => {
let topic = self.config().git.sync.queue_topic.clone();
let mut conn = self.redis().connection_clone();
let deleted: i64 = match &mut conn {
export::RedisConnection::Standalone(connection) => {
redis::cmd("DEL").arg(&topic).query_async(connection).await?
}
export::RedisConnection::Cluster(connection) => {
redis::cmd("DEL").arg(&topic).query_async(connection).await?
}
};
deleted.max(0) as u64
}
QUEUE_SEARCH_INDEX => index_job::Entity::delete_many()
.filter(index_job::Column::State.is_in(["succeeded", "cancelled"]))
.exec(self.db().write())
.await?
.rows_affected,
QUEUE_IMPORT => job::Entity::delete_many()
.filter(job::Column::State.is_in(["succeeded", "failed", "canceled"]))
.exec(self.db().write())
.await?
.rows_affected,
QUEUE_CHAT_WEBHOOK => webhook_deliveries::Entity::delete_many()
.filter(
webhook_deliveries::Column::Status
.is_in([WebhookDeliveryStatus::Done, WebhookDeliveryStatus::Failed]),
)
.exec(self.db().write())
.await?
.rows_affected,
QUEUE_PIPELINE => pipeline_run::Entity::delete_many()
.filter(
pipeline_run::Column::State.is_in([
PipelineRunState::Success,
PipelineRunState::Failure,
PipelineRunState::Cancelled,
]),
)
.exec(self.db().write())
.await?
.rows_affected,
QUEUE_DELETION => user_deletion_requests::Entity::delete_many()
.filter(user_deletion_requests::Column::Status.is_in([
UserDeletionRequestStatus::Completed,
UserDeletionRequestStatus::Cancelled,
]))
.exec(self.db().write())
.await?
.rows_affected,
QUEUE_SYSTEM_TASKS => system_tasks::Entity::delete_many()
.filter(
system_tasks::Column::Status
.is_in([TaskStatus::Success, TaskStatus::Failed]),
)
.exec(self.db().write())
.await?
.rows_affected,
_ => {
return Err(AdminError::BadRequest(format!(
"queue {queue} does not support purge"
)));
}
};
self.audit(
Some(actor),
AuditRecord {
action: "queue.purged",
target: Some(("queue", queue.to_string())),
detail: json!({ "affected": affected }),
},
ctx,
)
.await?;
Ok(QueueActionResp {
queue: queue.to_string(),
affected,
})
}
}