cnats/src/chat.rs

use leptos::prelude::*;
use serde::{Deserialize, Serialize};

/// Rooms available in the UI. Each maps to the NATS subject
/// `chat.room.<name>`, so any other NATS client on the bus can join in.
/// The third field is the Kanidm group (via the `groups` OIDC claim,
/// see `oauth2 update-claim-map`) required to read/post in that room -
/// `None` means open to anyone in `cnats_users`.
pub const ROOMS: &[(&str, &str, Option<&str>)] = &[
    ("lobby", "general traffic", None),
    ("dev", "build & ship", Some("developers")),
    ("ops", "incidents & infra", Some("developers")),
    ("random", "off the record", None),
];

pub const DEFAULT_ROOM: &str = "lobby";

pub fn is_valid_room(room: &str) -> bool {
    ROOMS.iter().any(|(name, _, _)| *name == room)
}

pub fn room_subject(room: &str) -> String {
    format!("chat.room.{room}")
}

/// Whether `user` may read/post in `room`. `false` for an unknown room -
/// callers should check `is_valid_room` separately if they need to tell
/// "unknown room" and "not authorized" apart in the error they return.
/// Synchronous and I/O-free: the user's groups are already baked into
/// their session (from the `groups` OIDC claim at login), so this never
/// needs a live Kanidm round-trip - and never gets more current than
/// that login until they sign in again.
pub fn is_authorized_for_room(user: &crate::auth::User, room: &str) -> bool {
    ROOMS
        .iter()
        .find(|(name, _, _)| *name == room)
        .is_some_and(|(_, _, required_group)| {
            required_group.is_none_or(|g| user.groups.iter().any(|ug| ug == g))
        })
}

/// A single chat message as it travels over NATS (JSON-encoded payload).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChatMessage {
    pub id: String,
    pub room: String,
    pub username: String,
    pub display_name: String,
    pub text: String,
    /// Pre-formatted UTC wall-clock time, e.g. "14:03:27".
    pub time: String,
    /// Publish instant as UTC epoch milliseconds; orders history.
    #[serde(default)]
    pub ts: i64,
}

/// How many messages `room_history` backfills when joining a room.
pub const HISTORY_LIMIT: i64 = 100;

/// Publishes a message to the room's NATS subject. Requires a signed-in
/// session; the sender identity always comes from the session, never from
/// the client.
#[server]
pub async fn send_message(room: String, text: String) -> Result<(), ServerFnError> {
    use crate::auth::{User, SESSION_USER_KEY};
    use crate::server::AppState;

    let text = text.trim().to_string();
    if text.is_empty() || text.len() > 2000 {
        return Err(ServerFnError::new("message must be 1..=2000 characters"));
    }
    if !is_valid_room(&room) {
        return Err(ServerFnError::new("unknown room"));
    }

    let session: tower_sessions::Session = leptos_axum::extract().await?;
    let Some(user) = session
        .get::<User>(SESSION_USER_KEY)
        .await
        .map_err(|e| ServerFnError::new(e.to_string()))?
    else {
        return Err(ServerFnError::new("not signed in"));
    };
    if !is_authorized_for_room(&user, &room) {
        return Err(ServerFnError::new("not authorized for this room"));
    }

    let state = expect_context::<AppState>();
    let now = chrono::Utc::now();
    let msg = ChatMessage {
        id: uuid::Uuid::new_v4().to_string(),
        room: room.clone(),
        username: user.username,
        display_name: user.display_name,
        text,
        time: now.format("%H:%M:%S").to_string(),
        ts: now.timestamp_millis(),
    };
    let payload = serde_json::to_vec(&msg).map_err(|e| ServerFnError::new(e.to_string()))?;
    state
        .nats
        .publish(room_subject(&room), payload.into())
        .await
        .map_err(|e| ServerFnError::new(format!("nats publish failed: {e}")))?;
    Ok(())
}

/// Returns the most recent messages for a room (oldest first), read from the
/// Postgres archive maintained by the JetStream consumer.
#[server]
pub async fn room_history(room: String) -> Result<Vec<ChatMessage>, ServerFnError> {
    use crate::auth::{User, SESSION_USER_KEY};
    use crate::server::AppState;

    if !is_valid_room(&room) {
        return Err(ServerFnError::new("unknown room"));
    }
    let session: tower_sessions::Session = leptos_axum::extract().await?;
    let Some(user) = session
        .get::<User>(SESSION_USER_KEY)
        .await
        .map_err(|e| ServerFnError::new(e.to_string()))?
    else {
        return Err(ServerFnError::new("not signed in"));
    };
    if !is_authorized_for_room(&user, &room) {
        return Err(ServerFnError::new("not authorized for this room"));
    }

    let state = expect_context::<AppState>();
    crate::server::store::recent(&state.pool, &room, HISTORY_LIMIT)
        .await
        .map_err(|e| ServerFnError::new(format!("history unavailable: {e}")))
}