use std::collections::HashMap; use export::pagination::Page; use models::admin::admin_user::AdminRole; use models::chat::attachments::{self, AttachmentStatus}; use models::chat::message_edits; use models::chat::messages::{self, AuthorType, ChatMessageType}; use models::chat::outbound_webhooks; use models::chat::reactions; use models::chat::read_states; use models::chat::room_bookmarks; use models::chat::room_seq; use models::chat::room_user_settings; use models::chat::room_webhooks; use models::chat::threads; use models::chat::user_drafts; use models::chat::webhook_deliveries; use models::room::room_bans; use models::room::room_invites; use models::room::room_members::{self, RoomMemberRole}; use models::room::rooms::{self, RoomKind}; use models::space::spaces; use models::users::user; use sea_orm::ExprTrait; use sea_orm::sea_query::Expr; use sea_orm::{ ActiveModelTrait, ColumnTrait, Condition, ConnectionTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect, Set, TransactionTrait, }; use serde::{Deserialize, Serialize}; use serde_json::json; use time::OffsetDateTime; use utoipa::{IntoParams, ToSchema}; use uuid::Uuid; use crate::errors::AdminError; use crate::services::audit::{AuditCtx, AuditRecord}; use crate::services::service::AdminService; use crate::share::roles::AdminActor; pub const MAX_CHAT_PAGE: u64 = 200; pub const MAX_BULK_CHAT_MESSAGES: usize = 100; pub const MAX_CHAT_EXPORT_ROWS: u64 = 10000; #[derive(Debug, Clone, Deserialize, IntoParams)] pub struct ListChatRoomsQuery { pub q: Option, pub space_id: Option, pub kind: Option, pub status: Option, pub visibility: Option, pub limit: Option, pub offset: Option, } #[derive(Debug, Clone, Deserialize, IntoParams)] pub struct ListChatMessagesQuery { pub room_id: Option, pub author_id: Option, pub status: Option, pub q: Option, pub from: Option, pub to: Option, pub limit: Option, pub offset: Option, } #[derive(Debug, Clone, Deserialize, IntoParams)] pub struct ListChatAttachmentsQuery { pub room_id: Option, pub user_id: Option, pub status: Option, pub q: Option, pub limit: Option, pub offset: Option, } #[derive(Debug, Clone, Deserialize, IntoParams)] pub struct ListChatMembersQuery { pub status: Option, pub limit: Option, pub offset: Option, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatKindCount { pub kind: String, pub count: i64, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatStatsDto { pub rooms: i64, pub active_rooms: i64, pub private_rooms: i64, pub kinds: Vec, pub messages: i64, pub deleted_messages: i64, pub messages_24h: i64, pub threads: i64, pub attachments: i64, pub attachment_bytes: i64, pub pending_attachments: i64, pub members: i64, pub bans: i64, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatRoomSummary { pub id: Uuid, pub space_id: Option, pub space_name: Option, pub kind: String, pub name: String, pub lower_name: String, pub topic: Option, pub owner_id: Uuid, pub owner_username: Option, pub is_private: bool, pub nsfw: bool, pub is_active: bool, pub members: i64, pub messages: i64, pub last_message_at: Option, pub created_at: OffsetDateTime, pub updated_at: OffsetDateTime, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatRoomDetail { pub room: AdminChatRoomSummary, pub deleted_messages: i64, pub threads: i64, pub attachments: i64, pub attachment_bytes: i64, pub bans: i64, pub invites: i64, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct UpdateChatRoomCmd { #[serde(default)] pub name: Option, #[serde(default)] pub topic: Option, #[serde(default)] pub is_private: Option, #[serde(default)] pub nsfw: Option, #[serde(default)] pub is_active: Option, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatRoomMemberDto { pub user_id: Uuid, pub username: Option, pub display_name: Option, pub role: String, pub nick: Option, pub joined_at: OffsetDateTime, pub left_at: Option, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatMessageDto { pub id: Uuid, pub room_id: Uuid, pub room_name: Option, pub author_id: Uuid, pub author_username: Option, pub author_display_name: Option, pub author_type: String, pub message_type: String, pub content: String, pub seq: i64, pub flags: i64, pub pinned: bool, pub thread_id: Option, pub reply_to_id: Option, pub attachment_count: i64, pub reaction_count: i64, pub edited_at: Option, pub deleted_at: Option, pub created_at: OffsetDateTime, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatMessageEditDto { pub id: Uuid, pub editor_id: Uuid, pub editor_username: Option, pub content: String, pub created_at: OffsetDateTime, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatReactionGroupDto { pub emoji: String, pub count: i64, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatAttachmentDto { pub id: Uuid, pub user_id: Uuid, pub username: Option, pub display_name: Option, pub room_id: Option, pub message_id: Option, pub filename: String, pub description: Option, pub content_type: String, pub size: i64, pub bucket: String, pub s3_key: String, pub status: String, pub created_at: OffsetDateTime, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatThreadDto { pub id: Uuid, pub name: Option, pub owner_id: Uuid, pub reply_count: i64, pub participant_count: i64, pub archived: bool, pub locked: bool, pub created_at: OffsetDateTime, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct AdminChatMessageDetail { pub message: AdminChatMessageDto, pub edits: Vec, pub reactions: Vec, pub attachments: Vec, pub thread: Option, } #[derive(Debug, Clone, Deserialize, ToSchema)] pub struct BulkDeleteChatMessagesCmd { pub message_ids: Vec, #[serde(default)] pub reason: Option, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct BulkChatMessageFailure { pub id: Uuid, pub error: String, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct BulkDeleteChatMessagesResp { pub succeeded: Vec, pub failed: Vec, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct PurgeChatMessagesResp { pub room_id: Uuid, pub messages_deleted: u64, pub attachments_deleted: u64, pub threads_deleted: u64, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct DeleteChatRoomResp { pub room_id: Uuid, pub messages_deleted: u64, pub members_deleted: u64, pub attachments_deleted: u64, pub threads_deleted: u64, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct DeleteChatAttachmentResp { pub id: Uuid, pub s3_key: String, pub deleted: bool, pub storage_deleted: bool, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] pub struct RemoveChatMemberResp { pub room_id: Uuid, pub user_id: Uuid, pub removed: bool, } fn room_kind_key(kind: RoomKind) -> &'static str { match kind { RoomKind::GuildRoom => "guild_room", RoomKind::GroupDm => "group_dm", RoomKind::Dm => "dm", RoomKind::Category => "category", RoomKind::Text => "text", RoomKind::Voice => "voice", RoomKind::Stage => "stage", RoomKind::News => "news", RoomKind::Thread => "thread", RoomKind::DirectMessage => "direct_message", } } fn parse_room_kind(raw: &str) -> Option { match raw.trim().to_ascii_lowercase().as_str() { "guild_room" | "guild" => Some(RoomKind::GuildRoom), "group_dm" => Some(RoomKind::GroupDm), "dm" => Some(RoomKind::Dm), "category" => Some(RoomKind::Category), "text" => Some(RoomKind::Text), "voice" => Some(RoomKind::Voice), "stage" => Some(RoomKind::Stage), "news" => Some(RoomKind::News), "thread" => Some(RoomKind::Thread), "direct_message" => Some(RoomKind::DirectMessage), _ => None, } } fn author_type_key(value: AuthorType) -> &'static str { match value { AuthorType::User => "user", AuthorType::Bot => "bot", AuthorType::Webhook => "webhook", } } fn message_type_key(value: ChatMessageType) -> &'static str { match value { ChatMessageType::Default => "default", ChatMessageType::Reply => "reply", ChatMessageType::ThreadCreated => "thread_created", ChatMessageType::ChannelJoin => "channel_join", ChatMessageType::ChannelLeave => "channel_leave", ChatMessageType::ChannelTopic => "channel_topic", ChatMessageType::ChannelPinnedItem => "channel_pinned_item", ChatMessageType::Reminder => "reminder", ChatMessageType::System => "system", } } fn member_role_key(role: RoomMemberRole) -> &'static str { match role { RoomMemberRole::Member => "member", RoomMemberRole::Moderator => "moderator", RoomMemberRole::Admin => "admin", RoomMemberRole::Owner => "owner", } } fn attachment_status_key(status: AttachmentStatus) -> &'static str { match status { AttachmentStatus::Pending => "pending", AttachmentStatus::Uploaded => "uploaded", } } impl AdminService { async fn username_map( &self, ids: &[Uuid], ) -> Result, AdminError> { if ids.is_empty() { return Ok(HashMap::new()); } let mut unique = ids.to_vec(); unique.sort(); unique.dedup(); let rows = user::Entity::find() .filter(user::Column::Uid.is_in(unique)) .all(self.db().read()) .await?; Ok(rows .into_iter() .map(|row| (row.uid, (row.username, row.display_name))) .collect()) } async fn publish_room_event(&self, room_id: Uuid, event: serde_json::Value) { let channel = format!("chat:room:{room_id}"); let payload = match serde_json::to_string(&event) { Ok(payload) => payload, Err(_) => return, }; if let Err(error) = self.redis().publish(&channel, &payload).await { tracing::warn!(%error, %channel, "admin chat event publish failed"); } } async fn current_room_seq(&self, room_id: Uuid) -> i64 { room_seq::Entity::find_by_id(room_id) .one(self.db().read()) .await .ok() .flatten() .map(|row| row.seq) .unwrap_or(0) } pub async fn chat_stats(&self) -> Result { let rooms = rooms::Entity::find().count(self.db().read()).await? as i64; let active_rooms = rooms::Entity::find() .filter(rooms::Column::IsActive.eq(true)) .count(self.db().read()) .await? as i64; let private_rooms = rooms::Entity::find() .filter(rooms::Column::IsPrivate.eq(true)) .count(self.db().read()) .await? as i64; let kinds = rooms::Entity::find() .select_only() .column(rooms::Column::Kind) .column_as(Expr::col(rooms::Column::Id).count(), "cnt") .group_by(rooms::Column::Kind) .into_tuple::<(RoomKind, i64)>() .all(self.db().read()) .await? .into_iter() .map(|(kind, count)| AdminChatKindCount { kind: room_kind_key(kind).to_string(), count, }) .collect(); let messages = messages::Entity::find().count(self.db().read()).await? as i64; let deleted_messages = messages::Entity::find() .filter(messages::Column::DeletedAt.is_not_null()) .count(self.db().read()) .await? as i64; let messages_24h = messages::Entity::find() .filter( messages::Column::CreatedAt .gte(OffsetDateTime::now_utc() - time::Duration::hours(24)), ) .count(self.db().read()) .await? as i64; let threads = threads::Entity::find().count(self.db().read()).await? as i64; let attachments = attachments::Entity::find().count(self.db().read()).await? as i64; let pending_attachments = attachments::Entity::find() .filter(attachments::Column::Status.eq(AttachmentStatus::Pending)) .count(self.db().read()) .await? as i64; let attachment_bytes = attachments::Entity::find() .select_only() .column_as( Expr::col(attachments::Column::Size).sum().cast_as("bigint"), "total", ) .into_tuple::>() .one(self.db().read()) .await? .flatten() .unwrap_or(0); let members = room_members::Entity::find() .filter(room_members::Column::LeftAt.is_null()) .count(self.db().read()) .await? as i64; let bans = room_bans::Entity::find().count(self.db().read()).await? as i64; Ok(AdminChatStatsDto { rooms, active_rooms, private_rooms, kinds, messages, deleted_messages, messages_24h, threads, attachments, attachment_bytes, pending_attachments, members, bans, }) } async fn room_message_counts( &self, room_ids: &[Uuid], ) -> Result, AdminError> { if room_ids.is_empty() { return Ok(HashMap::new()); } let rows = messages::Entity::find() .select_only() .column(messages::Column::RoomId) .column_as(Expr::col(messages::Column::Id).count(), "cnt") .filter(messages::Column::RoomId.is_in(room_ids.to_vec())) .filter(messages::Column::DeletedAt.is_null()) .group_by(messages::Column::RoomId) .into_tuple::<(Uuid, i64)>() .all(self.db().read()) .await?; Ok(rows.into_iter().collect()) } async fn room_member_counts( &self, room_ids: &[Uuid], ) -> Result, AdminError> { if room_ids.is_empty() { return Ok(HashMap::new()); } let rows = room_members::Entity::find() .select_only() .column(room_members::Column::RoomId) .column_as(Expr::col(room_members::Column::UserId).count(), "cnt") .filter(room_members::Column::RoomId.is_in(room_ids.to_vec())) .filter(room_members::Column::LeftAt.is_null()) .group_by(room_members::Column::RoomId) .into_tuple::<(Uuid, i64)>() .all(self.db().read()) .await?; Ok(rows.into_iter().collect()) } async fn room_last_message_at( &self, room_ids: &[Uuid], ) -> Result, AdminError> { if room_ids.is_empty() { return Ok(HashMap::new()); } let rows = messages::Entity::find() .select_only() .column(messages::Column::RoomId) .column_as(Expr::col(messages::Column::CreatedAt).max(), "latest") .filter(messages::Column::RoomId.is_in(room_ids.to_vec())) .filter(messages::Column::DeletedAt.is_null()) .group_by(messages::Column::RoomId) .into_tuple::<(Uuid, Option)>() .all(self.db().read()) .await?; Ok(rows .into_iter() .filter_map(|(room_id, latest)| latest.map(|value| (room_id, value))) .collect()) } async fn room_select( &self, query: &ListChatRoomsQuery, ) -> Result, AdminError> { let mut select = rooms::Entity::find(); if let Some(space_id) = query.space_id { select = select.filter(rooms::Column::SpaceId.eq(space_id)); } if let Some(kind) = query .kind .as_deref() .map(str::trim) .filter(|value| !value.is_empty()) { match parse_room_kind(kind) { Some(parsed) => select = select.filter(rooms::Column::Kind.eq(parsed)), None => { return Err(AdminError::BadRequest(format!("unknown room kind {kind}"))); } } } select = match query.status.as_deref() { Some("active") => select.filter(rooms::Column::IsActive.eq(true)), Some("disabled") | Some("inactive") => select.filter(rooms::Column::IsActive.eq(false)), _ => select, }; select = match query.visibility.as_deref() { Some("private") => select.filter(rooms::Column::IsPrivate.eq(true)), Some("public") => select.filter(rooms::Column::IsPrivate.eq(false)), _ => select, }; if let Some(raw) = query .q .as_deref() .map(str::trim) .filter(|value| !value.is_empty()) { let needle = raw.to_ascii_lowercase(); select = select.filter( Condition::any() .add(rooms::Column::LowerName.contains(needle)) .add(rooms::Column::Name.contains(raw)), ); } Ok(select) } pub async fn list_chat_rooms( &self, query: ListChatRoomsQuery, ) -> Result, AdminError> { let limit = query.limit.unwrap_or(50).clamp(1, MAX_CHAT_PAGE); let offset = query.offset.unwrap_or(0); let select = self.room_select(&query).await?; let total = select.clone().count(self.db().read()).await? as i64; let rows = select .order_by_desc(rooms::Column::UpdatedAt) .limit(limit) .offset(offset) .all(self.db().read()) .await?; let items = self.chat_room_summaries(rows).await?; Ok(Page { items, total, limit, offset, }) } async fn chat_room_summaries( &self, rows: Vec, ) -> Result, AdminError> { if rows.is_empty() { return Ok(Vec::new()); } let room_ids: Vec = rows.iter().map(|row| row.id).collect(); let space_ids: Vec = rows.iter().filter_map(|row| row.space_id).collect(); let owner_ids: Vec = rows.iter().map(|row| row.owner_id).collect(); let (message_counts, member_counts, last_messages, owners, space_names) = tokio::join!( self.room_message_counts(&room_ids), self.room_member_counts(&room_ids), self.room_last_message_at(&room_ids), self.username_map(&owner_ids), self.space_name_map(&space_ids), ); let message_counts = message_counts?; let member_counts = member_counts?; let last_messages = last_messages?; let owners = owners?; let space_names = space_names?; Ok(rows .into_iter() .map(|row| AdminChatRoomSummary { id: row.id, space_id: row.space_id, space_name: row.space_id.and_then(|id| space_names.get(&id).cloned()), kind: room_kind_key(row.kind).to_string(), name: row.name.clone(), lower_name: row.lower_name.clone(), topic: row.topic.clone(), owner_id: row.owner_id, owner_username: owners.get(&row.owner_id).map(|(name, _)| name.clone()), is_private: row.is_private, nsfw: row.nsfw, is_active: row.is_active, members: member_counts.get(&row.id).copied().unwrap_or(0), messages: message_counts.get(&row.id).copied().unwrap_or(0), last_message_at: last_messages.get(&row.id).copied(), created_at: row.created_at, updated_at: row.updated_at, }) .collect()) } async fn space_name_map(&self, ids: &[Uuid]) -> Result, AdminError> { if ids.is_empty() { return Ok(HashMap::new()); } let mut unique = ids.to_vec(); unique.sort(); unique.dedup(); let rows = spaces::Entity::find() .filter(spaces::Column::Id.is_in(unique)) .all(self.db().read()) .await?; Ok(rows.into_iter().map(|row| (row.id, row.name)).collect()) } pub async fn get_chat_room(&self, room_id: Uuid) -> Result { let row = rooms::Entity::find_by_id(room_id) .one(self.db().read()) .await? .ok_or(AdminError::NotFound)?; let summary = self .chat_room_summaries(vec![row]) .await? .into_iter() .next() .ok_or(AdminError::NotFound)?; let deleted_messages = messages::Entity::find() .filter(messages::Column::RoomId.eq(room_id)) .filter(messages::Column::DeletedAt.is_not_null()) .count(self.db().read()) .await? as i64; let threads = threads::Entity::find() .filter(threads::Column::RoomId.eq(room_id)) .count(self.db().read()) .await? as i64; let attachments = attachments::Entity::find() .filter(attachments::Column::RoomId.eq(room_id)) .count(self.db().read()) .await? as i64; let attachment_bytes = attachments::Entity::find() .filter(attachments::Column::RoomId.eq(room_id)) .select_only() .column_as( Expr::col(attachments::Column::Size).sum().cast_as("bigint"), "total", ) .into_tuple::>() .one(self.db().read()) .await? .flatten() .unwrap_or(0); let bans = room_bans::Entity::find() .filter(room_bans::Column::RoomId.eq(room_id)) .count(self.db().read()) .await? as i64; let invites = room_invites::Entity::find() .filter(room_invites::Column::RoomId.eq(room_id)) .count(self.db().read()) .await? as i64; Ok(AdminChatRoomDetail { room: summary, deleted_messages, threads, attachments, attachment_bytes, bans, invites, }) } pub async fn update_chat_room( &self, actor: &AdminActor, room_id: Uuid, cmd: UpdateChatRoomCmd, ctx: &AuditCtx, ) -> Result { actor.require(AdminRole::Operator)?; let row = rooms::Entity::find_by_id(room_id) .one(self.db().read()) .await? .ok_or(AdminError::NotFound)?; let mut changed = serde_json::Map::new(); let room_space_id = row.space_id; let mut am: rooms::ActiveModel = row.into(); if let Some(name) = cmd.name { let name = name.trim().to_string(); if name.is_empty() || name.chars().count() > 100 { return Err(AdminError::BadRequest( "room name must contain 1..=100 characters".to_string(), )); } let lower = name.to_lowercase(); let mut conflict = rooms::Entity::find() .filter(rooms::Column::LowerName.eq(lower.clone())) .filter(rooms::Column::Id.ne(room_id)); conflict = match room_space_id { Some(space_id) => conflict.filter(rooms::Column::SpaceId.eq(space_id)), None => conflict.filter(rooms::Column::SpaceId.is_null()), }; if conflict.count(self.db().read()).await? > 0 { return Err(AdminError::Conflict( "another room in this space already uses that name".to_string(), )); } changed.insert("name".to_string(), json!(name)); am.name = Set(name.clone()); am.lower_name = Set(lower); } if let Some(topic) = cmd.topic { changed.insert("topic".to_string(), json!(topic)); am.topic = Set(Some(topic)); } if let Some(value) = cmd.is_private { changed.insert("is_private".to_string(), json!(value)); am.is_private = Set(value); } if let Some(value) = cmd.nsfw { changed.insert("nsfw".to_string(), json!(value)); am.nsfw = Set(value); } if let Some(value) = cmd.is_active { changed.insert("is_active".to_string(), json!(value)); am.is_active = Set(value); } if changed.is_empty() { return self.get_chat_room(room_id).await; } am.updated_at = Set(OffsetDateTime::now_utc()); am.update(self.db().write()).await?; self.audit( Some(actor), AuditRecord { action: "chat.room_updated", target: Some(("chat_room", room_id.to_string())), detail: serde_json::Value::Object(changed), }, ctx, ) .await?; self.publish_room_event(room_id, json!({"kind": "room_updated", "room_id": room_id})) .await; self.get_chat_room(room_id).await } pub async fn list_chat_room_members( &self, room_id: Uuid, query: ListChatMembersQuery, ) -> Result, AdminError> { rooms::Entity::find_by_id(room_id) .one(self.db().read()) .await? .ok_or(AdminError::NotFound)?; let limit = query.limit.unwrap_or(50).clamp(1, MAX_CHAT_PAGE); let offset = query.offset.unwrap_or(0); let mut select = room_members::Entity::find().filter(room_members::Column::RoomId.eq(room_id)); select = match query.status.as_deref() { Some("active") => select.filter(room_members::Column::LeftAt.is_null()), Some("left") => select.filter(room_members::Column::LeftAt.is_not_null()), _ => select, }; let total = select.clone().count(self.db().read()).await? as i64; let rows = select .order_by_asc(room_members::Column::JoinedAt) .limit(limit) .offset(offset) .all(self.db().read()) .await?; let user_ids: Vec = rows.iter().map(|row| row.user_id).collect(); let users = self.username_map(&user_ids).await?; let items = rows .into_iter() .map(|row| AdminChatRoomMemberDto { user_id: row.user_id, username: users.get(&row.user_id).map(|(name, _)| name.clone()), display_name: users.get(&row.user_id).map(|(_, display)| display.clone()), role: member_role_key(row.role).to_string(), nick: row.nick, joined_at: row.joined_at, left_at: row.left_at, }) .collect(); Ok(Page { items, total, limit, offset, }) } pub async fn remove_chat_room_member( &self, actor: &AdminActor, room_id: Uuid, user_id: Uuid, ctx: &AuditCtx, ) -> Result { actor.require(AdminRole::Operator)?; let row = room_members::Entity::find_by_id((room_id, user_id)) .one(self.db().read()) .await? .ok_or(AdminError::NotFound)?; if row.left_at.is_some() { return Ok(RemoveChatMemberResp { room_id, user_id, removed: false, }); } let now = OffsetDateTime::now_utc(); let mut am: room_members::ActiveModel = row.into(); am.left_at = Set(Some(now)); am.updated_at = Set(Some(now)); am.update(self.db().write()).await?; self.audit( Some(actor), AuditRecord { action: "chat.member_removed", target: Some(("chat_room", room_id.to_string())), detail: json!({ "user_id": user_id }), }, ctx, ) .await?; self.publish_room_event( room_id, json!({"kind": "member_left", "room_id": room_id, "user_id": user_id}), ) .await; Ok(RemoveChatMemberResp { room_id, user_id, removed: true, }) } async fn chat_message_select( &self, query: &ListChatMessagesQuery, ) -> Result, AdminError> { let mut select = messages::Entity::find(); if let Some(room_id) = query.room_id { select = select.filter(messages::Column::RoomId.eq(room_id)); } if let Some(author_id) = query.author_id { select = select.filter(messages::Column::AuthorId.eq(author_id)); } select = match query.status.as_deref() { Some("deleted") => select.filter(messages::Column::DeletedAt.is_not_null()), Some("active") => select.filter(messages::Column::DeletedAt.is_null()), Some("pinned") => select .filter(messages::Column::DeletedAt.is_null()) .filter(messages::Column::Pinned.eq(true)), _ => select, }; if let Some(raw) = query .q .as_deref() .map(str::trim) .filter(|value| !value.is_empty()) { select = select.filter(messages::Column::Content.contains(raw)); } if let Some(from) = query.from { select = select.filter(messages::Column::CreatedAt.gte(from)); } if let Some(to) = query.to { select = select.filter(messages::Column::CreatedAt.lte(to)); } Ok(select) } async fn message_author_map( &self, author_ids: &[Uuid], ) -> Result, AdminError> { self.username_map(author_ids).await } fn message_dto( row: &messages::Model, room_names: &HashMap, authors: &HashMap, attachment_counts: &HashMap, reaction_counts: &HashMap, ) -> AdminChatMessageDto { let author = authors.get(&row.author_id); AdminChatMessageDto { id: row.id, room_id: row.room_id, room_name: room_names.get(&row.room_id).cloned(), author_id: row.author_id, author_username: author.map(|(name, _)| name.clone()), author_display_name: author.map(|(_, display)| display.clone()), author_type: author_type_key(row.author_type).to_string(), message_type: message_type_key(row.message_type).to_string(), content: row.content.clone(), seq: row.seq, flags: row.flags, pinned: row.pinned, thread_id: row.thread_id, reply_to_id: row.reply_to_id, attachment_count: attachment_counts.get(&row.id).copied().unwrap_or(0), reaction_count: reaction_counts.get(&row.id).copied().unwrap_or(0), edited_at: row.edited_at, deleted_at: row.deleted_at, created_at: row.created_at, } } async fn message_counter_maps( &self, message_ids: &[Uuid], ) -> Result<(HashMap, HashMap), AdminError> { if message_ids.is_empty() { return Ok((HashMap::new(), HashMap::new())); } let attachments = attachments::Entity::find() .select_only() .column(attachments::Column::MessageId) .column_as(Expr::col(attachments::Column::Id).count(), "cnt") .filter(attachments::Column::MessageId.is_in(message_ids.to_vec())) .group_by(attachments::Column::MessageId) .into_tuple::<(Option, i64)>() .all(self.db().read()) .await? .into_iter() .filter_map(|(id, count)| id.map(|id| (id, count))) .collect(); let reactions = reactions::Entity::find() .select_only() .column(reactions::Column::MessageId) .column_as(Expr::col(reactions::Column::UserId).count(), "cnt") .filter(reactions::Column::MessageId.is_in(message_ids.to_vec())) .group_by(reactions::Column::MessageId) .into_tuple::<(Uuid, i64)>() .all(self.db().read()) .await? .into_iter() .collect(); Ok((attachments, reactions)) } pub async fn list_chat_messages( &self, query: ListChatMessagesQuery, ) -> Result, AdminError> { self.chat_messages_page(query, MAX_CHAT_PAGE).await } async fn chat_messages_page( &self, query: ListChatMessagesQuery, max_limit: u64, ) -> Result, AdminError> { let limit = query.limit.unwrap_or(50).clamp(1, max_limit); let offset = query.offset.unwrap_or(0); let select = self.chat_message_select(&query).await?; let total = select.clone().count(self.db().read()).await? as i64; let rows = select .order_by_desc(messages::Column::CreatedAt) .limit(limit) .offset(offset) .all(self.db().read()) .await?; let items = self.chat_message_dtos(rows).await?; Ok(Page { items, total, limit, offset, }) } async fn chat_message_dtos( &self, rows: Vec, ) -> Result, AdminError> { if rows.is_empty() { return Ok(Vec::new()); } let message_ids: Vec = rows.iter().map(|row| row.id).collect(); let room_ids: Vec = rows.iter().map(|row| row.room_id).collect(); let author_ids: Vec = rows.iter().map(|row| row.author_id).collect(); let (authors, room_names, counters) = tokio::join!( self.message_author_map(&author_ids), self.room_name_map(&room_ids), self.message_counter_maps(&message_ids), ); let authors = authors?; let room_names = room_names?; let (attachment_counts, reaction_counts) = counters?; Ok(rows .iter() .map(|row| { Self::message_dto( row, &room_names, &authors, &attachment_counts, &reaction_counts, ) }) .collect()) } async fn room_name_map(&self, ids: &[Uuid]) -> Result, AdminError> { if ids.is_empty() { return Ok(HashMap::new()); } let mut unique = ids.to_vec(); unique.sort(); unique.dedup(); let rows = rooms::Entity::find() .filter(rooms::Column::Id.is_in(unique)) .all(self.db().read()) .await?; Ok(rows.into_iter().map(|row| (row.id, row.name)).collect()) } pub async fn get_chat_message( &self, message_id: Uuid, ) -> Result { let row = messages::Entity::find_by_id(message_id) .one(self.db().read()) .await? .ok_or(AdminError::NotFound)?; let room_id = row.room_id; let message = self .chat_message_dtos(vec![row]) .await? .into_iter() .next() .ok_or(AdminError::NotFound)?; let edits = message_edits::Entity::find() .filter(message_edits::Column::MessageId.eq(message_id)) .order_by_desc(message_edits::Column::CreatedAt) .all(self.db().read()) .await?; let editor_ids: Vec = edits.iter().map(|row| row.editor_id).collect(); let editors = self.username_map(&editor_ids).await?; let edits = edits .into_iter() .map(|row| AdminChatMessageEditDto { id: row.id, editor_id: row.editor_id, editor_username: editors.get(&row.editor_id).map(|(name, _)| name.clone()), content: row.content, created_at: row.created_at, }) .collect(); let reaction_rows = reactions::Entity::find() .filter(reactions::Column::MessageId.eq(message_id)) .select_only() .column(reactions::Column::Emoji) .column_as(Expr::col(reactions::Column::UserId).count(), "cnt") .group_by(reactions::Column::Emoji) .into_tuple::<(String, i64)>() .all(self.db().read()) .await?; let reactions = reaction_rows .into_iter() .map(|(emoji, count)| AdminChatReactionGroupDto { emoji, count }) .collect(); let attachment_rows = attachments::Entity::find() .filter(attachments::Column::MessageId.eq(message_id)) .order_by_asc(attachments::Column::CreatedAt) .all(self.db().read()) .await?; let attachment_user_ids: Vec = attachment_rows.iter().map(|row| row.user_id).collect(); let attachment_users = self.username_map(&attachment_user_ids).await?; let attachments = attachment_rows .into_iter() .map(|row| AdminChatAttachmentDto { id: row.id, user_id: row.user_id, username: attachment_users .get(&row.user_id) .map(|(name, _)| name.clone()), display_name: attachment_users .get(&row.user_id) .map(|(_, display)| display.clone()), room_id: row.room_id, message_id: row.message_id, filename: row.filename, description: row.description, content_type: row.content_type, size: row.size, bucket: row.bucket, s3_key: row.s3_key, status: attachment_status_key(row.status).to_string(), created_at: row.created_at, }) .collect(); let thread = threads::Entity::find() .filter(threads::Column::RootMessageId.eq(message_id)) .one(self.db().read()) .await? .map(|row| AdminChatThreadDto { id: row.id, name: row.name, owner_id: row.owner_id, reply_count: row.reply_count, participant_count: row.participant_count, archived: row.archived, locked: row.locked, created_at: row.created_at, }); let _ = room_id; Ok(AdminChatMessageDetail { message, edits, reactions, attachments, thread, }) } async fn delete_message_dependents( txn: &impl ConnectionTrait, message_ids: &[Uuid], ) -> Result<(), AdminError> { if message_ids.is_empty() { return Ok(()); } message_edits::Entity::delete_many() .filter(message_edits::Column::MessageId.is_in(message_ids.to_vec())) .exec(txn) .await?; reactions::Entity::delete_many() .filter(reactions::Column::MessageId.is_in(message_ids.to_vec())) .exec(txn) .await?; models::users::user_saved_messages::Entity::delete_many() .filter( models::users::user_saved_messages::Column::MessageId.is_in(message_ids.to_vec()), ) .exec(txn) .await?; models::users::user_recent_mentions::Entity::delete_many() .filter( models::users::user_recent_mentions::Column::MessageId.is_in(message_ids.to_vec()), ) .exec(txn) .await?; models::chat::message_reminders::Entity::delete_many() .filter(models::chat::message_reminders::Column::MessageId.is_in(message_ids.to_vec())) .exec(txn) .await?; Ok(()) } pub async fn bulk_delete_chat_messages( &self, actor: &AdminActor, cmd: BulkDeleteChatMessagesCmd, ctx: &AuditCtx, ) -> Result { actor.require(AdminRole::Operator)?; let mut ids = cmd.message_ids.clone(); ids.sort(); ids.dedup(); if ids.is_empty() || ids.len() > MAX_BULK_CHAT_MESSAGES { return Err(AdminError::BadRequest(format!( "message_ids must contain 1..={MAX_BULK_CHAT_MESSAGES} entries" ))); } let rows = messages::Entity::find() .filter(messages::Column::Id.is_in(ids.clone())) .all(self.db().read()) .await?; let now = OffsetDateTime::now_utc(); let mut succeeded = Vec::new(); let mut failed = Vec::new(); let mut affected_rooms: Vec = Vec::new(); for message_id in ids { let Some(row) = rows.iter().find(|row| row.id == message_id) else { failed.push(BulkChatMessageFailure { id: message_id, error: "message not found".to_string(), }); continue; }; if row.deleted_at.is_some() { failed.push(BulkChatMessageFailure { id: message_id, error: "already deleted".to_string(), }); continue; } let room_id = row.room_id; let result = self .db() .primary() .transaction::<_, (), AdminError>(move |txn| { Box::pin(async move { let ids = vec![message_id]; messages::Entity::update_many() .filter(messages::Column::Id.eq(message_id)) .col_expr(messages::Column::Content, Expr::value(String::new())) .col_expr(messages::Column::DeletedAt, Expr::value(now)) .exec(txn) .await?; Self::delete_message_dependents(txn, &ids).await?; Ok(()) }) }) .await; match result { Ok(()) => { succeeded.push(message_id); if !affected_rooms.contains(&room_id) { affected_rooms.push(room_id); } } Err(error) => failed.push(BulkChatMessageFailure { id: message_id, error: error.to_string(), }), } } for room_id in &affected_rooms { let seq = self.current_room_seq(*room_id).await; self.publish_room_event( *room_id, json!({ "kind": "messages_deleted", "room_id": room_id, "seq": seq, "message_ids": succeeded, }), ) .await; } self.audit( Some(actor), AuditRecord { action: "chat.messages_deleted", target: Some(("chat_message", format!("{} items", succeeded.len()))), detail: json!({ "succeeded": succeeded, "failed": failed.iter().map(|item| item.id.to_string()).collect::>(), "reason": cmd.reason, }), }, ctx, ) .await?; Ok(BulkDeleteChatMessagesResp { succeeded, failed }) } pub async fn purge_chat_room_messages( &self, actor: &AdminActor, room_id: Uuid, ctx: &AuditCtx, ) -> Result { actor.require(AdminRole::SuperAdmin)?; rooms::Entity::find_by_id(room_id) .one(self.db().read()) .await? .ok_or(AdminError::NotFound)?; let message_ids: Vec = messages::Entity::find() .select_only() .column(messages::Column::Id) .filter(messages::Column::RoomId.eq(room_id)) .into_tuple::() .all(self.db().read()) .await?; let attachment_keys: Vec = attachments::Entity::find() .select_only() .column(attachments::Column::S3Key) .filter(attachments::Column::RoomId.eq(room_id)) .into_tuple::() .all(self.db().read()) .await?; let thread_ids: Vec = threads::Entity::find() .select_only() .column(threads::Column::Id) .filter(threads::Column::RoomId.eq(room_id)) .into_tuple::() .all(self.db().read()) .await?; let messages_deleted = message_ids.len() as u64; let attachments_deleted = attachment_keys.len() as u64; let threads_deleted = thread_ids.len() as u64; self.db() .primary() .transaction::<_, (), AdminError>(move |txn| { Box::pin(async move { if !message_ids.is_empty() { Self::delete_message_dependents(txn, &message_ids).await?; } attachments::Entity::delete_many() .filter(attachments::Column::RoomId.eq(room_id)) .exec(txn) .await?; if !thread_ids.is_empty() { models::chat::thread_members::Entity::delete_many() .filter( models::chat::thread_members::Column::ThreadId .is_in(thread_ids.clone()), ) .exec(txn) .await?; threads::Entity::delete_many() .filter(threads::Column::Id.is_in(thread_ids)) .exec(txn) .await?; } messages::Entity::delete_many() .filter(messages::Column::RoomId.eq(room_id)) .exec(txn) .await?; models::users::user_saved_messages::Entity::delete_many() .filter(models::users::user_saved_messages::Column::ChannelId.eq(room_id)) .exec(txn) .await?; models::users::user_recent_mentions::Entity::delete_many() .filter(models::users::user_recent_mentions::Column::ChannelId.eq(room_id)) .exec(txn) .await?; Ok(()) }) }) .await?; for key in &attachment_keys { if let Err(error) = self.storage().client().delete(key).await { tracing::warn!(%error, %key, "admin chat purge storage delete failed"); } } rooms::Entity::update_many() .filter(rooms::Column::Id.eq(room_id)) .col_expr( rooms::Column::LastMessageId, Expr::value(Option::::None), ) .col_expr( rooms::Column::UpdatedAt, Expr::value(OffsetDateTime::now_utc()), ) .exec(self.db().write()) .await?; self.audit( Some(actor), AuditRecord { action: "chat.messages_purged", target: Some(("chat_room", room_id.to_string())), detail: json!({ "messages_deleted": messages_deleted, "attachments_deleted": attachments_deleted, "threads_deleted": threads_deleted, }), }, ctx, ) .await?; Ok(PurgeChatMessagesResp { room_id, messages_deleted, attachments_deleted, threads_deleted, }) } pub async fn delete_chat_room( &self, actor: &AdminActor, room_id: Uuid, ctx: &AuditCtx, ) -> Result { actor.require(AdminRole::SuperAdmin)?; let room = rooms::Entity::find_by_id(room_id) .one(self.db().read()) .await? .ok_or(AdminError::NotFound)?; let child_rooms = rooms::Entity::find() .filter(rooms::Column::ParentId.eq(room_id)) .count(self.db().read()) .await?; if child_rooms > 0 { return Err(AdminError::Conflict( "room still has child rooms; delete or move them first".to_string(), )); } let message_ids: Vec = messages::Entity::find() .select_only() .column(messages::Column::Id) .filter(messages::Column::RoomId.eq(room_id)) .into_tuple::() .all(self.db().read()) .await?; let attachment_keys: Vec = attachments::Entity::find() .select_only() .column(attachments::Column::S3Key) .filter(attachments::Column::RoomId.eq(room_id)) .into_tuple::() .all(self.db().read()) .await?; let thread_ids: Vec = threads::Entity::find() .select_only() .column(threads::Column::Id) .filter(threads::Column::RoomId.eq(room_id)) .into_tuple::() .all(self.db().read()) .await?; let members_deleted = room_members::Entity::find() .filter(room_members::Column::RoomId.eq(room_id)) .count(self.db().read()) .await? as u64; let messages_deleted = message_ids.len() as u64; let attachments_deleted = attachment_keys.len() as u64; let threads_deleted = thread_ids.len() as u64; let room_name = room.name.clone(); self.db() .primary() .transaction::<_, (), AdminError>(move |txn| { Box::pin(async move { if !message_ids.is_empty() { Self::delete_message_dependents(txn, &message_ids).await?; } attachments::Entity::delete_many() .filter(attachments::Column::RoomId.eq(room_id)) .exec(txn) .await?; if !thread_ids.is_empty() { models::chat::thread_members::Entity::delete_many() .filter( models::chat::thread_members::Column::ThreadId .is_in(thread_ids.clone()), ) .exec(txn) .await?; threads::Entity::delete_many() .filter(threads::Column::Id.is_in(thread_ids)) .exec(txn) .await?; } messages::Entity::delete_many() .filter(messages::Column::RoomId.eq(room_id)) .exec(txn) .await?; read_states::Entity::delete_many() .filter(read_states::Column::RoomId.eq(room_id)) .exec(txn) .await?; room_user_settings::Entity::delete_many() .filter(room_user_settings::Column::RoomId.eq(room_id)) .exec(txn) .await?; room_bookmarks::Entity::delete_many() .filter(room_bookmarks::Column::RoomId.eq(room_id)) .exec(txn) .await?; user_drafts::Entity::delete_many() .filter(user_drafts::Column::RoomId.eq(room_id)) .exec(txn) .await?; room_bans::Entity::delete_many() .filter(room_bans::Column::RoomId.eq(room_id)) .exec(txn) .await?; room_invites::Entity::delete_many() .filter(room_invites::Column::RoomId.eq(room_id)) .exec(txn) .await?; room_webhooks::Entity::delete_many() .filter(room_webhooks::Column::RoomId.eq(room_id)) .exec(txn) .await?; outbound_webhooks::Entity::delete_many() .filter(outbound_webhooks::Column::RoomId.eq(room_id)) .exec(txn) .await?; webhook_deliveries::Entity::delete_many() .filter(webhook_deliveries::Column::RoomId.eq(room_id)) .exec(txn) .await?; room_seq::Entity::delete_by_id(room_id).exec(txn).await?; models::users::user_saved_messages::Entity::delete_many() .filter(models::users::user_saved_messages::Column::ChannelId.eq(room_id)) .exec(txn) .await?; models::users::user_recent_mentions::Entity::delete_many() .filter(models::users::user_recent_mentions::Column::ChannelId.eq(room_id)) .exec(txn) .await?; room_members::Entity::delete_many() .filter(room_members::Column::RoomId.eq(room_id)) .exec(txn) .await?; rooms::Entity::delete_by_id(room_id).exec(txn).await?; Ok(()) }) }) .await?; for key in &attachment_keys { if let Err(error) = self.storage().client().delete(key).await { tracing::warn!(%error, %key, "admin chat room storage delete failed"); } } self.audit( Some(actor), AuditRecord { action: "chat.room_deleted", target: Some(("chat_room", room_id.to_string())), detail: json!({ "name": room_name, "messages_deleted": messages_deleted, "members_deleted": members_deleted, "attachments_deleted": attachments_deleted, "threads_deleted": threads_deleted, }), }, ctx, ) .await?; Ok(DeleteChatRoomResp { room_id, messages_deleted, members_deleted, attachments_deleted, threads_deleted, }) } pub async fn list_chat_attachments( &self, query: ListChatAttachmentsQuery, ) -> Result, AdminError> { let limit = query.limit.unwrap_or(50).clamp(1, MAX_CHAT_PAGE); let offset = query.offset.unwrap_or(0); let mut select = attachments::Entity::find(); if let Some(room_id) = query.room_id { select = select.filter(attachments::Column::RoomId.eq(room_id)); } if let Some(user_id) = query.user_id { select = select.filter(attachments::Column::UserId.eq(user_id)); } select = match query.status.as_deref() { Some("pending") => { select.filter(attachments::Column::Status.eq(AttachmentStatus::Pending)) } Some("uploaded") => { select.filter(attachments::Column::Status.eq(AttachmentStatus::Uploaded)) } _ => select, }; if let Some(raw) = query .q .as_deref() .map(str::trim) .filter(|value| !value.is_empty()) { select = select.filter( Condition::any() .add(attachments::Column::Filename.contains(raw)) .add(attachments::Column::S3Key.contains(raw)), ); } let total = select.clone().count(self.db().read()).await? as i64; let rows = select .order_by_desc(attachments::Column::CreatedAt) .limit(limit) .offset(offset) .all(self.db().read()) .await?; let user_ids: Vec = rows.iter().map(|row| row.user_id).collect(); let users = self.username_map(&user_ids).await?; let items = rows .into_iter() .map(|row| AdminChatAttachmentDto { id: row.id, user_id: row.user_id, username: users.get(&row.user_id).map(|(name, _)| name.clone()), display_name: users.get(&row.user_id).map(|(_, display)| display.clone()), room_id: row.room_id, message_id: row.message_id, filename: row.filename, description: row.description, content_type: row.content_type, size: row.size, bucket: row.bucket, s3_key: row.s3_key, status: attachment_status_key(row.status).to_string(), created_at: row.created_at, }) .collect(); Ok(Page { items, total, limit, offset, }) } pub async fn delete_chat_attachment( &self, actor: &AdminActor, attachment_id: Uuid, ctx: &AuditCtx, ) -> Result { actor.require(AdminRole::SuperAdmin)?; let row = attachments::Entity::find_by_id(attachment_id) .one(self.db().read()) .await? .ok_or(AdminError::NotFound)?; let shared = attachments::Entity::find() .filter(attachments::Column::S3Key.eq(row.s3_key.clone())) .filter(attachments::Column::Id.ne(row.id)) .count(self.db().read()) .await?; attachments::Entity::delete_by_id(attachment_id) .exec(self.db().write()) .await?; let mut storage_deleted = false; if shared == 0 { match self.storage().client().delete(&row.s3_key).await { Ok(()) => storage_deleted = true, Err(error) => { tracing::warn!(%error, key = %row.s3_key, "admin chat attachment storage delete failed"); } } // The preview is derived from this object; removing one without the // other would leave an unreachable file in the bucket forever. let key_base = export::imaging::preview_key_base_of(&row.s3_key); for candidate in imaging::layout::preview_candidate_keys(&key_base) { if let Err(error) = self.storage().client().delete(&candidate).await { tracing::warn!(%error, key = %candidate, "admin chat attachment preview delete failed"); } } } self.audit( Some(actor), AuditRecord { action: "chat.attachment_deleted", target: Some(("chat_attachment", attachment_id.to_string())), detail: json!({ "s3_key": row.s3_key, "storage_deleted": storage_deleted, "filename": row.filename, }), }, ctx, ) .await?; Ok(DeleteChatAttachmentResp { id: attachment_id, s3_key: row.s3_key, deleted: true, storage_deleted, }) } pub async fn export_chat_messages( &self, query: ListChatMessagesQuery, ) -> Result, AdminError> { let mut query = query; query.limit = Some(MAX_CHAT_EXPORT_ROWS); query.offset = Some(0); Ok(self .chat_messages_page(query, MAX_CHAT_EXPORT_ROWS) .await? .items) } }