From 7f7936d37f2523cab154d204c60cefa844ae7159 Mon Sep 17 00:00:00 2001 From: TolaMironcenko Date: Fri, 28 Nov 2025 01:38:59 +0700 Subject: [PATCH] more refactor on backend --- config.json | 38 +- src/api.rs | 9 + src/api/chats.rs | 214 +++++ src/api/debug.rs | 94 +++ src/api/login.rs | 77 ++ src/api/messages.rs | 39 + src/api/messages/delete.rs | 141 ++++ src/api/profile.rs | 2 + src/api/profile/avatar.rs | 116 +++ src/api/profile/get.rs | 54 ++ src/api/register.rs | 117 +++ src/api/search.rs | 48 ++ src/api/security.rs | 63 ++ src/api/websock.rs | 174 ++++ src/conf.rs | 71 ++ src/db.rs | 61 ++ src/main.rs | 1541 +----------------------------------- src/model.rs | 93 +++ src/state.rs | 14 + src/utils.rs | 164 ++++ 20 files changed, 1603 insertions(+), 1527 deletions(-) create mode 100644 src/api.rs create mode 100644 src/api/chats.rs create mode 100644 src/api/debug.rs create mode 100644 src/api/login.rs create mode 100644 src/api/messages.rs create mode 100644 src/api/messages/delete.rs create mode 100644 src/api/profile.rs create mode 100644 src/api/profile/avatar.rs create mode 100644 src/api/profile/get.rs create mode 100644 src/api/register.rs create mode 100644 src/api/search.rs create mode 100644 src/api/security.rs create mode 100644 src/api/websock.rs create mode 100644 src/conf.rs create mode 100644 src/db.rs create mode 100644 src/model.rs create mode 100644 src/state.rs create mode 100644 src/utils.rs diff --git a/config.json b/config.json index 9baf180..2355b4a 100644 --- a/config.json +++ b/config.json @@ -1,20 +1,20 @@ { - "server": { - "host": "0.0.0.0", - "port": 3333, - "max_connections": 10000 - }, - "database": { - "url": "sqlite:chat.db" - }, - "files": { - "avatars_dir": "web/avatars", - "web_dir": "web", - "max_avatar_size": 5242880 - }, - "limits": { - "max_message_length": 4096, - "max_username_length": 50, - "min_password_length": 6 - } -} \ No newline at end of file + "server": { + "host": "0.0.0.0", + "port": 3333, + "max_connections": 10000 + }, + "database": { + "url": "chat.db" + }, + "files": { + "avatars_dir": "web/avatars", + "web_dir": "web", + "max_avatar_size": 5242880 + }, + "limits": { + "max_message_length": 4096, + "max_username_length": 50, + "min_password_length": 6 + } +} diff --git a/src/api.rs b/src/api.rs new file mode 100644 index 0000000..3a0b5ab --- /dev/null +++ b/src/api.rs @@ -0,0 +1,9 @@ +pub mod chats; +pub mod debug; +pub mod login; +pub mod messages; +pub mod profile; +pub mod register; +pub mod search; +pub mod security; +pub mod websock; diff --git a/src/api/chats.rs b/src/api/chats.rs new file mode 100644 index 0000000..88ddfc9 --- /dev/null +++ b/src/api/chats.rs @@ -0,0 +1,214 @@ +use crate::model; +use crate::state; +use crate::utils; +use axum::extract::{Json, State}; +use sqlx::Row; +use std::sync::Arc; + +// УПРОЩЕННАЯ версия получения чатов +pub 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 utils::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, avatar_url 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 avatar_url: Option = user_row.get("avatar_url"); + + // Получаем последнее сообщение + 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 = model::ChatPartner { + username: partner_name.clone(), + phone, + avatar_url, + 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) +} diff --git a/src/api/debug.rs b/src/api/debug.rs new file mode 100644 index 0000000..59b9eff --- /dev/null +++ b/src/api/debug.rs @@ -0,0 +1,94 @@ +use crate::state; +use crate::utils; +use axum::extract::{Json, State}; +use sqlx::Row; +use std::sync::Arc; + +// Debug endpoint to see all messages +pub 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 utils::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)})), + } +} diff --git a/src/api/login.rs b/src/api/login.rs new file mode 100644 index 0000000..7574cd5 --- /dev/null +++ b/src/api/login.rs @@ -0,0 +1,77 @@ +use crate::model; +use crate::state; +use crate::utils; +use axum::extract::{Json, State}; +use bcrypt::verify; +use chrono::{Duration, Utc}; +use sqlx::Row; +use std::sync::Arc; + +pub async fn login_handler( + State(state): State>, + Json(payload): Json, +) -> Json { + // Get user from database + let user_result = sqlx::query("SELECT username, password_hash FROM users WHERE phone = ?") + .bind(&payload.phone) + .fetch_optional(&state.db_pool) + .await; + + match user_result { + Ok(Some(row)) => { + let username: String = row.get("username"); + let password_hash: String = row.get("password_hash"); + + // Verify password + match verify(&payload.password, &password_hash) { + Ok(true) => { + // Generate session token + let token = utils::generate_token(); + let expires_at = Utc::now() + Duration::days(30); + + // Store session + let _ = sqlx::query( + "INSERT INTO sessions (token, username, phone, expires_at) VALUES (?, ?, ?, ?)" + ) + .bind(&token) + .bind(&username) + .bind(&payload.phone) + .bind(expires_at.naive_utc()) + .execute(&state.db_pool) + .await; + + Json(model::AuthResponse { + success: true, + message: "Login successful".to_string(), + username: Some(username), + token: Some(token), + }) + } + Ok(false) => Json(model::AuthResponse { + success: false, + message: "Invalid password".to_string(), + username: None, + token: None, + }), + Err(_) => Json(model::AuthResponse { + success: false, + message: "Error verifying password".to_string(), + username: None, + token: None, + }), + } + } + Ok(None) => Json(model::AuthResponse { + success: false, + message: "Phone number not found".to_string(), + username: None, + token: None, + }), + Err(_) => Json(model::AuthResponse { + success: false, + message: "Login failed".to_string(), + username: None, + token: None, + }), + } +} diff --git a/src/api/messages.rs b/src/api/messages.rs new file mode 100644 index 0000000..18da808 --- /dev/null +++ b/src/api/messages.rs @@ -0,0 +1,39 @@ +pub mod delete; +use crate::model; +use crate::state; +use crate::utils; +use axum::extract::{Json, State}; +use std::sync::Arc; + +// Mark messages as read endpoint +pub 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 utils::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 utils::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"})) + } + } +} diff --git a/src/api/messages/delete.rs b/src/api/messages/delete.rs new file mode 100644 index 0000000..6318ec7 --- /dev/null +++ b/src/api/messages/delete.rs @@ -0,0 +1,141 @@ +use crate::model; +use crate::state; +use crate::utils; +use axum::extract::{Json, State, ws::Message}; +use sqlx::Row; +use std::sync::Arc; + +// Delete message endpoint +pub 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(model::DeleteMessageResponse { + success: false, + message: "No token".to_string(), + }); + }; + + // Get current user from token + let username = match utils::verify_token(&state.db_pool, &token).await { + Ok(Some(user)) => user, + _ => { + return Json(model::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(model::DeleteMessageResponse { + success: false, + message: "Message not found".to_string(), + }); + } + Err(e) => { + eprintln!("Error fetching message: {}", e); + return Json(model::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(model::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(model::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(model::DeleteMessageResponse { + success: true, + message: "Message deleted for everyone".to_string(), + }) + } + Err(e) => { + eprintln!("Error deleting message for everyone: {}", e); + Json(model::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(model::DeleteMessageResponse { + success: true, + message: "Message deleted for you".to_string(), + }), + Err(e) => { + eprintln!("Error deleting message for user: {}", e); + Json(model::DeleteMessageResponse { + success: false, + message: "Database error".to_string(), + }) + } + } + } +} diff --git a/src/api/profile.rs b/src/api/profile.rs new file mode 100644 index 0000000..2c77115 --- /dev/null +++ b/src/api/profile.rs @@ -0,0 +1,2 @@ +pub mod avatar; +pub mod get; diff --git a/src/api/profile/avatar.rs b/src/api/profile/avatar.rs new file mode 100644 index 0000000..4cff39d --- /dev/null +++ b/src/api/profile/avatar.rs @@ -0,0 +1,116 @@ +use crate::model; +use crate::state; +use crate::utils; +use axum::extract::{Json, Multipart, State}; +use std::sync::Arc; +use uuid::Uuid; + +// Upload avatar endpoint +pub async fn upload_avatar_handler( + State(state): State>, + headers: axum::http::HeaderMap, + mut multipart: Multipart, +) -> 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(model::UploadAvatarResponse { + success: false, + avatar_url: None, + message: "No token".to_string(), + }); + }; + + // Get current user from token + let username = match utils::verify_token(&state.db_pool, &token).await { + Ok(Some(user)) => user, + _ => { + return Json(model::UploadAvatarResponse { + success: false, + avatar_url: None, + message: "Invalid token".to_string(), + }); + } + }; + + // Process multipart form + while let Some(field) = multipart.next_field().await.unwrap() { + let field_name = field.name().unwrap_or("").to_string(); + + if field_name == "avatar" { + let data = field.bytes().await.unwrap(); + + // Validate image size using config + if data.len() > state.config.files.max_avatar_size { + return Json(model::UploadAvatarResponse { + success: false, + avatar_url: None, + message: format!( + "Image too large (max {} bytes)", + state.config.files.max_avatar_size + ), + }); + } + + // Process image + match utils::process_avatar(&data).await { + Ok(processed_data) => { + // Generate unique filename + let filename = format!("{}.jpg", Uuid::new_v4()); + let avatar_path = format!("{}/{}", state.config.files.avatars_dir, filename); + + // Save file + if let Err(e) = tokio::fs::write(&avatar_path, &processed_data).await { + eprintln!("Error saving avatar: {}", e); + return Json(model::UploadAvatarResponse { + success: false, + avatar_url: None, + message: "Error saving image".to_string(), + }); + } + + let avatar_url = format!("/avatars/{}", filename); + + // Update user in database + let result = sqlx::query("UPDATE users SET avatar_url = ? WHERE username = ?") + .bind(&avatar_url) + .bind(&username) + .execute(&state.db_pool) + .await; + + match result { + Ok(_) => { + return Json(model::UploadAvatarResponse { + success: true, + avatar_url: Some(avatar_url), + message: "Avatar uploaded successfully".to_string(), + }); + } + Err(e) => { + eprintln!("Error updating user avatar: {}", e); + return Json(model::UploadAvatarResponse { + success: false, + avatar_url: None, + message: "Database error".to_string(), + }); + } + } + } + Err(e) => { + return Json(model::UploadAvatarResponse { + success: false, + avatar_url: None, + message: e, + }); + } + } + } + } + + Json(model::UploadAvatarResponse { + success: false, + avatar_url: None, + message: "No avatar file provided".to_string(), + }) +} diff --git a/src/api/profile/get.rs b/src/api/profile/get.rs new file mode 100644 index 0000000..ac22f29 --- /dev/null +++ b/src/api/profile/get.rs @@ -0,0 +1,54 @@ +use crate::model; +use crate::state; +use crate::utils; +use axum::extract::{Json, State}; +use sqlx::Row; +use std::sync::Arc; + +pub async fn profile_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(model::UserProfile { + username: "user".to_string(), + phone: "+1234567890".to_string(), + avatar_url: None, + }); + }; + + // Get current user from token + let username = match utils::verify_token(&state.db_pool, &token).await { + Ok(Some(user)) => user, + _ => { + return Json(model::UserProfile { + username: "user".to_string(), + phone: "+1234567890".to_string(), + avatar_url: None, + }); + } + }; + + // Get user details + let user_result = + sqlx::query("SELECT username, phone, avatar_url FROM users WHERE username = ?") + .bind(&username) + .fetch_optional(&state.db_pool) + .await; + + match user_result { + Ok(Some(row)) => Json(model::UserProfile { + username: row.get("username"), + phone: row.get("phone"), + avatar_url: row.get("avatar_url"), + }), + _ => Json(model::UserProfile { + username: "user".to_string(), + phone: "+1234567890".to_string(), + avatar_url: None, + }), + } +} diff --git a/src/api/register.rs b/src/api/register.rs new file mode 100644 index 0000000..4e5b035 --- /dev/null +++ b/src/api/register.rs @@ -0,0 +1,117 @@ +use crate::model; +use crate::state; +use crate::utils; +use axum::extract::{Json, State}; +use bcrypt::{DEFAULT_COST, hash}; +use chrono::{Duration, Utc}; +use std::sync::Arc; + +pub async fn register_handler( + State(state): State>, + Json(payload): Json, +) -> Json { + // Validate input + if payload.phone.len() < 5 { + return Json(model::AuthResponse { + success: false, + message: "Phone number is too short".to_string(), + username: None, + token: None, + }); + } + + if payload.username.len() < 3 + || payload.username.len() > state.config.limits.max_username_length + { + return Json(model::AuthResponse { + success: false, + message: format!( + "Username must be between 3 and {} characters", + state.config.limits.max_username_length + ), + username: None, + token: None, + }); + } + + if payload.password.len() < state.config.limits.min_password_length { + return Json(model::AuthResponse { + success: false, + message: format!( + "Password must be at least {} characters", + state.config.limits.min_password_length + ), + username: None, + token: None, + }); + } + + // Hash password + let password_hash = match hash(&payload.password, DEFAULT_COST) { + Ok(hash) => hash, + Err(_) => { + return Json(model::AuthResponse { + success: false, + message: "Error hashing password".to_string(), + username: None, + token: None, + }); + } + }; + + // Insert user into database + let result = sqlx::query("INSERT INTO users (phone, username, password_hash) VALUES (?, ?, ?)") + .bind(&payload.phone) + .bind(&payload.username) + .bind(&password_hash) + .execute(&state.db_pool) + .await; + + match result { + Ok(_) => { + // Generate session token + let token = utils::generate_token(); + let expires_at = Utc::now() + Duration::days(30); + + // Store session + let _ = sqlx::query( + "INSERT INTO sessions (token, username, phone, expires_at) VALUES (?, ?, ?, ?)", + ) + .bind(&token) + .bind(&payload.username) + .bind(&payload.phone) + .bind(expires_at.naive_utc()) + .execute(&state.db_pool) + .await; + + Json(model::AuthResponse { + success: true, + message: "Registration successful".to_string(), + username: Some(payload.username), + token: Some(token), + }) + } + Err(e) => { + let error_message = if e + .to_string() + .contains("UNIQUE constraint failed: users.phone") + { + "Phone number already registered".to_string() + } else if e + .to_string() + .contains("UNIQUE constraint failed: users.username") + { + "Username already taken".to_string() + } else { + "Registration failed".to_string() + }; + + Json(model::AuthResponse { + success: false, + message: error_message, + username: None, + token: None, + }) + } + } +} diff --git a/src/api/search.rs b/src/api/search.rs new file mode 100644 index 0000000..ae8e89a --- /dev/null +++ b/src/api/search.rs @@ -0,0 +1,48 @@ +use crate::model; +use crate::state; +use axum::extract::{Json, Query, State}; +use sqlx::Row; +use std::sync::Arc; + +// New endpoint to search users +pub async fn search_users_handler( + State(state): State>, + Query(params): Query, + // headers: axum::http::HeaderMap, +) -> Json> { + let search_pattern = format!("%{}%", params.q); + + let users = sqlx::query( + r#" + SELECT username, phone, avatar_url + FROM users + WHERE (username LIKE ? OR phone LIKE ?) + LIMIT 20 + "#, + ) + .bind(&search_pattern) + .bind(&search_pattern) + .fetch_all(&state.db_pool) + .await; + + match users { + Ok(rows) => { + let mut results = Vec::new(); + for row in rows { + results.push(model::ChatPartner { + username: row.get("username"), + phone: row.get("phone"), + avatar_url: row.get("avatar_url"), + last_message: None, + last_message_time: None, + unread_count: 0, + }); + } + Json(results) + } + Err(e) => { + eprintln!("Error searching users: {}", e); + Json(Vec::new()) + } + } +} diff --git a/src/api/security.rs b/src/api/security.rs new file mode 100644 index 0000000..374c64f --- /dev/null +++ b/src/api/security.rs @@ -0,0 +1,63 @@ +use crate::model; +use crate::state; +use crate::utils; +use axum::extract::{Json, State}; +use sqlx::Row; +use std::sync::Arc; + +pub async fn validate_token_handler( + State(state): State>, + Json(payload): Json, +) -> Json { + match utils::verify_token(&state.db_pool, &payload.token).await { + Ok(Some(username)) => { + // Get user details + let user_result = sqlx::query("SELECT phone, avatar_url FROM users WHERE username = ?") + .bind(&username) + .fetch_optional(&state.db_pool) + .await; + + match user_result { + Ok(Some(row)) => { + let phone: String = row.get("phone"); + let avatar_url: Option = row.get("avatar_url"); + Json(model::ValidateTokenResponse { + success: true, + username: Some(username), + phone: Some(phone), + avatar_url, + message: None, + }) + } + Ok(None) => Json(model::ValidateTokenResponse { + success: false, + username: None, + phone: None, + avatar_url: None, + message: Some("User not found".to_string()), + }), + Err(_) => Json(model::ValidateTokenResponse { + success: false, + username: None, + phone: None, + avatar_url: None, + message: Some("Database error".to_string()), + }), + } + } + Ok(None) => Json(model::ValidateTokenResponse { + success: false, + username: None, + phone: None, + avatar_url: None, + message: Some("Invalid or expired token".to_string()), + }), + Err(_) => Json(model::ValidateTokenResponse { + success: false, + username: None, + phone: None, + avatar_url: None, + message: Some("Token verification error".to_string()), + }), + } +} diff --git a/src/api/websock.rs b/src/api/websock.rs new file mode 100644 index 0000000..c5ec7a4 --- /dev/null +++ b/src/api/websock.rs @@ -0,0 +1,174 @@ +use crate::model; +use crate::state; +use crate::utils; +use axum::{ + extract::{ + State, + ws::{Message, WebSocket, WebSocketUpgrade}, + }, + response::Response, +}; +use futures_util::{SinkExt, StreamExt}; +use std::sync::Arc; +use tokio::sync::mpsc; + +// WebSocket handlers +pub async fn websocket_handler( + ws: WebSocketUpgrade, + State(state): State>, +) -> Response { + ws.on_upgrade(|socket| handle_websocket(socket, state)) +} + +async fn handle_websocket(socket: WebSocket, state: Arc) { + let (mut sender, mut receiver) = socket.split(); + let (tx, mut rx) = mpsc::unbounded_channel(); + + // 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" { + match utils::verify_token(&state.db_pool, token).await { + Ok(Some(user_username)) => { + if state.users.read().await.contains_key(&user_username) { + let _ = sender + .send(Message::Text( + serde_json::json!({ + "type": "error", + "content": "User already connected" + }) + .to_string(), + )) + .await; + return; + } + + state.users.write().await.insert(user_username.clone(), tx); + + let _ = sender + .send(Message::Text( + serde_json::json!({ + "type": "system", + "content": "Connected successfully" + }) + .to_string(), + )) + .await; + + utils::broadcast_user_list(&state.users).await; + + if let Ok(messages) = utils::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; + } + } + + user_username + } + Ok(None) => { + let _ = sender + .send(Message::Text( + serde_json::json!({ + "type": "error", + "content": "Invalid or expired token" + }) + .to_string(), + )) + .await; + return; + } + Err(_) => { + let _ = sender + .send(Message::Text( + serde_json::json!({ + "type": "error", + "content": "Authentication error" + }) + .to_string(), + )) + .await; + return; + } + } + } else { + return; + } + } else { + return; + } + } + Err(_) => return, + } + } + _ => return, + }; + + println!("User {} connected", username); + + let state_clone = state.clone(); + let username_clone = username.clone(); + + let mut send_sender = sender; + let send_task = tokio::spawn(async move { + while let Some(message) = rx.recv().await { + if send_sender.send(message).await.is_err() { + break; + } + } + }); + + 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)) => { + if let Ok(mut chat_msg) = serde_json::from_str::(&text) { + chat_msg.from_user = username.clone(); + + if let Err(e) = utils::save_message(&recv_state.db_pool, &chat_msg).await { + eprintln!("Failed to save message: {}", e); + continue; + } + + let users = recv_state.users.read().await; + if chat_msg.to_user != chat_msg.from_user { + if let Some(recipient_tx) = users.get(&chat_msg.to_user) { + let message_json = serde_json::to_string(&chat_msg).unwrap(); + let _ = recipient_tx.send(Message::Text(message_json)); + } + } + + 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)); + } + } + } + Ok(Message::Close(_)) => break, + Err(_) => break, + _ => {} + } + } + }); + + tokio::select! { + _ = send_task => {}, + _ = recv_task => {}, + } + + state_clone.users.write().await.remove(&username_clone); + utils::broadcast_user_list(&state_clone.users).await; + println!("User {} disconnected", username_clone); +} diff --git a/src/conf.rs b/src/conf.rs new file mode 100644 index 0000000..8b57501 --- /dev/null +++ b/src/conf.rs @@ -0,0 +1,71 @@ +use config::{Config as ConfigLoader, File, FileFormat}; +use serde::Deserialize; + +// Структура для конфигурации +#[derive(Debug, Deserialize, Clone)] +pub struct AppConfig { + pub server: ServerConfig, + pub database: DatabaseConfig, + pub files: FilesConfig, + pub limits: LimitsConfig, +} + +#[derive(Debug, Deserialize, Clone)] +pub struct ServerConfig { + pub host: String, + pub port: u16, + pub max_connections: Option, +} + +#[derive(Debug, Deserialize, Clone)] +pub struct DatabaseConfig { + pub url: String, +} + +#[derive(Debug, Deserialize, Clone)] +pub struct FilesConfig { + pub avatars_dir: String, + pub web_dir: String, + pub max_avatar_size: usize, +} + +#[derive(Debug, Deserialize, Clone)] +pub struct LimitsConfig { + // max_message_length: usize, + pub max_username_length: usize, + pub min_password_length: usize, +} + +impl AppConfig { + pub fn load() -> Result> { + let config_loader = ConfigLoader::builder() + // Добавляем конфиг по умолчанию + .add_source(File::from_str( + r#" + [server] + host = "0.0.0.0" + port = 3333 + max_connections = 10000 + + [database] + url = "sqlite:chat.db" + + [files] + avatars_dir = "web/avatars" + web_dir = "web" + max_avatar_size = 5242880 + + [limits] + max_message_length = 4096 + max_username_length = 50 + min_password_length = 6 + "#, + FileFormat::Toml, + )) + // Перезаписываем значения из config.json если он существует + .add_source(File::with_name("config").required(false)) + .build()?; + + Ok(config_loader.try_deserialize()?) + } +} diff --git a/src/db.rs b/src/db.rs new file mode 100644 index 0000000..4d88731 --- /dev/null +++ b/src/db.rs @@ -0,0 +1,61 @@ +use sqlx::{SqlitePool, sqlite::SqliteConnectOptions}; + +pub async fn init_db(database_url: &str) -> Result> { + let options = SqliteConnectOptions::new() + .filename(database_url) + .create_if_missing(true); + let pool = SqlitePool::connect_with(options).await?; + + // Create users table + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + phone TEXT UNIQUE NOT NULL, + username TEXT UNIQUE NOT NULL, + password_hash TEXT NOT NULL, + avatar_url TEXT, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP + ) + "#, + ) + .execute(&pool) + .await?; + + // Create messages table + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS messages ( + id TEXT PRIMARY KEY, + from_user TEXT NOT NULL, + to_user TEXT NOT NULL, + content TEXT NOT NULL, + timestamp INTEGER 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 + ) + "#, + ) + .execute(&pool) + .await?; + + // Create sessions table + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS sessions ( + token TEXT PRIMARY KEY, + username TEXT NOT NULL, + phone TEXT NOT NULL, + expires_at DATETIME NOT NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP + ) + "#, + ) + .execute(&pool) + .await?; + + Ok(pool) +} diff --git a/src/main.rs b/src/main.rs index 5282c3d..3a6696e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,202 +1,17 @@ use axum::{ Router, - extract::{ - Json, Multipart, Query, State, - ws::{Message, WebSocket, WebSocketUpgrade}, - }, - response::Response, routing::{get, post}, }; -use bcrypt::{DEFAULT_COST, hash, verify}; -use chrono::{Duration, Utc}; -use config::{Config as ConfigLoader, File, FileFormat}; -use futures_util::{SinkExt, StreamExt}; -use rand::distributions::Alphanumeric; -use rand::{Rng, thread_rng}; -use serde::{Deserialize, Serialize}; -use sqlx::{Row, sqlite::SqlitePool}; -use std::io::Cursor; use std::{collections::HashMap, net::SocketAddr, sync::Arc}; -use tokio::sync::{RwLock, mpsc}; +use tokio::sync::RwLock; use tower_http::services::ServeDir; -use uuid::Uuid; -// Types for shared state -type UserMap = Arc>>>; - -// Структура для конфигурации -#[derive(Debug, Deserialize, Clone)] -struct AppConfig { - server: ServerConfig, - database: DatabaseConfig, - files: FilesConfig, - limits: LimitsConfig, -} - -#[derive(Debug, Deserialize, Clone)] -struct ServerConfig { - host: String, - port: u16, - max_connections: Option, -} - -#[derive(Debug, Deserialize, Clone)] -struct DatabaseConfig { - url: String, -} - -#[derive(Debug, Deserialize, Clone)] -struct FilesConfig { - avatars_dir: String, - web_dir: String, - max_avatar_size: usize, -} - -#[derive(Debug, Deserialize, Clone)] -struct LimitsConfig { - // max_message_length: usize, - max_username_length: usize, - min_password_length: usize, -} - -impl AppConfig { - fn load() -> Result> { - let config_loader = ConfigLoader::builder() - // Добавляем конфиг по умолчанию - .add_source(File::from_str( - r#" - [server] - host = "0.0.0.0" - port = 3333 - max_connections = 10000 - - [database] - url = "sqlite:chat.db" - - [files] - avatars_dir = "web/avatars" - web_dir = "web" - max_avatar_size = 5242880 - - [limits] - max_message_length = 4096 - max_username_length = 50 - min_password_length = 6 - "#, - FileFormat::Toml, - )) - // Перезаписываем значения из config.json если он существует - .add_source(File::with_name("config").required(false)) - .build()?; - - Ok(config_loader.try_deserialize()?) - } -} - -#[derive(Clone)] -struct AppState { - users: UserMap, - db_pool: SqlitePool, - config: AppConfig, -} - -#[derive(Debug, Serialize, Deserialize, Clone)] -struct ChatMessage { - id: String, - from_user: String, - to_user: String, - content: String, - timestamp: i64, - message_type: String, -} - -// #[derive(Debug, Serialize, Deserialize)] -// struct UserList { -// users: Vec, -// } - -// Структуры для аутентификации -#[derive(Debug, Deserialize)] -struct RegisterRequest { - phone: String, - username: String, - password: String, -} - -#[derive(Debug, Deserialize)] -struct LoginRequest { - phone: String, - password: String, -} - -#[derive(Debug, Serialize)] -struct AuthResponse { - success: bool, - message: String, - username: Option, - token: Option, -} - -#[derive(Debug, Serialize)] -struct UserProfile { - username: String, - phone: String, - avatar_url: Option, -} - -#[derive(Debug, Serialize)] -struct ValidateTokenResponse { - success: bool, - username: Option, - phone: Option, - avatar_url: Option, - message: Option, -} - -#[derive(Debug, Serialize)] -struct ChatPartner { - username: String, - phone: String, - avatar_url: Option, - 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, -} - -#[derive(Debug, Serialize)] -struct UploadAvatarResponse { - success: bool, - avatar_url: Option, - message: String, -} - -#[derive(Debug, Deserialize)] -struct ValidateTokenRequest { - token: String, -} +mod api; +mod conf; +mod db; +mod model; +mod state; +mod utils; #[tokio::main] async fn main() -> Result<(), Box> { @@ -204,7 +19,7 @@ async fn main() -> Result<(), Box> { tracing_subscriber::fmt::init(); // Load configuration - let config = AppConfig::load()?; + let config = conf::AppConfig::load()?; println!("📋 Loaded configuration: {:?}", config); // Create directories if they don't exist @@ -212,10 +27,10 @@ async fn main() -> Result<(), Box> { tokio::fs::create_dir_all(&config.files.web_dir).await?; // Initialize database with configured URL - let db_pool = init_db(&config.database.url).await?; + let db_pool = db::init_db(&config.database.url).await?; // Create shared state - let state = Arc::new(AppState { + let state = Arc::new(state::AppState { users: Arc::new(RwLock::new(HashMap::new())), db_pool, config, @@ -223,17 +38,29 @@ async fn main() -> Result<(), Box> { // Build the application let app = Router::new() - .route("/ws", get(websocket_handler)) - .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/upload-avatar", post(upload_avatar_handler)) - .route("/api/profile", get(profile_handler)) + .route("/ws", get(api::websock::websocket_handler)) + .route("/api/register", post(api::register::register_handler)) + .route("/api/login", post(api::login::login_handler)) + .route( + "/api/validate-token", + post(api::security::validate_token_handler), + ) + .route("/api/chats", get(api::chats::get_chats_handler)) + .route("/api/search-users", get(api::search::search_users_handler)) + .route( + "/api/debug-messages", + get(api::debug::debug_messages_handler), + ) + .route("/api/mark-read", post(api::messages::mark_read_handler)) + .route( + "/api/delete-message", + post(api::messages::delete::delete_message_handler), + ) + .route( + "/api/upload-avatar", + post(api::profile::avatar::upload_avatar_handler), + ) + .route("/api/profile", get(api::profile::get::profile_handler)) .nest_service("/", ServeDir::new(&state.config.files.web_dir)) .with_state(state.clone()); @@ -254,1305 +81,3 @@ async fn main() -> Result<(), Box> { Ok(()) } - -async fn init_db(database_url: &str) -> Result> { - let pool = SqlitePool::connect(database_url).await?; - - // Create users table - sqlx::query( - r#" - CREATE TABLE IF NOT EXISTS users ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - phone TEXT UNIQUE NOT NULL, - username TEXT UNIQUE NOT NULL, - password_hash TEXT NOT NULL, - avatar_url TEXT, - created_at DATETIME DEFAULT CURRENT_TIMESTAMP - ) - "#, - ) - .execute(&pool) - .await?; - - // Create messages table - sqlx::query( - r#" - CREATE TABLE IF NOT EXISTS messages ( - id TEXT PRIMARY KEY, - from_user TEXT NOT NULL, - to_user TEXT NOT NULL, - content TEXT NOT NULL, - timestamp INTEGER 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 - ) - "#, - ) - .execute(&pool) - .await?; - - // Create sessions table - sqlx::query( - r#" - CREATE TABLE IF NOT EXISTS sessions ( - token TEXT PRIMARY KEY, - username TEXT NOT NULL, - phone TEXT NOT NULL, - expires_at DATETIME NOT NULL, - created_at DATETIME DEFAULT CURRENT_TIMESTAMP - ) - "#, - ) - .execute(&pool) - .await?; - - Ok(pool) -} - -// Обработчики HTTP API -async fn register_handler( - State(state): State>, - Json(payload): Json, -) -> Json { - // Validate input - if payload.phone.len() < 5 { - return Json(AuthResponse { - success: false, - message: "Phone number is too short".to_string(), - username: None, - token: None, - }); - } - - if payload.username.len() < 3 - || payload.username.len() > state.config.limits.max_username_length - { - return Json(AuthResponse { - success: false, - message: format!( - "Username must be between 3 and {} characters", - state.config.limits.max_username_length - ), - username: None, - token: None, - }); - } - - if payload.password.len() < state.config.limits.min_password_length { - return Json(AuthResponse { - success: false, - message: format!( - "Password must be at least {} characters", - state.config.limits.min_password_length - ), - username: None, - token: None, - }); - } - - // Hash password - let password_hash = match hash(&payload.password, DEFAULT_COST) { - Ok(hash) => hash, - Err(_) => { - return Json(AuthResponse { - success: false, - message: "Error hashing password".to_string(), - username: None, - token: None, - }); - } - }; - - // Insert user into database - let result = sqlx::query("INSERT INTO users (phone, username, password_hash) VALUES (?, ?, ?)") - .bind(&payload.phone) - .bind(&payload.username) - .bind(&password_hash) - .execute(&state.db_pool) - .await; - - match result { - Ok(_) => { - // Generate session token - let token = generate_token(); - let expires_at = Utc::now() + Duration::days(30); - - // Store session - let _ = sqlx::query( - "INSERT INTO sessions (token, username, phone, expires_at) VALUES (?, ?, ?, ?)", - ) - .bind(&token) - .bind(&payload.username) - .bind(&payload.phone) - .bind(expires_at.naive_utc()) - .execute(&state.db_pool) - .await; - - Json(AuthResponse { - success: true, - message: "Registration successful".to_string(), - username: Some(payload.username), - token: Some(token), - }) - } - Err(e) => { - let error_message = if e - .to_string() - .contains("UNIQUE constraint failed: users.phone") - { - "Phone number already registered".to_string() - } else if e - .to_string() - .contains("UNIQUE constraint failed: users.username") - { - "Username already taken".to_string() - } else { - "Registration failed".to_string() - }; - - Json(AuthResponse { - success: false, - message: error_message, - username: None, - token: None, - }) - } - } -} - -async fn login_handler( - State(state): State>, - Json(payload): Json, -) -> Json { - // Get user from database - let user_result = sqlx::query("SELECT username, password_hash FROM users WHERE phone = ?") - .bind(&payload.phone) - .fetch_optional(&state.db_pool) - .await; - - match user_result { - Ok(Some(row)) => { - let username: String = row.get("username"); - let password_hash: String = row.get("password_hash"); - - // Verify password - match verify(&payload.password, &password_hash) { - Ok(true) => { - // Generate session token - let token = generate_token(); - let expires_at = Utc::now() + Duration::days(30); - - // Store session - let _ = sqlx::query( - "INSERT INTO sessions (token, username, phone, expires_at) VALUES (?, ?, ?, ?)" - ) - .bind(&token) - .bind(&username) - .bind(&payload.phone) - .bind(expires_at.naive_utc()) - .execute(&state.db_pool) - .await; - - Json(AuthResponse { - success: true, - message: "Login successful".to_string(), - username: Some(username), - token: Some(token), - }) - } - Ok(false) => Json(AuthResponse { - success: false, - message: "Invalid password".to_string(), - username: None, - token: None, - }), - Err(_) => Json(AuthResponse { - success: false, - message: "Error verifying password".to_string(), - username: None, - token: None, - }), - } - } - Ok(None) => Json(AuthResponse { - success: false, - message: "Phone number not found".to_string(), - username: None, - token: None, - }), - Err(_) => Json(AuthResponse { - success: false, - message: "Login failed".to_string(), - username: None, - token: None, - }), - } -} - -async fn validate_token_handler( - State(state): State>, - Json(payload): Json, -) -> Json { - match verify_token(&state.db_pool, &payload.token).await { - Ok(Some(username)) => { - // Get user details - let user_result = sqlx::query("SELECT phone, avatar_url FROM users WHERE username = ?") - .bind(&username) - .fetch_optional(&state.db_pool) - .await; - - match user_result { - Ok(Some(row)) => { - let phone: String = row.get("phone"); - let avatar_url: Option = row.get("avatar_url"); - Json(ValidateTokenResponse { - success: true, - username: Some(username), - phone: Some(phone), - avatar_url, - message: None, - }) - } - Ok(None) => Json(ValidateTokenResponse { - success: false, - username: None, - phone: None, - avatar_url: None, - message: Some("User not found".to_string()), - }), - Err(_) => Json(ValidateTokenResponse { - success: false, - username: None, - phone: None, - avatar_url: None, - message: Some("Database error".to_string()), - }), - } - } - Ok(None) => Json(ValidateTokenResponse { - success: false, - username: None, - phone: None, - avatar_url: None, - message: Some("Invalid or expired token".to_string()), - }), - Err(_) => Json(ValidateTokenResponse { - success: false, - username: None, - phone: None, - avatar_url: None, - message: Some("Token verification error".to_string()), - }), - } -} - -// 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, avatar_url 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 avatar_url: Option = user_row.get("avatar_url"); - - // Получаем последнее сообщение - 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, - avatar_url, - 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, avatar_url - FROM users - WHERE (username LIKE ? OR phone LIKE ?) - LIMIT 20 - "#, - ) - // AND username != ? - .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"), - avatar_url: row.get("avatar_url"), - 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(), - }) - } - } - } -} - -// Upload avatar endpoint -async fn upload_avatar_handler( - State(state): State>, - headers: axum::http::HeaderMap, - mut multipart: Multipart, -) -> 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(UploadAvatarResponse { - success: false, - avatar_url: None, - 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(UploadAvatarResponse { - success: false, - avatar_url: None, - message: "Invalid token".to_string(), - }); - } - }; - - // Process multipart form - while let Some(field) = multipart.next_field().await.unwrap() { - let field_name = field.name().unwrap_or("").to_string(); - - if field_name == "avatar" { - let data = field.bytes().await.unwrap(); - - // Validate image size using config - if data.len() > state.config.files.max_avatar_size { - return Json(UploadAvatarResponse { - success: false, - avatar_url: None, - message: format!( - "Image too large (max {} bytes)", - state.config.files.max_avatar_size - ), - }); - } - - // Process image - match process_avatar(&data).await { - Ok(processed_data) => { - // Generate unique filename - let filename = format!("{}.jpg", Uuid::new_v4()); - let avatar_path = format!("{}/{}", state.config.files.avatars_dir, filename); - - // Save file - if let Err(e) = tokio::fs::write(&avatar_path, &processed_data).await { - eprintln!("Error saving avatar: {}", e); - return Json(UploadAvatarResponse { - success: false, - avatar_url: None, - message: "Error saving image".to_string(), - }); - } - - let avatar_url = format!("/avatars/{}", filename); - - // Update user in database - let result = sqlx::query("UPDATE users SET avatar_url = ? WHERE username = ?") - .bind(&avatar_url) - .bind(&username) - .execute(&state.db_pool) - .await; - - match result { - Ok(_) => { - return Json(UploadAvatarResponse { - success: true, - avatar_url: Some(avatar_url), - message: "Avatar uploaded successfully".to_string(), - }); - } - Err(e) => { - eprintln!("Error updating user avatar: {}", e); - return Json(UploadAvatarResponse { - success: false, - avatar_url: None, - message: "Database error".to_string(), - }); - } - } - } - Err(e) => { - return Json(UploadAvatarResponse { - success: false, - avatar_url: None, - message: e, - }); - } - } - } - } - - Json(UploadAvatarResponse { - success: false, - avatar_url: None, - message: "No avatar file provided".to_string(), - }) -} - -// Process and resize avatar -async fn process_avatar(data: &[u8]) -> Result, String> { - // Load image - let img = match image::load_from_memory(data) { - Ok(img) => img, - Err(_) => return Err("Invalid image format".to_string()), - }; - - // Resize image to 150x150 pixels - let resized = img.resize_to_fill(150, 150, image::imageops::FilterType::Lanczos3); - - // Convert to JPEG - let mut jpeg_data = Vec::new(); - match resized.write_to(&mut Cursor::new(&mut jpeg_data), image::ImageFormat::Jpeg) { - Ok(_) => Ok(jpeg_data), - Err(_) => Err("Error processing image".to_string()), - } -} - -async fn profile_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(UserProfile { - username: "user".to_string(), - phone: "+1234567890".to_string(), - avatar_url: None, - }); - }; - - // Get current user from token - let username = match verify_token(&state.db_pool, &token).await { - Ok(Some(user)) => user, - _ => { - return Json(UserProfile { - username: "user".to_string(), - phone: "+1234567890".to_string(), - avatar_url: None, - }); - } - }; - - // Get user details - let user_result = - sqlx::query("SELECT username, phone, avatar_url FROM users WHERE username = ?") - .bind(&username) - .fetch_optional(&state.db_pool) - .await; - - match user_result { - Ok(Some(row)) => Json(UserProfile { - username: row.get("username"), - phone: row.get("phone"), - avatar_url: row.get("avatar_url"), - }), - _ => Json(UserProfile { - username: "user".to_string(), - phone: "+1234567890".to_string(), - avatar_url: None, - }), - } -} - -// WebSocket handlers -async fn websocket_handler(ws: WebSocketUpgrade, State(state): State>) -> Response { - ws.on_upgrade(|socket| handle_websocket(socket, state)) -} - -async fn handle_websocket(socket: WebSocket, state: Arc) { - let (mut sender, mut receiver) = socket.split(); - let (tx, mut rx) = mpsc::unbounded_channel(); - - // 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" { - match verify_token(&state.db_pool, token).await { - Ok(Some(user_username)) => { - if state.users.read().await.contains_key(&user_username) { - let _ = sender - .send(Message::Text( - serde_json::json!({ - "type": "error", - "content": "User already connected" - }) - .to_string(), - )) - .await; - return; - } - - state.users.write().await.insert(user_username.clone(), tx); - - let _ = sender - .send(Message::Text( - serde_json::json!({ - "type": "system", - "content": "Connected successfully" - }) - .to_string(), - )) - .await; - - broadcast_user_list(&state.users).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; - } - } - - user_username - } - Ok(None) => { - let _ = sender - .send(Message::Text( - serde_json::json!({ - "type": "error", - "content": "Invalid or expired token" - }) - .to_string(), - )) - .await; - return; - } - Err(_) => { - let _ = sender - .send(Message::Text( - serde_json::json!({ - "type": "error", - "content": "Authentication error" - }) - .to_string(), - )) - .await; - return; - } - } - } else { - return; - } - } else { - return; - } - } - Err(_) => return, - } - } - _ => return, - }; - - println!("User {} connected", username); - - let state_clone = state.clone(); - let username_clone = username.clone(); - - let mut send_sender = sender; - let send_task = tokio::spawn(async move { - while let Some(message) = rx.recv().await { - if send_sender.send(message).await.is_err() { - break; - } - } - }); - - 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)) => { - if let Ok(mut chat_msg) = serde_json::from_str::(&text) { - chat_msg.from_user = username.clone(); - - if let Err(e) = save_message(&recv_state.db_pool, &chat_msg).await { - eprintln!("Failed to save message: {}", e); - continue; - } - - let users = recv_state.users.read().await; - if chat_msg.to_user != chat_msg.from_user { - if let Some(recipient_tx) = users.get(&chat_msg.to_user) { - let message_json = serde_json::to_string(&chat_msg).unwrap(); - let _ = recipient_tx.send(Message::Text(message_json)); - } - } - - 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 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(_)) => break, - Err(_) => break, - _ => {} - } - } - }); - - tokio::select! { - _ = send_task => {}, - _ = recv_task => {}, - } - - state_clone.users.write().await.remove(&username_clone); - broadcast_user_list(&state_clone.users).await; - println!("User {} disconnected", username_clone); -} - -// Utility functions -fn generate_token() -> String { - thread_rng() - .sample_iter(&Alphanumeric) - .take(32) - .map(char::from) - .collect() -} - -async fn verify_token(pool: &SqlitePool, token: &str) -> Result, sqlx::Error> { - let result = sqlx::query( - "SELECT username FROM sessions WHERE token = ? AND expires_at > datetime('now')", - ) - .bind(token) - .fetch_optional(pool) - .await?; - - Ok(result.map(|row| row.get("username"))) -} - -async fn broadcast_user_list(users: &UserMap) { - let user_list: Vec = users.read().await.keys().cloned().collect(); - let user_list_msg = serde_json::json!({ - "type": "user_list", - "users": user_list - }); - - for tx in users.read().await.values() { - let _ = tx.send(Message::Text(user_list_msg.to_string())); - } -} - -async fn save_message(pool: &SqlitePool, msg: &ChatMessage) -> Result<(), sqlx::Error> { - sqlx::query( - "INSERT OR REPLACE INTO messages (id, from_user, to_user, content, timestamp, message_type) VALUES (?, ?, ?, ?, ?, ?)", - ) - .bind(&msg.id) - .bind(&msg.from_user) - .bind(&msg.to_user) - .bind(&msg.content) - .bind(msg.timestamp) - .bind(&msg.message_type) - .execute(pool) - .await?; - - Ok(()) -} - -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 { - let message = ChatMessage { - 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"), - }; - messages.push(message); - } - - Ok(messages) -} - -async fn mark_messages_as_read( - pool: &SqlitePool, - from_user: &str, - to_user: &str, -) -> 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(()) -} diff --git a/src/model.rs b/src/model.rs new file mode 100644 index 0000000..90522a3 --- /dev/null +++ b/src/model.rs @@ -0,0 +1,93 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Serialize)] +pub struct AuthResponse { + pub success: bool, + pub message: String, + pub username: Option, + pub token: Option, +} + +#[derive(Debug, Deserialize)] +pub struct RegisterRequest { + pub phone: String, + pub username: String, + pub password: String, +} + +#[derive(Debug, Deserialize)] +pub struct LoginRequest { + pub phone: String, + pub password: String, +} + +#[derive(Debug, Serialize)] +pub struct ValidateTokenResponse { + pub success: bool, + pub username: Option, + pub phone: Option, + pub avatar_url: Option, + pub message: Option, +} + +#[derive(Debug, Deserialize)] +pub struct ValidateTokenRequest { + pub token: String, +} + +#[derive(Debug, Serialize)] +pub struct ChatPartner { + pub username: String, + pub phone: String, + pub avatar_url: Option, + pub last_message: Option, + pub last_message_time: Option, + pub unread_count: i64, +} + +#[derive(Debug, Deserialize)] +pub struct SearchQuery { + pub q: String, +} + +#[derive(Debug, Deserialize)] +pub struct MarkReadRequest { + pub from_user: String, + pub to_user: String, +} + +#[derive(Debug, Deserialize)] +pub struct DeleteMessageRequest { + pub message_id: String, + pub for_everyone: bool, +} + +#[derive(Debug, Serialize)] +pub struct DeleteMessageResponse { + pub success: bool, + pub message: String, +} + +#[derive(Debug, Serialize)] +pub struct UploadAvatarResponse { + pub success: bool, + pub avatar_url: Option, + pub message: String, +} + +#[derive(Debug, Serialize)] +pub struct UserProfile { + pub username: String, + pub phone: String, + pub avatar_url: Option, +} + +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct ChatMessage { + pub id: String, + pub from_user: String, + pub to_user: String, + pub content: String, + pub timestamp: i64, + pub message_type: String, +} diff --git a/src/state.rs b/src/state.rs new file mode 100644 index 0000000..18fa99b --- /dev/null +++ b/src/state.rs @@ -0,0 +1,14 @@ +use crate::conf; +use axum::extract::ws::Message; +use sqlx::sqlite::SqlitePool; +use std::{collections::HashMap, sync::Arc}; +use tokio::sync::{RwLock, mpsc}; + +pub type UserMap = Arc>>>; + +#[derive(Clone)] +pub struct AppState { + pub users: UserMap, + pub db_pool: SqlitePool, + pub config: conf::AppConfig, +} diff --git a/src/utils.rs b/src/utils.rs new file mode 100644 index 0000000..b75980e --- /dev/null +++ b/src/utils.rs @@ -0,0 +1,164 @@ +use crate::model; +use crate::state; +use axum::extract::ws::Message; +use rand::distributions::Alphanumeric; +use rand::{Rng, thread_rng}; +use sqlx::{Row, SqlitePool}; +use std::io::Cursor; + +// Utility functions +pub fn generate_token() -> String { + thread_rng() + .sample_iter(&Alphanumeric) + .take(32) + .map(char::from) + .collect() +} + +pub async fn verify_token(pool: &SqlitePool, token: &str) -> Result, sqlx::Error> { + let result = sqlx::query( + "SELECT username FROM sessions WHERE token = ? AND expires_at > datetime('now')", + ) + .bind(token) + .fetch_optional(pool) + .await?; + + Ok(result.map(|row| row.get("username"))) +} + +pub async fn mark_messages_as_read( + pool: &SqlitePool, + from_user: &str, + to_user: &str, +) -> 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(()) +} + +// Process and resize avatar +pub async fn process_avatar(data: &[u8]) -> Result, String> { + // Load image + let img = match image::load_from_memory(data) { + Ok(img) => img, + Err(_) => return Err("Invalid image format".to_string()), + }; + + // Resize image to 150x150 pixels + let resized = img.resize_to_fill(150, 150, image::imageops::FilterType::Lanczos3); + + // Convert to JPEG + let mut jpeg_data = Vec::new(); + match resized.write_to(&mut Cursor::new(&mut jpeg_data), image::ImageFormat::Jpeg) { + Ok(_) => Ok(jpeg_data), + Err(_) => Err("Error processing image".to_string()), + } +} + +pub 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 { + let message = model::ChatMessage { + 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"), + }; + messages.push(message); + } + + Ok(messages) +} + +pub async fn broadcast_user_list(users: &state::UserMap) { + let user_list: Vec = users.read().await.keys().cloned().collect(); + let user_list_msg = serde_json::json!({ + "type": "user_list", + "users": user_list + }); + + for tx in users.read().await.values() { + let _ = tx.send(Message::Text(user_list_msg.to_string())); + } +} + +pub async fn save_message(pool: &SqlitePool, msg: &model::ChatMessage) -> Result<(), sqlx::Error> { + sqlx::query( + "INSERT OR REPLACE INTO messages (id, from_user, to_user, content, timestamp, message_type) VALUES (?, ?, ?, ?, ?, ?)", + ) + .bind(&msg.id) + .bind(&msg.from_user) + .bind(&msg.to_user) + .bind(&msg.content) + .bind(msg.timestamp) + .bind(&msg.message_type) + .execute(pool) + .await?; + + Ok(()) +}