gitks/service/im/channels.rs

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<String>,
    pub description: Option<String>,
    pub channel_type: Option<String>,
    pub channel_kind: Option<String>,
    pub visibility: Option<String>,
    pub category_id: Option<Uuid>,
    pub parent_channel_id: Option<Uuid>,
    pub nsfw: Option<bool>,
    pub rate_limit_per_user: Option<i32>,
}

#[derive(Deserialize, Serialize, Clone, Debug, utoipa::ToSchema)]
pub struct UpdateChannelParams {
    pub name: Option<String>,
    pub topic: Option<String>,
    pub description: Option<String>,
    pub visibility: Option<String>,
    pub category_id: Option<Uuid>,
    pub position: Option<i32>,
    pub nsfw: Option<bool>,
    pub rate_limit_per_user: Option<i32>,
    pub archived: Option<bool>,
    pub read_only: Option<bool>,
}

#[derive(Deserialize, Serialize, Clone, Debug, utoipa::ToSchema)]
pub struct ChannelListFilters {
    pub channel_type: Option<String>,
    pub channel_kind: Option<String>,
    pub category_id: Option<Uuid>,
    pub archived: Option<bool>,
}

impl ImService {
    pub async fn channel_list(
        &self,
        ctx: &ImSession,
        wk_name: &str,
        filters: ChannelListFilters,
        limit: i64,
        offset: i64,
    ) -> Result<Vec<Channel>, 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::<ChannelKind>().ok())
            .filter(|k| *k != ChannelKind::Unknown);
        let ch_type = filters
            .channel_type
            .as_deref()
            .and_then(|s| s.parse::<ChannelType>().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<Channel, AppError> {
        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<Channel, AppError> {
        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) = &params.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<Channel, AppError> {
        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) = &params.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, AppError> {
        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<Channel, AppError> {
        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<Role, AppError> {
        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<bool, AppError> {
        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<Role, AppError> {
        let role: Option<String> = 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::<Role>().ok())
            .unwrap_or(Role::Unknown))
    }

    pub(crate) async fn increment_channel_stat(
        &self,
        channel_id: Uuid,
        delta: i32,
        now: chrono::DateTime<chrono::Utc>,
        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(())
    }
}