From d9d3dd2de9ae904a952270d148301d66f899c327 Mon Sep 17 00:00:00 2001 From: TolaMironcenko Date: Sun, 23 Nov 2025 19:01:04 +0700 Subject: [PATCH] vorking with chats --- src/main.rs | 685 +++++++++++++++++++++++++++++++++---- web/index.html | 899 +++++++++++++++++++++++++------------------------ 2 files changed, 1081 insertions(+), 503 deletions(-) diff --git a/src/main.rs b/src/main.rs index fa90dba..9c816a1 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,10 +1,10 @@ use axum::{ extract::{ ws::{Message, WebSocket, WebSocketUpgrade}, - State, Json, + State, Json, Query, }, response::Response, - routing::{get, post}, + routing::{get, post, delete}, Router, }; use serde::{Deserialize, Serialize}; @@ -82,6 +82,38 @@ struct ValidateTokenResponse { message: Option, } +#[derive(Debug, Serialize)] +struct ChatPartner { + username: String, + phone: String, + last_message: Option, + last_message_time: Option, + unread_count: i64, +} + +#[derive(Debug, Deserialize)] +struct SearchQuery { + q: String, +} + +#[derive(Debug, Deserialize)] +struct MarkReadRequest { + from_user: String, + to_user: String, +} + +#[derive(Debug, Deserialize)] +struct DeleteMessageRequest { + message_id: String, + for_everyone: bool, +} + +#[derive(Debug, Serialize)] +struct DeleteMessageResponse { + success: bool, + message: String, +} + #[tokio::main] async fn main() -> Result<(), Box> { // Initialize logging @@ -102,11 +134,16 @@ async fn main() -> Result<(), Box> { .route("/api/register", post(register_handler)) .route("/api/login", post(login_handler)) .route("/api/validate-token", post(validate_token_handler)) + .route("/api/chats", get(get_chats_handler)) + .route("/api/search-users", get(search_users_handler)) + .route("/api/debug-messages", get(debug_messages_handler)) + .route("/api/mark-read", post(mark_read_handler)) + .route("/api/delete-message", post(delete_message_handler)) .route("/api/profile", get(profile_handler)) .nest_service("/", ServeDir::new("web")) .with_state(state); - let addr = SocketAddr::from(([127, 0, 0, 1], 3333)); + let addr = SocketAddr::from(([0, 0, 0, 0], 3333)); println!("Server running on http://{}", addr); let listener = tokio::net::TcpListener::bind(addr).await?; @@ -142,7 +179,11 @@ async fn init_db() -> Result> { to_user TEXT NOT NULL, content TEXT NOT NULL, timestamp INTEGER NOT NULL, - message_type TEXT NOT NULL + message_type TEXT NOT NULL, + read BOOLEAN DEFAULT FALSE, + deleted_for_sender BOOLEAN DEFAULT FALSE, + deleted_for_receiver BOOLEAN DEFAULT FALSE, + deleted_for_everyone BOOLEAN DEFAULT FALSE ) "#, ) @@ -395,9 +436,510 @@ async fn validate_token_handler( } } +// Debug endpoint to see all messages +async fn debug_messages_handler( + State(state): State>, + headers: axum::http::HeaderMap, +) -> Json { + // Extract token from header + let token = if let Some(auth_header) = headers.get("authorization") { + auth_header.to_str().unwrap_or("").replace("Bearer ", "") + } else { + return Json(serde_json::json!({"error": "No token"})); + }; + + // Get current user from token + let username = match verify_token(&state.db_pool, &token).await { + Ok(Some(user)) => user, + _ => return Json(serde_json::json!({"error": "Invalid token"})), + }; + + // Get all messages for this user + let messages = sqlx::query( + r#" + SELECT id, from_user, to_user, content, timestamp, message_type, read, + deleted_for_sender, deleted_for_receiver, deleted_for_everyone + FROM messages + WHERE from_user = ? OR to_user = ? + ORDER BY timestamp DESC + LIMIT 100 + "# + ) + .bind(&username) + .bind(&username) + .fetch_all(&state.db_pool) + .await; + + // Get chat partners count + let partners_result = sqlx::query( + r#" + SELECT COUNT(DISTINCT + CASE + WHEN from_user = ? THEN to_user + ELSE from_user + END + ) as partner_count + FROM messages + WHERE from_user = ? OR to_user = ? + "# + ) + .bind(&username) + .bind(&username) + .bind(&username) + .fetch_one(&state.db_pool) + .await; + + match messages { + Ok(rows) => { + let mut result = Vec::new(); + for row in rows { + result.push(serde_json::json!({ + "id": row.get::("id"), + "from_user": row.get::("from_user"), + "to_user": row.get::("to_user"), + "content": row.get::("content"), + "timestamp": row.get::("timestamp"), + "message_type": row.get::("message_type"), + "read": row.get::("read"), + "deleted_for_sender": row.get::("deleted_for_sender"), + "deleted_for_receiver": row.get::("deleted_for_receiver"), + "deleted_for_everyone": row.get::("deleted_for_everyone"), + "is_current_user_from": row.get::("from_user") == username, + "is_current_user_to": row.get::("to_user") == username, + })); + } + + let partner_count = match partners_result { + Ok(row) => row.get::("partner_count"), + Err(_) => 0, + }; + + Json(serde_json::json!({ + "user": username, + "total_messages": result.len(), + "chat_partners_count": partner_count, + "messages": result + })) + } + Err(e) => { + Json(serde_json::json!({"error": format!("Database error: {}", e)})) + } + } +} + +// УПРОЩЕННАЯ версия получения чатов +async fn get_chats_handler( + State(state): State>, + headers: axum::http::HeaderMap, +) -> Json> { + // Extract token from header + let token = if let Some(auth_header) = headers.get("authorization") { + auth_header.to_str().unwrap_or("").replace("Bearer ", "") + } else { + return Json(Vec::new()); + }; + + // Get current user from token + let username = match verify_token(&state.db_pool, &token).await { + Ok(Some(user)) => user, + _ => return Json(Vec::new()), + }; + + println!("🔍 Fetching chats for user: {}", username); + + // ПРОСТОЙ И ПРЯМОЙ ЗАПРОС: находим всех пользователей, с которыми есть сообщения + let chat_partners = sqlx::query( + r#" + -- Находим всех уникальных собеседников + SELECT DISTINCT + CASE + WHEN from_user = ? THEN to_user + ELSE from_user + END as partner_username + FROM messages + WHERE (from_user = ? OR to_user = ?) + AND deleted_for_everyone = FALSE + AND ((from_user = ? AND deleted_for_sender = FALSE) OR (to_user = ? AND deleted_for_receiver = FALSE)) + "# + ) + .bind(&username) + .bind(&username) + .bind(&username) + .bind(&username) + .bind(&username) + .fetch_all(&state.db_pool) + .await; + + let partner_usernames = match chat_partners { + Ok(rows) => { + let partners: Vec = rows.iter().map(|row| row.get::("partner_username")).collect(); + println!("🤝 Found partner usernames: {:?}", partners); + partners + } + Err(e) => { + eprintln!("❌ Error fetching partner usernames: {}", e); + Vec::new() + } + }; + + if partner_usernames.is_empty() { + println!("❌ No partners found for user {}", username); + + // Проверим, есть ли вообще сообщения у пользователя + let message_count = sqlx::query( + "SELECT COUNT(*) as count FROM messages WHERE (from_user = ? OR to_user = ?) AND deleted_for_everyone = FALSE" + ) + .bind(&username) + .bind(&username) + .fetch_one(&state.db_pool) + .await; + + match message_count { + Ok(row) => { + let count: i64 = row.get("count"); + println!("📊 User {} has {} total messages in database", username, count); + } + Err(e) => { + eprintln!("❌ Error counting messages: {}", e); + } + } + + return Json(Vec::new()); + } + + // Теперь получаем детальную информацию о каждом собеседнике + let mut chats = Vec::new(); + + for partner_username in partner_usernames { + println!("📞 Getting details for partner: {}", partner_username); + + // Получаем информацию о пользователе + let user_result = sqlx::query( + "SELECT username, phone FROM users WHERE username = ?" + ) + .bind(&partner_username) + .fetch_optional(&state.db_pool) + .await; + + match user_result { + Ok(Some(user_row)) => { + let partner_name: String = user_row.get("username"); + let phone: String = user_row.get("phone"); + + // Получаем последнее сообщение + let last_message_result = sqlx::query( + r#" + SELECT content, timestamp + FROM messages + WHERE ((from_user = ? AND to_user = ?) OR (from_user = ? AND to_user = ?)) + AND deleted_for_everyone = FALSE + AND ((from_user = ? AND deleted_for_sender = FALSE) OR (to_user = ? AND deleted_for_receiver = FALSE)) + ORDER BY timestamp DESC + LIMIT 1 + "# + ) + .bind(&username) // текущий пользователь + .bind(&partner_name) // собеседник + .bind(&partner_name) // собеседник + .bind(&username) // текущий пользователь + .bind(&username) // для проверки deleted_for_sender + .bind(&username) // для проверки deleted_for_receiver + .fetch_optional(&state.db_pool) + .await; + + let (last_message, last_message_time) = match last_message_result { + Ok(Some(row)) => { + let content: String = row.get("content"); + let timestamp: i64 = row.get("timestamp"); + println!("💬 Last message with {}: '{}' at {}", partner_name, content, timestamp); + (Some(content), Some(timestamp)) + } + Ok(None) => { + println!("❌ No last message found for chat with {}", partner_name); + (None, None) + } + Err(e) => { + eprintln!("❌ Error fetching last message: {}", e); + (None, None) + } + }; + + // Получаем количество непрочитанных сообщений + let unread_count_result = sqlx::query( + "SELECT COUNT(*) as count FROM messages WHERE from_user = ? AND to_user = ? AND read = FALSE AND deleted_for_everyone = FALSE AND deleted_for_receiver = FALSE" + ) + .bind(&partner_name) + .bind(&username) + .fetch_one(&state.db_pool) + .await; + + let unread_count = match unread_count_result { + Ok(row) => row.get::("count"), + Err(e) => { + eprintln!("❌ Error counting unread messages: {}", e); + 0 + } + }; + + let chat = ChatPartner { + username: partner_name.clone(), + phone, + last_message, + last_message_time, + unread_count, + }; + + println!("✅ Added chat with {} (phone: {}), last_message: {:?}, unread: {}", + chat.username, chat.phone, chat.last_message, chat.unread_count); + chats.push(chat); + } + Ok(None) => { + eprintln!("❌ User not found: {}", partner_username); + } + Err(e) => { + eprintln!("❌ Error fetching user details for {}: {}", partner_username, e); + } + } + } + + // Сортируем по времени последнего сообщения (новые сверху) + chats.sort_by(|a, b| { + b.last_message_time.unwrap_or(0).cmp(&a.last_message_time.unwrap_or(0)) + }); + + println!("🎉 Final chat list for {}: {} chats", username, chats.len()); + for chat in &chats { + println!(" - {}: {:?} (unread: {})", chat.username, chat.last_message, chat.unread_count); + } + + Json(chats) +} + +// New endpoint to search users +async fn search_users_handler( + State(state): State>, + Query(params): Query, + headers: axum::http::HeaderMap, +) -> Json> { + // Extract token from header + let token = if let Some(auth_header) = headers.get("authorization") { + auth_header.to_str().unwrap_or("").replace("Bearer ", "") + } else { + return Json(Vec::new()); + }; + + // Get current user from token + let current_user = match verify_token(&state.db_pool, &token).await { + Ok(Some(user)) => user, + _ => return Json(Vec::new()), + }; + + let search_pattern = format!("%{}%", params.q); + + let users = sqlx::query( + r#" + SELECT username, phone + FROM users + WHERE (username LIKE ? OR phone LIKE ?) + AND username != ? + LIMIT 20 + "# + ) + .bind(&search_pattern) + .bind(&search_pattern) + .bind(¤t_user) + .fetch_all(&state.db_pool) + .await; + + match users { + Ok(rows) => { + let mut results = Vec::new(); + for row in rows { + results.push(ChatPartner { + username: row.get("username"), + phone: row.get("phone"), + last_message: None, + last_message_time: None, + unread_count: 0, + }); + } + Json(results) + } + Err(e) => { + eprintln!("Error searching users: {}", e); + Json(Vec::new()) + } + } +} + +// Mark messages as read endpoint +async fn mark_read_handler( + State(state): State>, + headers: axum::http::HeaderMap, + Json(payload): Json, +) -> Json { + // Extract token from header + let token = if let Some(auth_header) = headers.get("authorization") { + auth_header.to_str().unwrap_or("").replace("Bearer ", "") + } else { + return Json(serde_json::json!({"success": false, "error": "No token"})); + }; + + // Get current user from token + let username = match verify_token(&state.db_pool, &token).await { + Ok(Some(user)) => user, + _ => return Json(serde_json::json!({"success": false, "error": "Invalid token"})), + }; + + // Проверяем, что текущий пользователь - это получатель сообщений + if username != payload.to_user { + return Json(serde_json::json!({"success": false, "error": "Unauthorized"})); + } + + match mark_messages_as_read(&state.db_pool, &payload.from_user, &payload.to_user).await { + Ok(_) => Json(serde_json::json!({"success": true})), + Err(e) => { + eprintln!("Error marking messages as read: {}", e); + Json(serde_json::json!({"success": false, "error": "Database error"})) + } + } +} + +// Delete message endpoint +async fn delete_message_handler( + State(state): State>, + headers: axum::http::HeaderMap, + Json(payload): Json, +) -> Json { + // Extract token from header + let token = if let Some(auth_header) = headers.get("authorization") { + auth_header.to_str().unwrap_or("").replace("Bearer ", "") + } else { + return Json(DeleteMessageResponse { + success: false, + message: "No token".to_string(), + }); + }; + + // Get current user from token + let username = match verify_token(&state.db_pool, &token).await { + Ok(Some(user)) => user, + _ => return Json(DeleteMessageResponse { + success: false, + message: "Invalid token".to_string(), + }), + }; + + // Получаем информацию о сообщении + let message_result = sqlx::query( + "SELECT from_user, to_user FROM messages WHERE id = ?" + ) + .bind(&payload.message_id) + .fetch_optional(&state.db_pool) + .await; + + let message = match message_result { + Ok(Some(row)) => row, + Ok(None) => { + return Json(DeleteMessageResponse { + success: false, + message: "Message not found".to_string(), + }); + } + Err(e) => { + eprintln!("Error fetching message: {}", e); + return Json(DeleteMessageResponse { + success: false, + message: "Database error".to_string(), + }); + } + }; + + let from_user: String = message.get("from_user"); + let to_user: String = message.get("to_user"); + + // Проверяем права доступа + if from_user != username && to_user != username { + return Json(DeleteMessageResponse { + success: false, + message: "You don't have permission to delete this message".to_string(), + }); + } + + if payload.for_everyone { + // Удаление для всех + if from_user != username { + return Json(DeleteMessageResponse { + success: false, + message: "Only the sender can delete message for everyone".to_string(), + }); + } + + let result = sqlx::query( + "UPDATE messages SET deleted_for_everyone = TRUE WHERE id = ?" + ) + .bind(&payload.message_id) + .execute(&state.db_pool) + .await; + + match result { + Ok(_) => { + // Отправляем уведомление через WebSocket + let users = state.users.read().await; + if let Some(recipient_tx) = users.get(&to_user) { + let delete_msg = serde_json::json!({ + "type": "message_deleted", + "message_id": payload.message_id, + "for_everyone": true + }); + let _ = recipient_tx.send(Message::Text(delete_msg.to_string())); + } + + Json(DeleteMessageResponse { + success: true, + message: "Message deleted for everyone".to_string(), + }) + } + Err(e) => { + eprintln!("Error deleting message for everyone: {}", e); + Json(DeleteMessageResponse { + success: false, + message: "Database error".to_string(), + }) + } + } + } else { + // Удаление только для себя + let field_to_update = if username == from_user { + "deleted_for_sender" + } else { + "deleted_for_receiver" + }; + + let query = format!("UPDATE messages SET {} = TRUE WHERE id = ?", field_to_update); + + let result = sqlx::query(&query) + .bind(&payload.message_id) + .execute(&state.db_pool) + .await; + + match result { + Ok(_) => Json(DeleteMessageResponse { + success: true, + message: "Message deleted for you".to_string(), + }), + Err(e) => { + eprintln!("Error deleting message for user: {}", e); + Json(DeleteMessageResponse { + success: false, + message: "Database error".to_string(), + }) + } + } + } +} + async fn profile_handler() -> Json { - // This would normally verify the token and return user profile - // For now, return a placeholder Json(UserProfile { username: "user".to_string(), phone: "+1234567890".to_string(), @@ -413,23 +955,18 @@ async fn websocket_handler( } async fn handle_websocket(socket: WebSocket, state: Arc) { - // Split the socket into sender and receiver let (mut sender, mut receiver) = socket.split(); - - // Create channel for sending messages to this client let (tx, mut rx) = mpsc::unbounded_channel(); - // Handle authentication - receive first message with token + // Handle authentication let username = match receiver.next().await { Some(Ok(Message::Text(auth_msg))) => { match serde_json::from_str::(&auth_msg) { Ok(auth) => { if let (Some(token), Some(auth_type)) = (auth["token"].as_str(), auth["type"].as_str()) { if auth_type == "auth" { - // Verify token and get username match verify_token(&state.db_pool, token).await { Ok(Some(user_username)) => { - // Check if username is already connected if state.users.read().await.contains_key(&user_username) { let _ = sender.send(Message::Text( serde_json::json!({ @@ -440,10 +977,8 @@ async fn handle_websocket(socket: WebSocket, state: Arc) { return; } - // Register user state.users.write().await.insert(user_username.clone(), tx); - // Send success message let _ = sender.send(Message::Text( serde_json::json!({ "type": "system", @@ -451,11 +986,9 @@ async fn handle_websocket(socket: WebSocket, state: Arc) { }).to_string(), )).await; - // Send user list broadcast_user_list(&state.users).await; - // Send message history - if let Ok(messages) = get_message_history(&state.db_pool, &user_username).await { + if let Ok(messages) = get_message_history(&state.db_pool, &user_username, "").await { for msg in messages { let message_json = serde_json::to_string(&msg).unwrap(); let _ = sender.send(Message::Text(message_json)).await; @@ -483,12 +1016,8 @@ async fn handle_websocket(socket: WebSocket, state: Arc) { return; } } - } else { - return; - } - } else { - return; - } + } else { return; } + } else { return; } } Err(_) => return, } @@ -498,11 +1027,9 @@ async fn handle_websocket(socket: WebSocket, state: Arc) { println!("User {} connected", username); - // Clone state for cleanup let state_clone = state.clone(); let username_clone = username.clone(); - // Spawn a task to handle sending messages to the client let mut send_sender = sender; let send_task = tokio::spawn(async move { while let Some(message) = rx.recv().await { @@ -512,66 +1039,56 @@ async fn handle_websocket(socket: WebSocket, state: Arc) { } }); - // Handle receiving messages from the client in the main task let recv_state = state.clone(); - let recv_task = tokio::spawn(async move { while let Some(result) = receiver.next().await { match result { Ok(Message::Text(text)) => { - println!("Received message from {}: {}", username, text); - if let Ok(mut chat_msg) = serde_json::from_str::(&text) { - // Set the from_user to the authenticated username chat_msg.from_user = username.clone(); - println!("Processing message from {} to {}", chat_msg.from_user, chat_msg.to_user); - - // Save message to database if let Err(e) = save_message(&recv_state.db_pool, &chat_msg).await { eprintln!("Failed to save message: {}", e); continue; } - // Send to recipient if online let users = recv_state.users.read().await; if let Some(recipient_tx) = users.get(&chat_msg.to_user) { let message_json = serde_json::to_string(&chat_msg).unwrap(); - println!("Sending message to {}: {}", chat_msg.to_user, message_json); let _ = recipient_tx.send(Message::Text(message_json)); - } else { - println!("Recipient {} not found online", chat_msg.to_user); } - // Also send the message back to the sender for confirmation if let Some(sender_tx) = users.get(&username) { let message_json = serde_json::to_string(&chat_msg).unwrap(); let _ = sender_tx.send(Message::Text(message_json)); } - } else { - println!("Failed to parse message as ChatMessage: {}", text); + } else if let Ok(delete_msg) = serde_json::from_str::(&text) { + // Обработка удаления сообщений через WebSocket + if let (Some(msg_type), Some(message_id)) = ( + delete_msg["type"].as_str(), + delete_msg["message_id"].as_str() + ) { + if msg_type == "delete_message" { + let for_everyone = delete_msg["for_everyone"].as_bool().unwrap_or(false); + + // Здесь можно добавить логику удаления через WebSocket + // Пока что используем HTTP API + } + } } } - Ok(Message::Close(_)) => { - println!("WebSocket closed by client"); - break; - } - Err(e) => { - eprintln!("WebSocket error: {}", e); - break; - } + Ok(Message::Close(_)) => break, + Err(_) => break, _ => {} } } }); - // Wait for both tasks to finish tokio::select! { _ = send_task => {}, _ = recv_task => {}, } - // Cleanup state_clone.users.write().await.remove(&username_clone); broadcast_user_list(&state_clone.users).await; println!("User {} disconnected", username_clone); @@ -625,14 +1142,53 @@ async fn save_message(pool: &SqlitePool, msg: &ChatMessage) -> Result<(), sqlx:: Ok(()) } -async fn get_message_history(pool: &SqlitePool, username: &str) -> Result, sqlx::Error> { - let rows = sqlx::query( - "SELECT id, from_user, to_user, content, timestamp, message_type FROM messages WHERE from_user = ? OR to_user = ? ORDER BY timestamp" - ) - .bind(username) - .bind(username) - .fetch_all(pool) - .await?; +async fn get_message_history(pool: &SqlitePool, username: &str, other_user: &str) -> Result, sqlx::Error> { + // Если указан конкретный собеседник, помечаем сообщения как прочитанные + if !other_user.is_empty() { + mark_messages_as_read(pool, other_user, username).await?; + } + + let query = if other_user.is_empty() { + // Получаем все сообщения пользователя (исключая удаленные) + r#" + SELECT id, from_user, to_user, content, timestamp, message_type + FROM messages + WHERE (from_user = ? OR to_user = ?) + AND deleted_for_everyone = FALSE + AND ((from_user = ? AND deleted_for_sender = FALSE) OR (to_user = ? AND deleted_for_receiver = FALSE)) + ORDER BY timestamp + "# + } else { + // Получаем сообщения только с конкретным пользователем (исключая удаленные) + r#" + SELECT id, from_user, to_user, content, timestamp, message_type + FROM messages + WHERE ((from_user = ? AND to_user = ?) OR (from_user = ? AND to_user = ?)) + AND deleted_for_everyone = FALSE + AND ((from_user = ? AND deleted_for_sender = FALSE) OR (to_user = ? AND deleted_for_receiver = FALSE)) + ORDER BY timestamp + "# + }; + + let rows = if other_user.is_empty() { + sqlx::query(query) + .bind(username) + .bind(username) + .bind(username) + .bind(username) + .fetch_all(pool) + .await? + } else { + sqlx::query(query) + .bind(username) + .bind(other_user) + .bind(other_user) + .bind(username) + .bind(username) + .bind(username) + .fetch_all(pool) + .await? + }; let mut messages = Vec::new(); for row in rows { @@ -648,4 +1204,17 @@ async fn get_message_history(pool: &SqlitePool, username: &str) -> Result Result<(), sqlx::Error> { + let result = sqlx::query( + "UPDATE messages SET read = TRUE WHERE from_user = ? AND to_user = ? AND read = FALSE AND deleted_for_everyone = FALSE" + ) + .bind(from_user) + .bind(to_user) + .execute(pool) + .await?; + + println!("✅ Marked {} messages as read from {} to {}", result.rows_affected(), from_user, to_user); + Ok(()) } \ No newline at end of file diff --git a/web/index.html b/web/index.html index 8d3c350..5a170bc 100644 --- a/web/index.html +++ b/web/index.html @@ -5,348 +5,93 @@ Secure Chat App @@ -392,14 +137,31 @@