From 5fcd49e84786e241c6a54408f4e128620f52ee75e33a4e8b0b0c2732edbb7d4d Mon Sep 17 00:00:00 2001 From: Evan Carroll Date: Sat, 17 Jan 2026 23:47:02 -0600 Subject: [PATCH] update to support user expire, timeout, and disconnect --- apps/chattyness-app/Cargo.toml | 2 + apps/chattyness-app/src/main.rs | 60 +- config.toml | 18 + .../src/queries/channel_members.rs | 24 + crates/chattyness-db/src/ws_messages.rs | 23 + crates/chattyness-shared/Cargo.toml | 5 + crates/chattyness-shared/src/config.rs | 88 +++ crates/chattyness-shared/src/lib.rs | 6 +- crates/chattyness-user-ui/Cargo.toml | 1 + .../chattyness-user-ui/src/api/websocket.rs | 525 ++++++++++-------- crates/chattyness-user-ui/src/app.rs | 11 + .../src/components/avatar_canvas.rs | 14 +- .../src/components/scene_viewer.rs | 41 +- .../src/components/ws_client.rs | 77 ++- crates/chattyness-user-ui/src/pages/realm.rs | 46 +- db/schema/functions/001_helpers.sql | 41 ++ 16 files changed, 744 insertions(+), 238 deletions(-) create mode 100644 config.toml create mode 100644 crates/chattyness-shared/src/config.rs diff --git a/apps/chattyness-app/Cargo.toml b/apps/chattyness-app/Cargo.toml index 18c510b..ce87e27 100644 --- a/apps/chattyness-app/Cargo.toml +++ b/apps/chattyness-app/Cargo.toml @@ -15,6 +15,7 @@ chattyness-admin-ui.workspace = true chattyness-user-ui.workspace = true chattyness-db.workspace = true chattyness-error.workspace = true +chattyness-shared.workspace = true leptos.workspace = true leptos_meta.workspace = true leptos_router.workspace = true @@ -57,6 +58,7 @@ ssr = [ "chattyness-user-ui/ssr", "chattyness-db/ssr", "chattyness-error/ssr", + "chattyness-shared/ssr", ] # Unified hydrate feature - admin routes are lazy-loaded via #[lazy] macro hydrate = [ diff --git a/apps/chattyness-app/src/main.rs b/apps/chattyness-app/src/main.rs index d5d5a4a..79019da 100644 --- a/apps/chattyness-app/src/main.rs +++ b/apps/chattyness-app/src/main.rs @@ -14,12 +14,14 @@ mod server { use leptos_axum::{generate_route_list, LeptosRoutes}; use sqlx::postgres::PgPoolOptions; use std::net::SocketAddr; - use std::path::Path; + use std::path::{Path, PathBuf}; use std::sync::Arc; + use std::time::Duration; use tower_http::services::ServeDir; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; use chattyness_app::{combined_shell, CombinedApp, CombinedAppState}; + use chattyness_shared::AppConfig; use chattyness_user_ui::api::WebSocketState; /// CLI arguments. @@ -42,6 +44,10 @@ mod server { /// Use secure cookies #[arg(long, env = "SECURE_COOKIES", default_value = "false")] secure_cookies: bool, + + /// Path to TOML configuration file + #[arg(long, short = 'c', env = "CONFIG_FILE")] + config: Option, } pub async fn main() -> Result<(), Box> { @@ -60,6 +66,10 @@ mod server { // Parse arguments let args = Args::parse(); + // Load configuration + let config = AppConfig::load(args.config.as_deref())?; + tracing::info!("Configuration loaded: {:?}", config); + tracing::info!("Starting Chattyness App Server"); // Create database pool for app access (fixed connection string, RLS-constrained) @@ -74,6 +84,53 @@ mod server { tracing::info!("Connected to database (app role with RLS)"); + // Startup cleanup: clear all instance_members if configured + // Uses SECURITY DEFINER function to bypass RLS + if config.cleanup.clear_on_startup { + tracing::info!("Clearing stale instance_members on startup..."); + let deleted: i64 = sqlx::query_scalar("SELECT scene.clear_all_instance_members()") + .fetch_one(&pool) + .await?; + tracing::info!("Cleared {} stale instance_members", deleted); + } + + // Spawn background task for periodic stale member cleanup + // Uses SECURITY DEFINER function to bypass RLS + { + let cleanup_pool = pool.clone(); + let cleanup_config = config.cleanup.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs( + cleanup_config.reap_interval_secs, + )); + loop { + interval.tick().await; + let threshold = cleanup_config.stale_threshold_secs as f64; + match sqlx::query_scalar::<_, i64>( + "SELECT scene.clear_stale_instance_members($1)", + ) + .bind(threshold) + .fetch_one(&cleanup_pool) + .await + { + Ok(deleted) => { + if deleted > 0 { + tracing::info!("Reaped {} stale instance_members", deleted); + } + } + Err(e) => { + tracing::error!("Stale member cleanup failed: {:?}", e); + } + } + } + }); + tracing::info!( + "Started stale member reaper (interval: {}s, threshold: {}s)", + config.cleanup.reap_interval_secs, + config.cleanup.stale_threshold_secs + ); + } + // Configure Leptos let cargo_toml = concat!(env!("CARGO_MANIFEST_DIR"), "/Cargo.toml"); let conf = get_configuration(Some(cargo_toml)).unwrap(); @@ -118,6 +175,7 @@ mod server { pool: pool.clone(), leptos_options: leptos_options.clone(), ws_state: ws_state.clone(), + ws_config: config.websocket.clone(), }; let admin_api_state = chattyness_admin_ui::AdminAppState { pool: pool.clone(), diff --git a/config.toml b/config.toml new file mode 100644 index 0000000..1d30ae2 --- /dev/null +++ b/config.toml @@ -0,0 +1,18 @@ +# Chattyness Application Configuration + +[websocket] +# Timeout for receiving messages from client before considering connection dead (seconds) +recv_timeout_secs = 45 + +# Interval for client to send ping to keep connection alive (seconds) +client_ping_interval_secs = 30 + +[cleanup] +# Interval for running stale member cleanup job (seconds) +reap_interval_secs = 120 + +# Members with no activity for this duration are considered stale (seconds) +stale_threshold_secs = 120 + +# Clear all instance_members on server startup (recommended for single-server deployments) +clear_on_startup = true diff --git a/crates/chattyness-db/src/queries/channel_members.rs b/crates/chattyness-db/src/queries/channel_members.rs index 1567fb5..9e19d89 100644 --- a/crates/chattyness-db/src/queries/channel_members.rs +++ b/crates/chattyness-db/src/queries/channel_members.rs @@ -273,3 +273,27 @@ pub async fn set_afk<'e>( Ok(()) } + +/// Update the last_moved_at timestamp to keep the member alive. +/// +/// Called when a ping is received from the client to prevent +/// the member from being reaped by the stale member cleanup. +pub async fn touch_member<'e>( + executor: impl PgExecutor<'e>, + channel_id: Uuid, + user_id: Uuid, +) -> Result<(), AppError> { + sqlx::query( + r#" + UPDATE scene.instance_members + SET last_moved_at = now() + WHERE instance_id = $1 AND user_id = $2 + "#, + ) + .bind(channel_id) + .bind(user_id) + .execute(executor) + .await?; + + Ok(()) +} diff --git a/crates/chattyness-db/src/ws_messages.rs b/crates/chattyness-db/src/ws_messages.rs index 2102dd1..4a54832 100644 --- a/crates/chattyness-db/src/ws_messages.rs +++ b/crates/chattyness-db/src/ws_messages.rs @@ -7,6 +7,25 @@ use uuid::Uuid; use crate::models::{AvatarRenderData, ChannelMemberInfo, ChannelMemberWithAvatar, LooseProp}; +/// WebSocket configuration sent to client on connect. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WsConfig { + /// Interval for client to send ping to keep connection alive (seconds). + pub ping_interval_secs: u64, +} + +/// Reason for member disconnect. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum DisconnectReason { + /// Graceful disconnect (browser close, normal WebSocket close). + Graceful, + /// Scene navigation (custom close code 4000). + SceneChange, + /// Timeout (connection lost, no ping response). + Timeout, +} + /// Client-to-server WebSocket messages. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] @@ -60,6 +79,8 @@ pub enum ServerMessage { member: ChannelMemberInfo, /// All current members with avatars. members: Vec, + /// WebSocket configuration for the client. + config: WsConfig, }, /// A member joined the channel. @@ -74,6 +95,8 @@ pub enum ServerMessage { user_id: Option, /// Guest session ID (if guest). guest_session_id: Option, + /// Reason for disconnect. + reason: DisconnectReason, }, /// A member updated their position. diff --git a/crates/chattyness-shared/Cargo.toml b/crates/chattyness-shared/Cargo.toml index f32cda1..6480bbf 100644 --- a/crates/chattyness-shared/Cargo.toml +++ b/crates/chattyness-shared/Cargo.toml @@ -7,3 +7,8 @@ edition.workspace = true chattyness-error.workspace = true regex.workspace = true serde.workspace = true +toml.workspace = true + +[features] +default = [] +ssr = [] diff --git a/crates/chattyness-shared/src/config.rs b/crates/chattyness-shared/src/config.rs new file mode 100644 index 0000000..74fc396 --- /dev/null +++ b/crates/chattyness-shared/src/config.rs @@ -0,0 +1,88 @@ +//! Application configuration for chattyness. +//! +//! Provides TOML-based configuration with sensible defaults for WebSocket +//! timeouts and stale member cleanup. + +use serde::Deserialize; +use std::path::Path; + +use chattyness_error::AppError; + +/// Root configuration structure. +#[derive(Debug, Clone, Deserialize)] +#[serde(default)] +pub struct AppConfig { + /// WebSocket-related configuration. + pub websocket: WebSocketConfig, + /// Stale member cleanup configuration. + pub cleanup: CleanupConfig, +} + +/// WebSocket configuration. +#[derive(Debug, Clone, Deserialize)] +#[serde(default)] +pub struct WebSocketConfig { + /// Timeout for receiving messages from client before considering connection dead (seconds). + pub recv_timeout_secs: u64, + /// Interval for client to send ping to keep connection alive (seconds). + pub client_ping_interval_secs: u64, +} + +/// Cleanup configuration for stale instance members. +#[derive(Debug, Clone, Deserialize)] +#[serde(default)] +pub struct CleanupConfig { + /// Interval for running stale member cleanup job (seconds). + pub reap_interval_secs: u64, + /// Members with no activity for this duration are considered stale (seconds). + pub stale_threshold_secs: u64, + /// Clear all instance_members on server startup (recommended for single-server deployments). + pub clear_on_startup: bool, +} + +impl Default for AppConfig { + fn default() -> Self { + Self { + websocket: WebSocketConfig::default(), + cleanup: CleanupConfig::default(), + } + } +} + +impl Default for WebSocketConfig { + fn default() -> Self { + Self { + recv_timeout_secs: 45, + client_ping_interval_secs: 30, + } + } +} + +impl Default for CleanupConfig { + fn default() -> Self { + Self { + reap_interval_secs: 120, + stale_threshold_secs: 120, + clear_on_startup: true, + } + } +} + +impl AppConfig { + /// Load configuration from a TOML file. + /// + /// If no path is provided, returns default configuration. + pub fn load(path: Option<&Path>) -> Result { + match path { + Some(p) => { + let content = std::fs::read_to_string(p).map_err(|e| { + AppError::Internal(format!("Failed to read config file {:?}: {}", p, e)) + })?; + toml::from_str(&content).map_err(|e| { + AppError::Internal(format!("Failed to parse config file {:?}: {}", p, e)) + }) + } + None => Ok(Self::default()), + } + } +} diff --git a/crates/chattyness-shared/src/lib.rs b/crates/chattyness-shared/src/lib.rs index 0093236..b04cf0a 100644 --- a/crates/chattyness-shared/src/lib.rs +++ b/crates/chattyness-shared/src/lib.rs @@ -1,8 +1,10 @@ //! Shared utilities for chattyness. //! -//! This crate provides common validation functions and utilities -//! used across the application. +//! This crate provides common validation functions, configuration, +//! and utilities used across the application. +pub mod config; pub mod validation; +pub use config::*; pub use validation::*; diff --git a/crates/chattyness-user-ui/Cargo.toml b/crates/chattyness-user-ui/Cargo.toml index a5e2b01..d3cd07e 100644 --- a/crates/chattyness-user-ui/Cargo.toml +++ b/crates/chattyness-user-ui/Cargo.toml @@ -52,6 +52,7 @@ ssr = [ "leptos_router/ssr", "chattyness-db/ssr", "chattyness-error/ssr", + "chattyness-shared/ssr", "dep:chattyness-error", "dep:chattyness-shared", "dep:axum", diff --git a/crates/chattyness-user-ui/src/api/websocket.rs b/crates/chattyness-user-ui/src/api/websocket.rs index eea093d..17b62ba 100644 --- a/crates/chattyness-user-ui/src/api/websocket.rs +++ b/crates/chattyness-user-ui/src/api/websocket.rs @@ -4,7 +4,7 @@ use axum::{ extract::{ - ws::{Message, WebSocket, WebSocketUpgrade}, + ws::{CloseFrame, Message, WebSocket, WebSocketUpgrade}, FromRef, Path, State, }, response::IntoResponse, @@ -13,15 +13,20 @@ use dashmap::DashMap; use futures::{SinkExt, StreamExt}; use sqlx::PgPool; use std::sync::Arc; -use tokio::sync::broadcast; +use std::time::Duration; +use tokio::sync::{broadcast, mpsc}; use uuid::Uuid; use chattyness_db::{ models::{AvatarRenderData, ChannelMemberWithAvatar, EmotionState, User}, queries::{avatars, channel_members, loose_props, realms, scenes}, - ws_messages::{ClientMessage, ServerMessage}, + ws_messages::{ClientMessage, DisconnectReason, ServerMessage, WsConfig}, }; use chattyness_error::AppError; +use chattyness_shared::WebSocketConfig; + +/// Close code for scene change (custom code). +const SCENE_CHANGE_CLOSE_CODE: u16 = 4000; use crate::auth::AuthUser; @@ -71,12 +76,14 @@ pub async fn ws_handler( auth_result: Result, State(pool): State, State(ws_state): State>, + State(ws_config): State, ws: WebSocketUpgrade, ) -> Result where S: Send + Sync, PgPool: FromRef, Arc: FromRef, + WebSocketConfig: FromRef, { // Log auth result before checking #[cfg(debug_assertions)] @@ -117,7 +124,7 @@ where ); Ok(ws.on_upgrade(move |socket| { - handle_socket(socket, user, channel_id, realm.id, pool, ws_state) + handle_socket(socket, user, channel_id, realm.id, pool, ws_state, ws_config) })) } @@ -141,6 +148,7 @@ async fn handle_socket( realm_id: Uuid, pool: PgPool, ws_state: Arc, + ws_config: WebSocketConfig, ) { tracing::info!( "[WS] handle_socket started for user {} channel {} realm {}", @@ -217,10 +225,13 @@ async fn handle_socket( } }; - // Send welcome message + // Send welcome message with config let welcome = ServerMessage::Welcome { member: member.clone(), members, + config: WsConfig { + ping_interval_secs: ws_config.client_ping_interval_secs, + }, }; if let Ok(json) = serde_json::to_string(&welcome) { #[cfg(debug_assertions)] @@ -284,269 +295,339 @@ async fn handle_socket( // and pool for cleanup (leave_channel needs user_id match anyway) drop(conn); + // Channel for sending direct messages (Pong) to client + let (direct_tx, mut direct_rx) = mpsc::channel::(16); + + // Create recv timeout from config + let recv_timeout = Duration::from_secs(ws_config.recv_timeout_secs); + // Spawn task to handle incoming messages from client let recv_task = tokio::spawn(async move { - while let Some(Ok(msg)) = receiver.next().await { - if let Message::Text(text) = msg { - #[cfg(debug_assertions)] - tracing::debug!("[WS<-Client] {}", text); + let mut disconnect_reason = DisconnectReason::Graceful; - let Ok(client_msg) = serde_json::from_str::(&text) else { - continue; - }; + loop { + // Use timeout to detect connection loss + let msg_result = tokio::time::timeout(recv_timeout, receiver.next()).await; - match client_msg { - ClientMessage::UpdatePosition { x, y } => { - if let Err(e) = - channel_members::update_position(&mut *recv_conn, channel_id, user_id, x, y) - .await - { + match msg_result { + Ok(Some(Ok(msg))) => { + match msg { + Message::Text(text) => { #[cfg(debug_assertions)] - tracing::error!("[WS] Position update failed: {:?}", e); - continue; - } - let _ = tx.send(ServerMessage::PositionUpdated { - user_id: Some(user_id), - guest_session_id: None, - x, - y, - }); - } - ClientMessage::UpdateEmotion { emotion } => { - // Parse emotion name to EmotionState - let emotion_state = match emotion.parse::() { - Ok(e) => e, - Err(_) => { - #[cfg(debug_assertions)] - tracing::warn!("[WS] Invalid emotion name: {}", emotion); - continue; - } - }; - let emotion_layer = match avatars::set_emotion( - &mut *recv_conn, - user_id, - realm_id, - emotion_state, - ) - .await - { - Ok(layer) => layer, - Err(e) => { - #[cfg(debug_assertions)] - tracing::error!("[WS] Emotion update failed: {:?}", e); - continue; - } - }; - let _ = tx.send(ServerMessage::EmotionUpdated { - user_id: Some(user_id), - guest_session_id: None, - emotion, - emotion_layer, - }); - } - ClientMessage::Ping => { - // Respond with pong directly (not broadcast) - // This is handled in the send task via individual message - } - ClientMessage::SendChatMessage { content } => { - // Validate message - if content.is_empty() || content.len() > 500 { - continue; - } + tracing::debug!("[WS<-Client] {}", text); - // Get member's current position and emotion - let member_info = channel_members::get_channel_member( - &mut *recv_conn, - channel_id, - user_id, - realm_id, - ) - .await; - - if let Ok(Some(member)) = member_info { - // Convert emotion index to name - let emotion_name = EmotionState::from_index(member.current_emotion as u8) - .map(|e| e.to_string()) - .unwrap_or_else(|| "neutral".to_string()); - let msg = ServerMessage::ChatMessageReceived { - message_id: Uuid::new_v4(), - user_id: Some(user_id), - guest_session_id: None, - display_name: member.display_name.clone(), - content, - emotion: emotion_name, - x: member.position_x, - y: member.position_y, - timestamp: chrono::Utc::now().timestamp_millis(), + let Ok(client_msg) = serde_json::from_str::(&text) else { + continue; }; - let _ = tx.send(msg); - } - } - ClientMessage::DropProp { inventory_item_id } => { - // Get user's current position for random offset - let member_info = channel_members::get_channel_member( - &mut *recv_conn, - channel_id, - user_id, - realm_id, - ) - .await; - if let Ok(Some(member)) = member_info { - // Generate random offset (within ~50 pixels) - let offset_x = (rand::random::() - 0.5) * 100.0; - let offset_y = (rand::random::() - 0.5) * 100.0; - let pos_x = member.position_x + offset_x; - let pos_y = member.position_y + offset_y; - - match loose_props::drop_prop_to_canvas( - &mut *recv_conn, - inventory_item_id, - user_id, - channel_id, - pos_x, - pos_y, - ) - .await - { - Ok(prop) => { - #[cfg(debug_assertions)] - tracing::debug!( - "[WS] User {} dropped prop {} at ({}, {})", - user_id, - prop.id, - pos_x, - pos_y - ); - let _ = tx.send(ServerMessage::PropDropped { prop }); + match client_msg { + ClientMessage::UpdatePosition { x, y } => { + if let Err(e) = + channel_members::update_position(&mut *recv_conn, channel_id, user_id, x, y) + .await + { + #[cfg(debug_assertions)] + tracing::error!("[WS] Position update failed: {:?}", e); + continue; + } + let _ = tx.send(ServerMessage::PositionUpdated { + user_id: Some(user_id), + guest_session_id: None, + x, + y, + }); } - Err(e) => { - tracing::error!("[WS] Drop prop failed: {:?}", e); - let (code, message) = match &e { - chattyness_error::AppError::Forbidden(msg) => { - ("PROP_NOT_DROPPABLE".to_string(), msg.clone()) + ClientMessage::UpdateEmotion { emotion } => { + // Parse emotion name to EmotionState + let emotion_state = match emotion.parse::() { + Ok(e) => e, + Err(_) => { + #[cfg(debug_assertions)] + tracing::warn!("[WS] Invalid emotion name: {}", emotion); + continue; } - chattyness_error::AppError::NotFound(msg) => { - ("PROP_NOT_FOUND".to_string(), msg.clone()) - } - _ => ("DROP_FAILED".to_string(), format!("{:?}", e)), }; - let _ = tx.send(ServerMessage::Error { code, message }); + let emotion_layer = match avatars::set_emotion( + &mut *recv_conn, + user_id, + realm_id, + emotion_state, + ) + .await + { + Ok(layer) => layer, + Err(e) => { + #[cfg(debug_assertions)] + tracing::error!("[WS] Emotion update failed: {:?}", e); + continue; + } + }; + let _ = tx.send(ServerMessage::EmotionUpdated { + user_id: Some(user_id), + guest_session_id: None, + emotion, + emotion_layer, + }); + } + ClientMessage::Ping => { + // Update last_moved_at to keep member alive for cleanup + let _ = channel_members::touch_member(&mut *recv_conn, channel_id, user_id).await; + // Respond with pong directly (not broadcast) + let _ = direct_tx.send(ServerMessage::Pong).await; + } + ClientMessage::SendChatMessage { content } => { + // Validate message + if content.is_empty() || content.len() > 500 { + continue; + } + + // Get member's current position and emotion + let member_info = channel_members::get_channel_member( + &mut *recv_conn, + channel_id, + user_id, + realm_id, + ) + .await; + + if let Ok(Some(member)) = member_info { + // Convert emotion index to name + let emotion_name = EmotionState::from_index(member.current_emotion as u8) + .map(|e| e.to_string()) + .unwrap_or_else(|| "neutral".to_string()); + let msg = ServerMessage::ChatMessageReceived { + message_id: Uuid::new_v4(), + user_id: Some(user_id), + guest_session_id: None, + display_name: member.display_name.clone(), + content, + emotion: emotion_name, + x: member.position_x, + y: member.position_y, + timestamp: chrono::Utc::now().timestamp_millis(), + }; + let _ = tx.send(msg); + } + } + ClientMessage::DropProp { inventory_item_id } => { + // Get user's current position for random offset + let member_info = channel_members::get_channel_member( + &mut *recv_conn, + channel_id, + user_id, + realm_id, + ) + .await; + + if let Ok(Some(member)) = member_info { + // Generate random offset (within ~50 pixels) + let offset_x = (rand::random::() - 0.5) * 100.0; + let offset_y = (rand::random::() - 0.5) * 100.0; + let pos_x = member.position_x + offset_x; + let pos_y = member.position_y + offset_y; + + match loose_props::drop_prop_to_canvas( + &mut *recv_conn, + inventory_item_id, + user_id, + channel_id, + pos_x, + pos_y, + ) + .await + { + Ok(prop) => { + #[cfg(debug_assertions)] + tracing::debug!( + "[WS] User {} dropped prop {} at ({}, {})", + user_id, + prop.id, + pos_x, + pos_y + ); + let _ = tx.send(ServerMessage::PropDropped { prop }); + } + Err(e) => { + tracing::error!("[WS] Drop prop failed: {:?}", e); + let (code, message) = match &e { + chattyness_error::AppError::Forbidden(msg) => { + ("PROP_NOT_DROPPABLE".to_string(), msg.clone()) + } + chattyness_error::AppError::NotFound(msg) => { + ("PROP_NOT_FOUND".to_string(), msg.clone()) + } + _ => ("DROP_FAILED".to_string(), format!("{:?}", e)), + }; + let _ = tx.send(ServerMessage::Error { code, message }); + } + } + } + } + ClientMessage::PickUpProp { loose_prop_id } => { + match loose_props::pick_up_loose_prop( + &mut *recv_conn, + loose_prop_id, + user_id, + ) + .await + { + Ok(_inventory_item) => { + #[cfg(debug_assertions)] + tracing::debug!( + "[WS] User {} picked up prop {}", + user_id, + loose_prop_id + ); + let _ = tx.send(ServerMessage::PropPickedUp { + prop_id: loose_prop_id, + picked_up_by_user_id: Some(user_id), + picked_up_by_guest_id: None, + }); + } + Err(e) => { + tracing::error!("[WS] Pick up prop failed: {:?}", e); + let _ = tx.send(ServerMessage::Error { + code: "PICKUP_FAILED".to_string(), + message: format!("{:?}", e), + }); + } + } + } + ClientMessage::SyncAvatar => { + // Fetch current avatar from database and broadcast to channel + match avatars::get_avatar_with_paths_conn( + &mut *recv_conn, + user_id, + realm_id, + ) + .await + { + Ok(Some(avatar_with_paths)) => { + let render_data = avatar_with_paths.to_render_data(); + #[cfg(debug_assertions)] + tracing::debug!( + "[WS] User {} syncing avatar to channel", + user_id + ); + let _ = tx.send(ServerMessage::AvatarUpdated { + user_id: Some(user_id), + guest_session_id: None, + avatar: render_data, + }); + } + Ok(None) => { + #[cfg(debug_assertions)] + tracing::warn!("[WS] No avatar found for user {} to sync", user_id); + } + Err(e) => { + tracing::error!("[WS] Avatar sync failed: {:?}", e); + } + } } } } - } - ClientMessage::PickUpProp { loose_prop_id } => { - match loose_props::pick_up_loose_prop( - &mut *recv_conn, - loose_prop_id, - user_id, - ) - .await - { - Ok(_inventory_item) => { - #[cfg(debug_assertions)] - tracing::debug!( - "[WS] User {} picked up prop {}", - user_id, - loose_prop_id - ); - let _ = tx.send(ServerMessage::PropPickedUp { - prop_id: loose_prop_id, - picked_up_by_user_id: Some(user_id), - picked_up_by_guest_id: None, - }); - } - Err(e) => { - tracing::error!("[WS] Pick up prop failed: {:?}", e); - let _ = tx.send(ServerMessage::Error { - code: "PICKUP_FAILED".to_string(), - message: format!("{:?}", e), - }); + Message::Close(close_frame) => { + // Check close code for scene change + if let Some(CloseFrame { code, .. }) = close_frame { + if code == SCENE_CHANGE_CLOSE_CODE { + disconnect_reason = DisconnectReason::SceneChange; + } else { + disconnect_reason = DisconnectReason::Graceful; + } } + break; } - } - ClientMessage::SyncAvatar => { - // Fetch current avatar from database and broadcast to channel - match avatars::get_avatar_with_paths_conn( - &mut *recv_conn, - user_id, - realm_id, - ) - .await - { - Ok(Some(avatar_with_paths)) => { - let render_data = avatar_with_paths.to_render_data(); - #[cfg(debug_assertions)] - tracing::debug!( - "[WS] User {} syncing avatar to channel", - user_id - ); - let _ = tx.send(ServerMessage::AvatarUpdated { - user_id: Some(user_id), - guest_session_id: None, - avatar: render_data, - }); - } - Ok(None) => { - #[cfg(debug_assertions)] - tracing::warn!("[WS] No avatar found for user {} to sync", user_id); - } - Err(e) => { - tracing::error!("[WS] Avatar sync failed: {:?}", e); - } + _ => { + // Ignore binary, ping, pong messages } } } - } - } - // Return the connection so we can use it for cleanup - recv_conn - }); - - // Spawn task to forward broadcasts to this client - let send_task = tokio::spawn(async move { - while let Ok(msg) = rx.recv().await { - if let Ok(json) = serde_json::to_string(&msg) { - #[cfg(debug_assertions)] - tracing::debug!("[WS->Client] {}", json); - if sender.send(Message::Text(json.into())).await.is_err() { + Ok(Some(Err(e))) => { + // WebSocket error + tracing::warn!("[WS] Connection error: {:?}", e); + disconnect_reason = DisconnectReason::Timeout; + break; + } + Ok(None) => { + // Stream ended gracefully + break; + } + Err(_) => { + // Timeout elapsed - connection likely lost + tracing::info!("[WS] Connection timeout for user {}", user_id); + disconnect_reason = DisconnectReason::Timeout; break; } } } + // Return the connection and disconnect reason for cleanup + (recv_conn, disconnect_reason) }); - // Wait for either task to complete - tokio::select! { + // Spawn task to forward broadcasts and direct messages to this client + let send_task = tokio::spawn(async move { + loop { + tokio::select! { + // Handle broadcast messages + Ok(msg) = rx.recv() => { + if let Ok(json) = serde_json::to_string(&msg) { + #[cfg(debug_assertions)] + tracing::debug!("[WS->Client] {}", json); + if sender.send(Message::Text(json.into())).await.is_err() { + break; + } + } + } + // Handle direct messages (Pong) + Some(msg) = direct_rx.recv() => { + if let Ok(json) = serde_json::to_string(&msg) { + #[cfg(debug_assertions)] + tracing::debug!("[WS->Client] {}", json); + if sender.send(Message::Text(json.into())).await.is_err() { + break; + } + } + } + else => break + } + } + }); + + // Wait for either task to complete, track disconnect reason + let disconnect_reason = tokio::select! { recv_result = recv_task => { - // recv_task finished, get connection back for cleanup - if let Ok(mut cleanup_conn) = recv_result { + // recv_task finished, get connection and reason back for cleanup + if let Ok((mut cleanup_conn, reason)) = recv_result { let _ = channel_members::leave_channel(&mut *cleanup_conn, channel_id, user_id).await; + reason } else { // Task panicked, use pool (RLS may fail but try anyway) let _ = channel_members::leave_channel(&pool, channel_id, user_id).await; + DisconnectReason::Timeout } } _ = send_task => { - // send_task finished first, need to acquire a new connection for cleanup + // send_task finished first (likely client disconnect), need to acquire a new connection for cleanup if let Ok(mut cleanup_conn) = pool.acquire().await { let _ = set_rls_user_id(&mut cleanup_conn, user_id).await; let _ = channel_members::leave_channel(&mut *cleanup_conn, channel_id, user_id).await; } + DisconnectReason::Graceful } - } + }; tracing::info!( - "[WS] User {} disconnected from channel {}", + "[WS] User {} disconnected from channel {} (reason: {:?})", user_id, - channel_id + channel_id, + disconnect_reason ); - // Broadcast departure + // Broadcast departure with reason let _ = channel_state.tx.send(ServerMessage::MemberLeft { user_id: Some(user_id), guest_session_id: None, + reason: disconnect_reason, }); } diff --git a/crates/chattyness-user-ui/src/app.rs b/crates/chattyness-user-ui/src/app.rs index f714246..74bdb41 100644 --- a/crates/chattyness-user-ui/src/app.rs +++ b/crates/chattyness-user-ui/src/app.rs @@ -12,6 +12,9 @@ use std::sync::Arc; #[cfg(feature = "ssr")] use crate::api::WebSocketState; +#[cfg(feature = "ssr")] +use chattyness_shared::WebSocketConfig; + /// Application state for the public app. #[cfg(feature = "ssr")] #[derive(Clone)] @@ -19,6 +22,7 @@ pub struct AppState { pub pool: sqlx::PgPool, pub leptos_options: LeptosOptions, pub ws_state: Arc, + pub ws_config: WebSocketConfig, } #[cfg(feature = "ssr")] @@ -42,6 +46,13 @@ impl axum::extract::FromRef for Arc { } } +#[cfg(feature = "ssr")] +impl axum::extract::FromRef for WebSocketConfig { + fn from_ref(state: &AppState) -> Self { + state.ws_config.clone() + } +} + /// Shell component for SSR. pub fn shell(options: LeptosOptions) -> impl IntoView { view! { diff --git a/crates/chattyness-user-ui/src/components/avatar_canvas.rs b/crates/chattyness-user-ui/src/components/avatar_canvas.rs index 14809ae..c556d8a 100644 --- a/crates/chattyness-user-ui/src/components/avatar_canvas.rs +++ b/crates/chattyness-user-ui/src/components/avatar_canvas.rs @@ -115,6 +115,9 @@ pub fn AvatarCanvas( /// Text size multiplier for display names, chat bubbles, and badges. #[prop(default = 1.0)] text_em_size: f64, + /// Opacity for fade-out animation (0.0 to 1.0, default 1.0). + #[prop(default = 1.0)] + opacity: f64, ) -> impl IntoView { let canvas_ref = NodeRef::::new(); @@ -171,19 +174,24 @@ pub fn AvatarCanvas( let adjusted_y = canvas_y - bubble_extra; // CSS positioning via transform (GPU-accelerated) + // Disable pointer-events when fading (opacity < 1.0) so fading avatars aren't clickable + let pointer_events = if opacity < 1.0 { "none" } else { "auto" }; let style = format!( "position: absolute; \ left: 0; top: 0; \ transform: translate({}px, {}px); \ z-index: {}; \ - pointer-events: auto; \ + pointer-events: {}; \ width: {}px; \ - height: {}px;", + height: {}px; \ + opacity: {};", canvas_x - (canvas_width - avatar_size) / 2.0, adjusted_y, z_index, + pointer_events, canvas_width, - canvas_height + canvas_height, + opacity ); // Store references for the effect diff --git a/crates/chattyness-user-ui/src/components/scene_viewer.rs b/crates/chattyness-user-ui/src/components/scene_viewer.rs index cda3f03..0e3cc24 100644 --- a/crates/chattyness-user-ui/src/components/scene_viewer.rs +++ b/crates/chattyness-user-ui/src/components/scene_viewer.rs @@ -20,6 +20,7 @@ use super::chat_types::ActiveBubble; use super::settings::{ calculate_min_zoom, ViewerSettings, BASE_PROP_SIZE, REFERENCE_HEIGHT, REFERENCE_WIDTH, }; +use super::ws_client::FadingMember; use crate::utils::parse_bounds_dimensions; /// Scene viewer component for displaying a realm scene with avatars. @@ -49,6 +50,9 @@ pub fn RealmSceneViewer( /// Callback for zoom changes (from mouse wheel). Receives zoom delta. #[prop(optional)] on_zoom_change: Option>, + /// Members that are fading out after timeout disconnect. + #[prop(optional, into)] + fading_members: Option>>, ) -> impl IntoView { // Use default settings if none provided let settings = settings.unwrap_or_else(|| Signal::derive(ViewerSettings::default)); @@ -774,7 +778,8 @@ pub fn RealmSceneViewer( let ps = prop_size.get(); let te = text_em_size.get(); - sorted_members.get() + // Render active members + let mut views: Vec<_> = sorted_members.get() .into_iter() .enumerate() .map(|(idx, member)| { @@ -795,7 +800,39 @@ pub fn RealmSceneViewer( /> } }) - .collect_view() + .collect(); + + // Render fading members with calculated opacity + if let Some(fading_signal) = fading_members { + #[cfg(feature = "hydrate")] + let now = js_sys::Date::now() as i64; + #[cfg(not(feature = "hydrate"))] + let now = 0i64; + + for fading in fading_signal.get() { + let elapsed = now - fading.fade_start; + if elapsed < fading.fade_duration { + let opacity = 1.0 - (elapsed as f64 / fading.fade_duration as f64); + let opacity = opacity.max(0.0).min(1.0); + views.push(view! { + + }); + } + } + } + + views.into_iter().collect_view() }} // Click overlay - captures clicks for movement and hit-testing diff --git a/crates/chattyness-user-ui/src/components/ws_client.rs b/crates/chattyness-user-ui/src/components/ws_client.rs index c59994c..711ac9d 100644 --- a/crates/chattyness-user-ui/src/components/ws_client.rs +++ b/crates/chattyness-user-ui/src/components/ws_client.rs @@ -6,11 +6,32 @@ use leptos::prelude::*; use leptos::reactive::owner::LocalStorage; -use chattyness_db::models::{ChannelMemberWithAvatar, EmotionState, LooseProp}; -use chattyness_db::ws_messages::{ClientMessage, ServerMessage}; +use chattyness_db::models::{ChannelMemberWithAvatar, LooseProp}; +#[cfg(feature = "hydrate")] +use chattyness_db::models::EmotionState; +use chattyness_db::ws_messages::ClientMessage; +#[cfg(feature = "hydrate")] +use chattyness_db::ws_messages::{DisconnectReason, ServerMessage, WsConfig}; use super::chat_types::ChatMessage; +/// Close code for scene change (must match server constant). +pub const SCENE_CHANGE_CLOSE_CODE: u16 = 4000; + +/// Duration for fade-out animation in milliseconds. +pub const FADE_DURATION_MS: i64 = 5000; + +/// A member that is currently fading out after a timeout disconnect. +#[derive(Clone, Debug)] +pub struct FadingMember { + /// The member data. + pub member: ChannelMemberWithAvatar, + /// Timestamp when the fade started (milliseconds since epoch). + pub fade_start: i64, + /// Duration of the fade in milliseconds. + pub fade_duration: i64, +} + /// WebSocket connection state. #[derive(Clone, Copy, PartialEq, Eq, Debug)] pub enum WsState { @@ -44,6 +65,7 @@ pub fn use_channel_websocket( on_loose_props_sync: Callback>, on_prop_dropped: Callback, on_prop_picked_up: Callback, + on_member_fading: Callback, ) -> (Signal, WsSenderStorage) { use std::cell::RefCell; use std::rc::Rc; @@ -139,6 +161,11 @@ pub fn use_channel_websocket( let on_loose_props_sync_clone = on_loose_props_sync.clone(); let on_prop_dropped_clone = on_prop_dropped.clone(); let on_prop_picked_up_clone = on_prop_picked_up.clone(); + let on_member_fading_clone = on_member_fading.clone(); + // For starting heartbeat on Welcome + let ws_ref_for_heartbeat = ws_ref.clone(); + let heartbeat_started: Rc> = Rc::new(RefCell::new(false)); + let heartbeat_started_clone = heartbeat_started.clone(); let onmessage = Closure::wrap(Box::new(move |e: MessageEvent| { if let Ok(text) = e.data().dyn_into::() { let text: String = text.into(); @@ -146,6 +173,28 @@ pub fn use_channel_websocket( web_sys::console::log_1(&format!("[WS<-Server] {}", text).into()); if let Ok(msg) = serde_json::from_str::(&text) { + // Check for Welcome message to start heartbeat with server-provided config + if let ServerMessage::Welcome { ref config, .. } = msg { + if !*heartbeat_started_clone.borrow() { + *heartbeat_started_clone.borrow_mut() = true; + let ping_interval_ms = config.ping_interval_secs * 1000; + let ws_ref_ping = ws_ref_for_heartbeat.clone(); + #[cfg(debug_assertions)] + web_sys::console::log_1( + &format!("[WS] Starting heartbeat with interval {}ms", ping_interval_ms).into(), + ); + let heartbeat = gloo_timers::callback::Interval::new(ping_interval_ms as u32, move || { + if let Some(ws) = ws_ref_ping.borrow().as_ref() { + if ws.ready_state() == WebSocket::OPEN { + if let Ok(json) = serde_json::to_string(&ClientMessage::Ping) { + let _ = ws.send_with_str(&json); + } + } + } + }); + std::mem::forget(heartbeat); + } + } handle_server_message( msg, &members_for_msg, @@ -154,6 +203,7 @@ pub fn use_channel_websocket( &on_loose_props_sync_clone, &on_prop_dropped_clone, &on_prop_picked_up_clone, + &on_member_fading_clone, ); } } @@ -199,6 +249,7 @@ fn handle_server_message( on_loose_props_sync: &Callback>, on_prop_dropped: &Callback, on_prop_picked_up: &Callback, + on_member_fading: &Callback, ) { let mut members_vec = members.borrow_mut(); @@ -206,6 +257,7 @@ fn handle_server_message( ServerMessage::Welcome { member: _, members: initial_members, + config: _, // Config is handled in the caller for heartbeat setup } => { *members_vec = initial_members; on_update.run(members_vec.clone()); @@ -222,11 +274,31 @@ fn handle_server_message( ServerMessage::MemberLeft { user_id, guest_session_id, + reason, } => { + // Find the member before removing + let leaving_member = members_vec + .iter() + .find(|m| m.member.user_id == user_id && m.member.guest_session_id == guest_session_id) + .cloned(); + + // Always remove from active members list members_vec.retain(|m| { m.member.user_id != user_id || m.member.guest_session_id != guest_session_id }); on_update.run(members_vec.clone()); + + // For timeout disconnects, trigger fading animation + if reason == DisconnectReason::Timeout { + if let Some(member) = leaving_member { + let fading = FadingMember { + member, + fade_start: js_sys::Date::now() as i64, + fade_duration: FADE_DURATION_MS, + }; + on_member_fading.run(fading); + } + } } ServerMessage::PositionUpdated { user_id, @@ -333,6 +405,7 @@ pub fn use_channel_websocket( _on_loose_props_sync: Callback>, _on_prop_dropped: Callback, _on_prop_picked_up: Callback, + _on_member_fading: Callback, ) -> (Signal, WsSenderStorage) { let (ws_state, _) = signal(WsState::Disconnected); let sender: WsSenderStorage = StoredValue::new_local(None); diff --git a/crates/chattyness-user-ui/src/pages/realm.rs b/crates/chattyness-user-ui/src/pages/realm.rs index 54b508d..4df3712 100644 --- a/crates/chattyness-user-ui/src/pages/realm.rs +++ b/crates/chattyness-user-ui/src/pages/realm.rs @@ -12,13 +12,15 @@ use leptos_router::hooks::use_params_map; use uuid::Uuid; use crate::components::{ - ActiveBubble, AvatarEditorPopup, Card, ChatInput, ChatMessage, EmotionKeybindings, + ActiveBubble, AvatarEditorPopup, Card, ChatInput, EmotionKeybindings, FadingMember, InventoryPopup, KeybindingsPopup, MessageLog, RealmHeader, RealmSceneViewer, SettingsPopup, - ViewerSettings, DEFAULT_BUBBLE_TIMEOUT_MS, + ViewerSettings, }; #[cfg(feature = "hydrate")] -use crate::components::use_channel_websocket; -use crate::utils::{parse_bounds_dimensions, LocalStoragePersist}; +use crate::components::{use_channel_websocket, ChatMessage, DEFAULT_BUBBLE_TIMEOUT_MS, FADE_DURATION_MS}; +use crate::utils::LocalStoragePersist; +#[cfg(feature = "hydrate")] +use crate::utils::parse_bounds_dimensions; use chattyness_db::models::{ AvatarWithPaths, ChannelMemberWithAvatar, EmotionAvailability, LooseProp, RealmRole, RealmWithUserRole, Scene, @@ -82,6 +84,9 @@ pub fn RealmPage() -> impl IntoView { // Loose props state let (loose_props, set_loose_props) = signal(Vec::::new()); + // Fading members state (members that are fading out after timeout disconnect) + let (fading_members, set_fading_members) = signal(Vec::::new()); + // Track user's current position for saving on beforeunload let (current_position, set_current_position) = signal((400.0_f64, 300.0_f64)); @@ -173,6 +178,15 @@ pub fn RealmPage() -> impl IntoView { // WebSocket connection for real-time updates #[cfg(feature = "hydrate")] let on_members_update = Callback::new(move |new_members: Vec| { + // When members are updated (including rejoins), remove any matching fading members + set_fading_members.update(|fading| { + fading.retain(|f| { + !new_members.iter().any(|m| { + m.member.user_id == f.member.member.user_id + && m.member.guest_session_id == f.member.member.guest_session_id + }) + }); + }); set_members.set(new_members); }); @@ -217,6 +231,19 @@ pub fn RealmPage() -> impl IntoView { }); }); + // Callback when a member starts fading (timeout disconnect) + #[cfg(feature = "hydrate")] + let on_member_fading = Callback::new(move |fading: FadingMember| { + set_fading_members.update(|members| { + // Remove any existing entry for this user (shouldn't happen, but be safe) + members.retain(|m| { + m.member.member.user_id != fading.member.member.user_id + || m.member.member.guest_session_id != fading.member.member.guest_session_id + }); + members.push(fading); + }); + }); + #[cfg(feature = "hydrate")] let (_ws_state, ws_sender) = use_channel_websocket( slug, @@ -226,6 +253,7 @@ pub fn RealmPage() -> impl IntoView { on_loose_props_sync, on_prop_dropped, on_prop_picked_up, + on_member_fading, ); // Set channel ID and scene dimensions when scene loads @@ -246,16 +274,21 @@ pub fn RealmPage() -> impl IntoView { }); } - // Cleanup expired speech bubbles every 5 seconds + // Cleanup expired speech bubbles and fading members every second #[cfg(feature = "hydrate")] { use gloo_timers::callback::Interval; - let cleanup_interval = Interval::new(5000, move || { + let cleanup_interval = Interval::new(1000, move || { let now = js_sys::Date::now() as i64; + // Clean up expired bubbles set_active_bubbles.update(|bubbles| { bubbles.retain(|_, bubble| bubble.expires_at > now); }); + // Clean up completed fading members + set_fading_members.update(|members| { + members.retain(|m| now - m.fade_start < FADE_DURATION_MS); + }); }); // Keep interval alive std::mem::forget(cleanup_interval); @@ -638,6 +671,7 @@ pub fn RealmPage() -> impl IntoView { s.save(); }); }) + fading_members=Signal::derive(move || fading_members.get()) />
threshold_seconds); + GET DIAGNOSTICS deleted_count = ROW_COUNT; + RETURN deleted_count; +END; +$$ LANGUAGE plpgsql SECURITY DEFINER; + +COMMENT ON FUNCTION scene.clear_stale_instance_members(DOUBLE PRECISION) IS + 'Clears stale instance members older than threshold. Bypasses RLS.'; + +-- Grant execute to chattyness_app +GRANT EXECUTE ON FUNCTION scene.clear_all_instance_members() TO chattyness_app; +GRANT EXECUTE ON FUNCTION scene.clear_stale_instance_members(DOUBLE PRECISION) TO chattyness_app; + COMMIT;