use serde::{Deserialize, Serialize}; use uuid::Uuid; use crate::error::AppError; use crate::models::channels::Channel; use crate::models::common::{ChannelKind, ChannelType, Role, Visibility}; use crate::models::workspaces::Workspace; use crate::service::ImService; use super::session::ImSession; use super::util::*; use crate::service::im::events::{ChannelAction, ChannelEvent, ImEvent}; #[derive(Deserialize, Serialize, Clone, Debug, utoipa::ToSchema)] pub struct CreateChannelParams { pub name: String, pub topic: Option, pub description: Option, pub channel_type: Option, pub channel_kind: Option, pub visibility: Option, pub category_id: Option, pub parent_channel_id: Option, pub nsfw: Option, pub rate_limit_per_user: Option, } #[derive(Deserialize, Serialize, Clone, Debug, utoipa::ToSchema)] pub struct UpdateChannelParams { pub name: Option, pub topic: Option, pub description: Option, pub visibility: Option, pub category_id: Option, pub position: Option, pub nsfw: Option, pub rate_limit_per_user: Option, pub archived: Option, pub read_only: Option, } #[derive(Deserialize, Serialize, Clone, Debug, utoipa::ToSchema)] pub struct ChannelListFilters { pub channel_type: Option, pub channel_kind: Option, pub category_id: Option, pub archived: Option, } impl ImService { pub async fn channel_list( &self, ctx: &ImSession, wk_name: &str, filters: ChannelListFilters, limit: i64, offset: i64, ) -> Result, AppError> { let user_uid = ctx.user; let ws = self.resolve_workspace(wk_name).await?; self.ensure_workspace_readable(user_uid, &ws).await?; let (limit, offset) = clamp_limit_offset(limit, offset); let kind = filters .channel_kind .as_deref() .and_then(|s| s.parse::().ok()) .filter(|k| *k != ChannelKind::Unknown); let ch_type = filters .channel_type .as_deref() .and_then(|s| s.parse::().ok()) .filter(|t| *t != ChannelType::Unknown); sqlx::query_as::<_, Channel>( "SELECT id, workspace_id, repo_id, category_id, created_by, name, topic, description, \ channel_type, channel_kind, visibility, position, nsfw, archived, read_only, \ bitrate, user_limit, rtc_region, \ default_auto_archive_duration, default_reaction_emoji, default_sort_order, \ default_forum_layout, require_tag, available_tags, default_thread_rate_limit, \ rate_limit_per_user, parent_channel_id, \ last_message_id, last_message_at, archived_at, created_at, updated_at, deleted_at \ FROM channel \ WHERE workspace_id = $1 AND deleted_at IS NULL \ AND ($2::text IS NULL OR channel_kind::text = $2) \ AND ($3::text IS NULL OR channel_type::text = $3) \ AND ($4::uuid IS NULL OR category_id = $4) \ AND ($5::bool IS NULL OR archived = $5) \ ORDER BY position ASC NULLS LAST, name ASC \ LIMIT $6 OFFSET $7", ) .bind(ws.id) .bind(kind.map(|k| k.to_string())) .bind(ch_type.map(|t| t.to_string())) .bind(filters.category_id) .bind(filters.archived) .bind(limit) .bind(offset) .fetch_all(self.ctx.db.reader()) .await .map_err(AppError::Database) } pub async fn channel_get( &self, ctx: &ImSession, _wk_name: &str, channel_id: Uuid, ) -> Result { let user_uid = ctx.user; let channel = self.resolve_channel(channel_id).await?; self.ensure_channel_readable(user_uid, &channel).await?; Ok(channel) } #[tracing::instrument(skip(self, ctx, params), fields(name = %params.name))] pub async fn channel_create( &self, ctx: &ImSession, wk_name: &str, params: CreateChannelParams, request_id: Uuid, ) -> Result { let user_uid = ctx.user; let name = required_text(params.name, "name")?; if name.len() > MAX_CHANNEL_NAME { return Err(AppError::BadRequest("channel name too long".into())); } let ws = self.resolve_workspace(wk_name).await?; self.ensure_workspace_role_at_least(user_uid, &ws, Role::Member) .await?; if let Some(topic) = ¶ms.topic && topic.len() > MAX_CHANNEL_TOPIC { return Err(AppError::BadRequest("channel topic too long".into())); } let ch_kind = parse_enum( params.channel_kind, ChannelKind::Text, ChannelKind::Unknown, "channel_kind", )?; let ch_type = parse_enum( params.channel_type, ChannelType::Public, ChannelType::Unknown, "channel_type", )?; let visibility = parse_enum( params.visibility, Visibility::Public, Visibility::Unknown, "visibility", )?; let now = chrono::Utc::now(); let channel_id = Uuid::now_v7(); let mut txn = self .ctx .db .writer() .begin() .await .map_err(|_| AppError::TxnError)?; let channel = sqlx::query_as::<_, Channel>( "INSERT INTO channel \ (id, workspace_id, repo_id, category_id, created_by, name, topic, description, \ channel_type, channel_kind, visibility, position, nsfw, archived, read_only, \ rate_limit_per_user, parent_channel_id, created_at, updated_at) \ VALUES ($1, $2, NULL, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, false, false, \ $13, $14, $15, $15) \ RETURNING id, workspace_id, repo_id, category_id, created_by, name, topic, description, \ channel_type, channel_kind, visibility, position, nsfw, archived, read_only, \ bitrate, user_limit, rtc_region, \ default_auto_archive_duration, default_reaction_emoji, default_sort_order, \ default_forum_layout, require_tag, available_tags, default_thread_rate_limit, \ rate_limit_per_user, parent_channel_id, \ last_message_id, last_message_at, archived_at, created_at, updated_at, deleted_at", ) .bind(channel_id) .bind(ws.id) .bind(params.category_id) .bind(user_uid) .bind(&name) .bind(params.topic.as_deref()) .bind(params.description.as_deref()) .bind(ch_type) .bind(ch_kind) .bind(visibility) .bind(0_i32) // position .bind(params.nsfw.unwrap_or(false)) .bind(params.rate_limit_per_user) .bind(params.parent_channel_id) .bind(now) .fetch_one(&mut *txn) .await .map_err(AppError::Database)?; // Auto-add creator as channel member with owner role sqlx::query( "INSERT INTO channel_member \ (id, channel_id, user_id, role, status, muted, pinned, created_at, updated_at) \ VALUES ($1, $2, $3, 'owner', 'active', false, false, $4, $4)", ) .bind(Uuid::now_v7()) .bind(channel_id) .bind(user_uid) .bind(now) .execute(&mut *txn) .await .map_err(AppError::Database)?; txn.commit().await.map_err(|_| AppError::TxnError)?; tracing::info!(channel_id = %channel_id, name = %name, "Channel created"); let event = ChannelEvent { channel_id: channel.id, action: ChannelAction::Created, workspace_name: Some(ws.name.clone()), }; self.publish(&format!("im.channel.{}", ws.name), request_id, &event) .await; self.emit_event(ImEvent::Channel { request_id, data: event, }); Ok(channel) } pub async fn channel_update( &self, ctx: &ImSession, _wk_name: &str, channel_id: Uuid, params: UpdateChannelParams, request_id: Uuid, ) -> Result { let user_uid = ctx.user; let channel = self.resolve_channel(channel_id).await?; self.ensure_channel_editable(user_uid, &channel).await?; if let Some(name) = ¶ms.name && name.len() > MAX_CHANNEL_NAME { return Err(AppError::BadRequest("channel name too long".into())); } let visibility = match params.visibility { Some(ref v) => parse_enum( Some(v.clone()), channel.visibility, Visibility::Unknown, "visibility", )?, None => channel.visibility, }; let now = chrono::Utc::now(); let new_name = merge_optional_text(params.name, Some(channel.name.clone())) .map(|s| s.trim().to_string()) .unwrap_or(channel.name); let new_topic = merge_optional_text(params.topic, channel.topic.clone()); let new_desc = merge_optional_text(params.description, channel.description.clone()); let updated = sqlx::query_as::<_, Channel>( "UPDATE channel SET \ name = $1, topic = $2, description = $3, visibility = $4, \ category_id = COALESCE($5, category_id), position = COALESCE($6, position), \ nsfw = COALESCE($7, nsfw), archived = COALESCE($8, archived), \ read_only = COALESCE($9, read_only), \ rate_limit_per_user = COALESCE($10, rate_limit_per_user), \ updated_at = $11 \ WHERE id = $12 AND deleted_at IS NULL \ RETURNING id, workspace_id, repo_id, category_id, created_by, name, topic, description, \ channel_type, channel_kind, visibility, position, nsfw, archived, read_only, \ bitrate, user_limit, rtc_region, \ default_auto_archive_duration, default_reaction_emoji, default_sort_order, \ default_forum_layout, require_tag, available_tags, default_thread_rate_limit, \ rate_limit_per_user, parent_channel_id, \ last_message_id, last_message_at, archived_at, created_at, updated_at, deleted_at", ) .bind(&new_name) .bind(&new_topic) .bind(&new_desc) .bind(visibility) .bind(params.category_id) .bind(params.position) .bind(params.nsfw) .bind(params.archived) .bind(params.read_only) .bind(params.rate_limit_per_user) .bind(now) .bind(channel_id) .fetch_one(self.ctx.db.writer()) .await .map_err(AppError::Database)?; let event = ChannelEvent { channel_id, action: ChannelAction::Updated, workspace_name: None, }; self.publish(&format!("im.channel.{}", channel_id), request_id, &event) .await; self.emit_event(ImEvent::Channel { request_id, data: event, }); Ok(updated) } pub async fn channel_delete( &self, ctx: &ImSession, _wk_name: &str, channel_id: Uuid, request_id: Uuid, ) -> Result<(), AppError> { let user_uid = ctx.user; let channel = self.resolve_channel(channel_id).await?; self.ensure_channel_admin(user_uid, &channel).await?; let now = chrono::Utc::now(); let result = sqlx::query( "UPDATE channel SET deleted_at = $1, updated_at = $1 \ WHERE id = $2 AND deleted_at IS NULL", ) .bind(now) .bind(channel_id) .execute(self.ctx.db.writer()) .await .map_err(AppError::Database)?; ensure_affected(result.rows_affected(), "channel not found")?; let event = ChannelEvent { channel_id, action: ChannelAction::Deleted, workspace_name: None, }; self.publish(&format!("im.channel.{}", channel_id), request_id, &event) .await; self.emit_event(ImEvent::Channel { request_id, data: event, }); Ok(()) } pub(crate) async fn resolve_workspace(&self, wk_name: &str) -> Result { Workspace::find_by_name(self.ctx.db.reader(), wk_name) .await .map_err(AppError::Database)? .ok_or(AppError::NotFound("workspace not found".into())) } pub(crate) async fn resolve_channel(&self, channel_id: Uuid) -> Result { sqlx::query_as::<_, Channel>( "SELECT id, workspace_id, repo_id, category_id, created_by, name, topic, description, \ channel_type, channel_kind, visibility, position, nsfw, archived, read_only, \ bitrate, user_limit, rtc_region, \ default_auto_archive_duration, default_reaction_emoji, default_sort_order, \ default_forum_layout, require_tag, available_tags, default_thread_rate_limit, \ rate_limit_per_user, parent_channel_id, \ last_message_id, last_message_at, archived_at, created_at, updated_at, deleted_at \ FROM channel WHERE id = $1 AND deleted_at IS NULL", ) .bind(channel_id) .fetch_optional(self.ctx.db.reader()) .await .map_err(AppError::Database)? .ok_or(AppError::NotFound("channel not found".into())) } pub(crate) async fn ensure_workspace_readable( &self, user_uid: Uuid, ws: &Workspace, ) -> Result<(), AppError> { if Workspace::is_readable(self.ctx.db.reader(), ws, user_uid) .await .map_err(AppError::Database)? { Ok(()) } else { Err(AppError::Unauthorized) } } pub(crate) async fn ensure_workspace_role_at_least( &self, user_uid: Uuid, ws: &Workspace, min_role: Role, ) -> Result { let role = Workspace::user_role(self.ctx.db.reader(), ws.id, user_uid, ws.owner_id) .await .map_err(AppError::Database)? .unwrap_or(Role::Unknown); if role_level(role) < role_level(min_role) { return Err(AppError::Unauthorized); } Ok(role) } pub(crate) async fn ensure_channel_readable( &self, user_uid: Uuid, channel: &Channel, ) -> Result<(), AppError> { if channel.created_by == user_uid { return Ok(()); } let is_member = self.is_channel_member(channel.id, user_uid).await?; if is_member { return Ok(()); } if channel.visibility == Visibility::Public { let ws = Workspace::find_by_id(self.ctx.db.reader(), channel.workspace_id) .await .map_err(AppError::Database)? .ok_or(AppError::NotFound("workspace not found".into()))?; if Workspace::is_readable(self.ctx.db.reader(), &ws, user_uid) .await .map_err(AppError::Database)? { return Ok(()); } } Err(AppError::Unauthorized) } #[allow(dead_code)] pub(crate) async fn ensure_channel_member( &self, user_uid: Uuid, channel: &Channel, ) -> Result<(), AppError> { if channel.created_by == user_uid { return Ok(()); } let is_member = self.is_channel_member(channel.id, user_uid).await?; if is_member { Ok(()) } else { Err(AppError::Forbidden("not a channel member".into())) } } pub(crate) async fn ensure_channel_editable( &self, user_uid: Uuid, channel: &Channel, ) -> Result<(), AppError> { if channel.created_by == user_uid { return Ok(()); } let role = self.channel_member_role(channel.id, user_uid).await?; if role_level(role) >= role_level(Role::Member) { return Ok(()); } self.ensure_workspace_role_at_least( user_uid, &Workspace::find_by_id(self.ctx.db.reader(), channel.workspace_id) .await .map_err(AppError::Database)? .ok_or(AppError::NotFound("workspace not found".into()))?, Role::Admin, ) .await?; Ok(()) } pub(crate) async fn ensure_channel_admin( &self, user_uid: Uuid, channel: &Channel, ) -> Result<(), AppError> { let role = self.channel_member_role(channel.id, user_uid).await?; if role_level(role) >= role_level(Role::Admin) { return Ok(()); } self.ensure_workspace_role_at_least( user_uid, &Workspace::find_by_id(self.ctx.db.reader(), channel.workspace_id) .await .map_err(AppError::Database)? .ok_or(AppError::NotFound("workspace not found".into()))?, Role::Admin, ) .await?; Ok(()) } pub(crate) async fn is_channel_member( &self, channel_id: Uuid, user_uid: Uuid, ) -> Result { sqlx::query_scalar( "SELECT EXISTS(SELECT 1 FROM channel_member \ WHERE channel_id = $1 AND user_id = $2 AND status = 'active')", ) .bind(channel_id) .bind(user_uid) .fetch_one(self.ctx.db.reader()) .await .map_err(AppError::Database) } pub(crate) async fn channel_member_role( &self, channel_id: Uuid, user_uid: Uuid, ) -> Result { let role: Option = sqlx::query_scalar( "SELECT role::text FROM channel_member \ WHERE channel_id = $1 AND user_id = $2 AND status = 'active'", ) .bind(channel_id) .bind(user_uid) .fetch_optional(self.ctx.db.reader()) .await .map_err(AppError::Database)?; Ok(role .as_deref() .and_then(|s| s.parse::().ok()) .unwrap_or(Role::Unknown)) } pub(crate) async fn increment_channel_stat( &self, channel_id: Uuid, delta: i32, now: chrono::DateTime, txn: &mut sqlx::Transaction<'_, sqlx::Postgres>, ) -> Result<(), AppError> { sqlx::query( "UPDATE channel_stats SET members_count = members_count + $1, \ last_activity_at = $2, updated_at = $2 WHERE channel_id = $3", ) .bind(delta) .bind(now) .bind(channel_id) .execute(&mut **txn) .await .map_err(AppError::Database)?; Ok(()) } }