oxspeak_server/src/routes/conversation/handlers.rs

use crate::core::state::AppState;
use crate::domain::dto::conversation::{
    ConversationParticipantResponse, ConversationResponse, CreateConversationRequest,
    ForkConversationRequest,
};
use crate::domain::events::channel::ChannelCreatedEvent;
use crate::http::context::CurrentUser;
use crate::http::error::HTTPError;
use crate::models::{channel, channel_user, message, user};
use axum::{
    Json,
    extract::{Path, State},
};
use chrono::Utc;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, QueryOrder, Set, TransactionTrait,
};
use std::collections::HashSet;
use uuid::Uuid;

async fn member_ids(state: &AppState, channel_id: Uuid) -> Result<Vec<Uuid>, HTTPError> {
    Ok(channel_user::Entity::find()
        .filter(channel_user::Column::ChannelId.eq(channel_id))
        .all(&state.db)
        .await?
        .into_iter()
        .map(|m| m.user_id)
        .collect())
}

async fn response(
    state: &AppState,
    current_user_id: Uuid,
    channel: channel::Model,
) -> Result<ConversationResponse, HTTPError> {
    let ids = member_ids(state, channel.id).await?;
    let users = user::Entity::find()
        .filter(user::Column::Id.is_in(ids.clone()))
        .all(&state.db)
        .await?;
    let mut participants: Vec<_> = users
        .into_iter()
        .map(|u| ConversationParticipantResponse {
            id: u.id,
            username: u.username,
        })
        .collect();
    participants.sort_by(|a, b| a.username.to_lowercase().cmp(&b.username.to_lowercase()));
    let title = participants
        .iter()
        .filter(|p| p.id != current_user_id)
        .map(|p| p.username.clone())
        .collect::<Vec<_>>()
        .join(", ");
    let last_message = message::Entity::find()
        .filter(message::Column::ChannelId.eq(channel.id))
        .order_by_desc(message::Column::Id)
        .one(&state.db)
        .await?
        .map(|m| m.content);
    let unread_count = state
        .repositories
        .read_state
        .unread_counts(&[channel.id], current_user_id)
        .await?
        .get(&channel.id)
        .copied()
        .unwrap_or(0);
    Ok(ConversationResponse {
        id: channel.id,
        title: if title.is_empty() {
            "Discussion".into()
        } else {
            title
        },
        participants,
        last_message,
        unread_count,
        created_at: channel.created_at,
        updated_at: channel.updated_at,
    })
}

async fn create_channel(state: &AppState, ids: &[Uuid]) -> Result<channel::Model, HTTPError> {
    let txn = state.db.begin().await?;
    let channel = channel::ActiveModel {
        server_id: Set(None),
        category_id: Set(None),
        channel_type: Set(channel::ChannelType::DM),
        name: Set(None),
        ..Default::default()
    }
    .insert(&txn)
    .await?;
    for user_id in ids {
        channel_user::ActiveModel {
            channel_id: Set(channel.id),
            user_id: Set(*user_id),
            role: Set("member".into()),
            joined_at: Set(Utc::now()),
            ..Default::default()
        }
        .insert(&txn)
        .await?;
    }
    txn.commit().await?;
    state
        .services
        .realtime_registry
        .set_channel_users(channel.id, ids.iter().copied());
    state.event_bus.emit(ChannelCreatedEvent {
        channel: channel.clone(),
    });
    Ok(channel)
}

async fn validate_ids(state: &AppState, ids: &[Uuid]) -> Result<(), HTTPError> {
    if ids.is_empty() {
        return Err(HTTPError::BadRequest(
            "At least one participant is required".into(),
        ));
    }
    let found = user::Entity::find()
        .filter(user::Column::Id.is_in(ids.to_vec()))
        .all(&state.db)
        .await?;
    if found.len() != ids.iter().collect::<HashSet<_>>().len() {
        return Err(HTTPError::BadRequest("Unknown participant".into()));
    }
    Ok(())
}

#[utoipa::path(get, path = "/conversations", responses((status = 200, body = [ConversationResponse])), tag = "Conversations", security(("bearerAuth" = [])))]
pub async fn list(
    user: CurrentUser,
    State(state): State<AppState>,
) -> Result<Json<Vec<ConversationResponse>>, HTTPError> {
    let memberships = channel_user::Entity::find()
        .filter(channel_user::Column::UserId.eq(user.id))
        .all(&state.db)
        .await?;
    let ids: Vec<_> = memberships.into_iter().map(|m| m.channel_id).collect();
    if ids.is_empty() {
        return Ok(Json(Vec::new()));
    }
    let channels = channel::Entity::find()
        .filter(channel::Column::Id.is_in(ids))
        .filter(channel::Column::ChannelType.eq(channel::ChannelType::DM))
        .order_by_desc(channel::Column::UpdatedAt)
        .all(&state.db)
        .await?;
    let mut result = Vec::with_capacity(channels.len());
    for channel in channels {
        result.push(response(&state, user.id, channel).await?);
    }
    Ok(Json(result))
}

#[utoipa::path(post, path = "/conversations", request_body = CreateConversationRequest, responses((status = 200, body = ConversationResponse)), tag = "Conversations", security(("bearerAuth" = [])))]
pub async fn create(
    user: CurrentUser,
    State(state): State<AppState>,
    Json(payload): Json<CreateConversationRequest>,
) -> Result<Json<ConversationResponse>, HTTPError> {
    let mut ids = payload.user_ids;
    ids.push(user.id);
    ids.sort_unstable();
    ids.dedup();
    validate_ids(&state, &ids).await?;
    let channels = channel_user::Entity::find()
        .filter(channel_user::Column::UserId.eq(user.id))
        .all(&state.db)
        .await?;
    for membership in channels {
        if let Some(channel) = channel::Entity::find_by_id(membership.channel_id)
            .filter(channel::Column::ChannelType.eq(channel::ChannelType::DM))
            .one(&state.db)
            .await?
        {
            let existing: HashSet<_> = member_ids(&state, channel.id).await?.into_iter().collect();
            if existing == ids.iter().copied().collect() {
                return Ok(Json(response(&state, user.id, channel).await?));
            }
        }
    }
    let channel = create_channel(&state, &ids).await?;
    Ok(Json(response(&state, user.id, channel).await?))
}

#[utoipa::path(post, path = "/conversations/{id}/fork", request_body = ForkConversationRequest, params(("id" = Uuid, Path)), responses((status = 200, body = ConversationResponse)), tag = "Conversations", security(("bearerAuth" = [])))]
pub async fn fork(
    user: CurrentUser,
    State(state): State<AppState>,
    Path(id): Path<Uuid>,
    Json(payload): Json<ForkConversationRequest>,
) -> Result<Json<ConversationResponse>, HTTPError> {
    let source = channel::Entity::find_by_id(id)
        .one(&state.db)
        .await?
        .ok_or(HTTPError::NotFound)?;
    if source.channel_type != channel::ChannelType::DM
        || !member_ids(&state, id).await?.contains(&user.id)
    {
        return Err(HTTPError::Forbidden);
    }
    let mut ids = member_ids(&state, id).await?;
    ids.extend(payload.user_ids);
    ids.sort_unstable();
    ids.dedup();
    validate_ids(&state, &ids).await?;
    let channel = create_channel(&state, &ids).await?;
    Ok(Json(response(&state, user.id, channel).await?))
}