eba4fd1
use std::collections::HashMap;
use export::pagination::Page;
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::service::AdminService;
use crate::share::roles::{AdminActor, AdminRole};
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<String>,
pub space_id: Option<Uuid>,
pub kind: Option<String>,
pub status: Option<String>,
pub visibility: Option<String>,
pub limit: Option<u64>,
pub offset: Option<u64>,
}
#[derive(Debug, Clone, Deserialize, IntoParams)]
pub struct ListChatMessagesQuery {
pub room_id: Option<Uuid>,
pub author_id: Option<Uuid>,
pub status: Option<String>,
pub q: Option<String>,
pub from: Option<OffsetDateTime>,
pub to: Option<OffsetDateTime>,
pub limit: Option<u64>,
pub offset: Option<u64>,
}
#[derive(Debug, Clone, Deserialize, IntoParams)]
pub struct ListChatAttachmentsQuery {
pub room_id: Option<Uuid>,
pub user_id: Option<Uuid>,
pub status: Option<String>,
pub q: Option<String>,
pub limit: Option<u64>,
pub offset: Option<u64>,
}
#[derive(Debug, Clone, Deserialize, IntoParams)]
pub struct ListChatMembersQuery {
pub status: Option<String>,
pub limit: Option<u64>,
pub offset: Option<u64>,
}
#[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<AdminChatKindCount>,
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<Uuid>,
pub space_name: Option<String>,
pub kind: String,
pub name: String,
pub lower_name: String,
pub topic: Option<String>,
pub owner_id: Uuid,
pub owner_username: Option<String>,
pub is_private: bool,
pub nsfw: bool,
pub is_active: bool,
pub members: i64,
pub messages: i64,
pub last_message_at: Option<OffsetDateTime>,
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<String>,
#[serde(default)]
pub topic: Option<String>,
#[serde(default)]
pub is_private: Option<bool>,
#[serde(default)]
pub nsfw: Option<bool>,
#[serde(default)]
pub is_active: Option<bool>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminChatRoomMemberDto {
pub user_id: Uuid,
pub username: Option<String>,
pub display_name: Option<String>,
pub role: String,
pub nick: Option<String>,
pub joined_at: OffsetDateTime,
pub left_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminChatMessageDto {
pub id: Uuid,
pub room_id: Uuid,
pub room_name: Option<String>,
pub author_id: Uuid,
pub author_username: Option<String>,
pub author_display_name: Option<String>,
pub author_type: String,
pub message_type: String,
pub content: String,
pub seq: i64,
pub flags: i64,
pub pinned: bool,
pub thread_id: Option<Uuid>,
pub reply_to_id: Option<Uuid>,
pub attachment_count: i64,
pub reaction_count: i64,
pub edited_at: Option<OffsetDateTime>,
pub deleted_at: Option<OffsetDateTime>,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AdminChatMessageEditDto {
pub id: Uuid,
pub editor_id: Uuid,
pub editor_username: Option<String>,
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<String>,
pub display_name: Option<String>,
pub room_id: Option<Uuid>,
pub message_id: Option<Uuid>,
pub filename: String,
pub description: Option<String>,
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<String>,
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<AdminChatMessageEditDto>,
pub reactions: Vec<AdminChatReactionGroupDto>,
pub attachments: Vec<AdminChatAttachmentDto>,
pub thread: Option<AdminChatThreadDto>,
}
#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct BulkDeleteChatMessagesCmd {
pub message_ids: Vec<Uuid>,
#[serde(default)]
pub reason: Option<String>,
}
#[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<Uuid>,
pub failed: Vec<BulkChatMessageFailure>,
}
#[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<RoomKind> {
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<HashMap<Uuid, (String, String)>, 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<AdminChatStatsDto, AdminError> {
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::<Option<i64>>()
.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<HashMap<Uuid, i64>, 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<HashMap<Uuid, i64>, 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<HashMap<Uuid, OffsetDateTime>, 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<OffsetDateTime>)>()
.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<sea_orm::Select<rooms::Entity>, 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<Page<AdminChatRoomSummary>, 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<rooms::Model>,
) -> Result<Vec<AdminChatRoomSummary>, AdminError> {
if rows.is_empty() {
return Ok(Vec::new());
}
let room_ids: Vec<Uuid> = rows.iter().map(|row| row.id).collect();
let space_ids: Vec<Uuid> = rows.iter().filter_map(|row| row.space_id).collect();
let owner_ids: Vec<Uuid> = 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<HashMap<Uuid, String>, 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<AdminChatRoomDetail, AdminError> {
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::<Option<i64>>()
.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,
) -> Result<AdminChatRoomDetail, AdminError> {
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.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<Page<AdminChatRoomMemberDto>, 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<Uuid> = 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,
) -> Result<RemoveChatMemberResp, AdminError> {
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.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<sea_orm::Select<messages::Entity>, 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<HashMap<Uuid, (String, String)>, AdminError> {
self.username_map(author_ids).await
}
fn message_dto(
row: &messages::Model,
room_names: &HashMap<Uuid, String>,
authors: &HashMap<Uuid, (String, String)>,
attachment_counts: &HashMap<Uuid, i64>,
reaction_counts: &HashMap<Uuid, i64>,
) -> 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<Uuid, i64>, HashMap<Uuid, i64>), 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<Uuid>, 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<Page<AdminChatMessageDto>, AdminError> {
self.chat_messages_page(query, MAX_CHAT_PAGE).await
}
async fn chat_messages_page(
&self,
query: ListChatMessagesQuery,
max_limit: u64,
) -> Result<Page<AdminChatMessageDto>, 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<messages::Model>,
) -> Result<Vec<AdminChatMessageDto>, AdminError> {
if rows.is_empty() {
return Ok(Vec::new());
}
let message_ids: Vec<Uuid> = rows.iter().map(|row| row.id).collect();
let room_ids: Vec<Uuid> = rows.iter().map(|row| row.room_id).collect();
let author_ids: Vec<Uuid> = 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<HashMap<Uuid, String>, 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<AdminChatMessageDetail, AdminError> {
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<Uuid> = 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<Uuid> =
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,
) -> Result<BulkDeleteChatMessagesResp, AdminError> {
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<Uuid> = 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;
}
Ok(BulkDeleteChatMessagesResp { succeeded, failed })
}
pub async fn purge_chat_room_messages(
&self,
actor: &AdminActor,
room_id: Uuid,
) -> Result<PurgeChatMessagesResp, AdminError> {
actor.require(AdminRole::SuperAdmin)?;
rooms::Entity::find_by_id(room_id)
.one(self.db().read())
.await?
.ok_or(AdminError::NotFound)?;
let message_ids: Vec<Uuid> = messages::Entity::find()
.select_only()
.column(messages::Column::Id)
.filter(messages::Column::RoomId.eq(room_id))
.into_tuple::<Uuid>()
.all(self.db().read())
.await?;
let attachment_keys: Vec<String> = attachments::Entity::find()
.select_only()
.column(attachments::Column::S3Key)
.filter(attachments::Column::RoomId.eq(room_id))
.into_tuple::<String>()
.all(self.db().read())
.await?;
let thread_ids: Vec<Uuid> = threads::Entity::find()
.select_only()
.column(threads::Column::Id)
.filter(threads::Column::RoomId.eq(room_id))
.into_tuple::<Uuid>()
.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::<Uuid>::None),
)
.col_expr(
rooms::Column::UpdatedAt,
Expr::value(OffsetDateTime::now_utc()),
)
.exec(self.db().write())
.await?;
Ok(PurgeChatMessagesResp {
room_id,
messages_deleted,
attachments_deleted,
threads_deleted,
})
}
pub async fn delete_chat_room(
&self,
actor: &AdminActor,
room_id: Uuid,
) -> Result<DeleteChatRoomResp, AdminError> {
actor.require(AdminRole::SuperAdmin)?;
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<Uuid> = messages::Entity::find()
.select_only()
.column(messages::Column::Id)
.filter(messages::Column::RoomId.eq(room_id))
.into_tuple::<Uuid>()
.all(self.db().read())
.await?;
let attachment_keys: Vec<String> = attachments::Entity::find()
.select_only()
.column(attachments::Column::S3Key)
.filter(attachments::Column::RoomId.eq(room_id))
.into_tuple::<String>()
.all(self.db().read())
.await?;
let thread_ids: Vec<Uuid> = threads::Entity::find()
.select_only()
.column(threads::Column::Id)
.filter(threads::Column::RoomId.eq(room_id))
.into_tuple::<Uuid>()
.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;
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");
}
}
Ok(DeleteChatRoomResp {
room_id,
messages_deleted,
members_deleted,
attachments_deleted,
threads_deleted,
})
}
pub async fn list_chat_attachments(
&self,
query: ListChatAttachmentsQuery,
) -> Result<Page<AdminChatAttachmentDto>, 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<Uuid> = 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,
) -> Result<DeleteChatAttachmentResp, AdminError> {
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");
}
}
}
Ok(DeleteChatAttachmentResp {
id: attachment_id,
s3_key: row.s3_key,
deleted: true,
storage_deleted,
})
}
pub async fn export_chat_messages(
&self,
query: ListChatMessagesQuery,
) -> Result<Vec<AdminChatMessageDto>, 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)
}
}