imks/svc/thread.rs

//! Thread event handlers on `MessageService`.
//!
//! Threads are anchored by a root message. Participants are added when they
//! reply, get mentioned, or explicitly join. Thread events broadcast to the
//! channel room so all clients see thread activity updates.

use std::sync::Arc;

use uuid::Uuid;

use crate::ImksError;
use crate::models::message_thread_participant::JoinReason;
use crate::socket::socket::Socket;

use super::message::MessageService;

impl MessageService {
    /// Handle `thread:create` — create a new thread anchored on a root message.
    pub async fn create_thread(
        &self,
        socket: Arc<Socket>,
        data: &serde_json::Value,
    ) -> crate::ImksResult<()> {
        let user_id = self.user_id(&socket)?;
        let arr = data
            .as_array()
            .and_then(|a| a.first())
            .ok_or_else(|| ImksError::InvalidInput("Expected [payload] array".into()))?;

        let root_message_id: Uuid = Self::parse_field(arr, "root_message_id")?;
        let channel_id: Uuid = Self::parse_field(arr, "channel_id")?;

        let root_message = self
            .repo
            .get(root_message_id)
            .await?
            .ok_or_else(|| ImksError::NotFound(format!("message {root_message_id}")))?;
        if root_message.channel_id != channel_id {
            return Err(ImksError::InvalidInput(
                "Root message does not belong to channel".into(),
            ));
        }

        self.validate_channel_write(&channel_id.to_string(), &user_id.to_string())
            .await?;

        let thread = self
            .repo
            .create_thread(root_message_id, channel_id, user_id)
            .await?;

        // Creator is automatically a participant
        self.repo
            .add_thread_participant(thread.id, user_id, JoinReason::Reply.as_str())
            .await?;

        if let Some(ns) = self.namespaces.get_namespace(&socket.namespace) {
            ns.emit_to_room(
                &channel_id.to_string(),
                "thread:created",
                serde_json::to_value(&thread).unwrap_or_default(),
            )
            .await;
        }

        tracing::info!(thread_id = %thread.id, %channel_id, %user_id, "Thread created");
        Ok(())
    }

    /// Handle `thread:resolve` — toggle the resolved state of a thread.
    pub async fn resolve_thread(
        &self,
        socket: Arc<Socket>,
        data: &serde_json::Value,
    ) -> crate::ImksResult<()> {
        let user_id = self.user_id(&socket)?;
        let arr = data
            .as_array()
            .and_then(|a| a.first())
            .ok_or_else(|| ImksError::InvalidInput("Expected [payload] array".into()))?;

        let thread_id: Uuid = Self::parse_field(arr, "thread_id")?;
        let resolved: bool = Self::parse_field(arr, "resolved")?;
        let thread = self
            .repo
            .get_thread(thread_id)
            .await?
            .ok_or_else(|| ImksError::NotFound(format!("thread {thread_id}")))?;
        let channel_id = thread.channel_id;
        let channel_id_str = channel_id.to_string();
        let user_id_str = user_id.to_string();

        self.ensure_readable(&channel_id_str, &user_id_str).await?;
        self.ensure_member(&channel_id_str, &user_id_str).await?;
        self.ensure_author_or_mod(thread.created_by, &channel_id_str, user_id)
            .await?;

        self.repo
            .resolve_thread(thread_id, user_id, resolved)
            .await?;

        if let Some(ns) = self.namespaces.get_namespace(&socket.namespace) {
            ns.emit_to_room(
                &channel_id.to_string(),
                "thread:updated",
                serde_json::json!({
                    "thread_id": thread_id.to_string(),
                    "resolved": resolved,
                    "resolved_by": user_id.to_string(),
                }),
            )
            .await;
        }

        tracing::info!(%thread_id, %resolved, %user_id, "Thread resolve toggled");
        Ok(())
    }

    /// Handle `thread:join` — explicitly join a thread.
    pub async fn join_thread(
        &self,
        socket: Arc<Socket>,
        data: &serde_json::Value,
    ) -> crate::ImksResult<()> {
        let user_id = self.user_id(&socket)?;
        let arr = data
            .as_array()
            .and_then(|a| a.first())
            .ok_or_else(|| ImksError::InvalidInput("Expected [payload] array".into()))?;

        let thread_id: Uuid = Self::parse_field(arr, "thread_id")?;
        let thread = self
            .repo
            .get_thread(thread_id)
            .await?
            .ok_or_else(|| ImksError::NotFound(format!("thread {thread_id}")))?;
        let channel_id = thread.channel_id;
        let channel_id_str = channel_id.to_string();
        let user_id_str = user_id.to_string();

        self.ensure_readable(&channel_id_str, &user_id_str).await?;
        self.ensure_member(&channel_id_str, &user_id_str).await?;

        self.repo
            .add_thread_participant(thread_id, user_id, JoinReason::Joined.as_str())
            .await?;

        if let Some(ns) = self.namespaces.get_namespace(&socket.namespace) {
            ns.emit_to_room(
                &channel_id.to_string(),
                "thread:participant_joined",
                serde_json::json!({
                    "thread_id": thread_id.to_string(),
                    "user_id": user_id.to_string(),
                }),
            )
            .await;
        }

        tracing::info!(%thread_id, %user_id, "User joined thread");
        Ok(())
    }

    /// Handle `thread:leave` — leave a thread.
    pub async fn leave_thread(
        &self,
        socket: Arc<Socket>,
        data: &serde_json::Value,
    ) -> crate::ImksResult<()> {
        let user_id = self.user_id(&socket)?;
        let arr = data
            .as_array()
            .and_then(|a| a.first())
            .ok_or_else(|| ImksError::InvalidInput("Expected [payload] array".into()))?;

        let thread_id: Uuid = Self::parse_field(arr, "thread_id")?;
        let thread = self
            .repo
            .get_thread(thread_id)
            .await?
            .ok_or_else(|| ImksError::NotFound(format!("thread {thread_id}")))?;
        let channel_id = thread.channel_id;
        let channel_id_str = channel_id.to_string();
        let user_id_str = user_id.to_string();

        self.ensure_readable(&channel_id_str, &user_id_str).await?;
        self.ensure_member(&channel_id_str, &user_id_str).await?;

        self.repo
            .remove_thread_participant(thread_id, user_id)
            .await?;

        if let Some(ns) = self.namespaces.get_namespace(&socket.namespace) {
            ns.emit_to_room(
                &channel_id.to_string(),
                "thread:participant_left",
                serde_json::json!({
                    "thread_id": thread_id.to_string(),
                    "user_id": user_id.to_string(),
                }),
            )
            .await;
        }

        tracing::info!(%thread_id, %user_id, "User left thread");
        Ok(())
    }

    /// Handle `thread:list` — list threads in a channel.
    pub async fn list_threads(
        &self,
        socket: Arc<Socket>,
        data: &serde_json::Value,
    ) -> crate::ImksResult<()> {
        let arr = data
            .as_array()
            .and_then(|a| a.first())
            .ok_or_else(|| ImksError::InvalidInput("Expected [payload] array".into()))?;

        let user_id = self.user_id(&socket)?;
        let channel_id: Uuid = Self::parse_field(arr, "channel_id")?;
        let channel_id_str = channel_id.to_string();
        let user_id_str = user_id.to_string();

        self.ensure_readable(&channel_id_str, &user_id_str).await?;
        self.ensure_member(&channel_id_str, &user_id_str).await?;

        let threads = self.repo.list_threads(channel_id).await?;
        let _ = socket.emit(
            "thread:loaded",
            serde_json::to_value(&threads).unwrap_or_default(),
        );
        Ok(())
    }
}