wt/msgs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

more refactor on backend

Browse Source
This commit is contained in:
wt
2025-11-28 01:38:59 +07:00
parent 941afa2485
commit 7f7936d37f
20 changed files with 1603 additions and 1527 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.json
+19 -19
View File
@@ -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
}
}
"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
}
}
src/api.rs
+9
View File
@@ -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;
src/api/chats.rs
+214
View File
@@ -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<Arc<state::AppState>>,
headers: axum::http::HeaderMap,
) -> Json<Vec<model::ChatPartner>> {
// 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<String> = rows
.iter()
.map(|row| row.get::<String, _>("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<String> = 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::<i64, _>("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)
}
src/api/debug.rs
+94
View File
@@ -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<Arc<state::AppState>>,
headers: axum::http::HeaderMap,
) -> Json<serde_json::Value> {
// 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::<String, _>("id"),
"from_user": row.get::<String, _>("from_user"),
"to_user": row.get::<String, _>("to_user"),
"content": row.get::<String, _>("content"),
"timestamp": row.get::<i64, _>("timestamp"),
"message_type": row.get::<String, _>("message_type"),
"read": row.get::<bool, _>("read"),
"deleted_for_sender": row.get::<bool, _>("deleted_for_sender"),
"deleted_for_receiver": row.get::<bool, _>("deleted_for_receiver"),
"deleted_for_everyone": row.get::<bool, _>("deleted_for_everyone"),
"is_current_user_from": row.get::<String, _>("from_user") == username,
"is_current_user_to": row.get::<String, _>("to_user") == username,
}));
}
let partner_count = match partners_result {
Ok(row) => row.get::<i64, _>("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)})),
}
}
src/api/login.rs
+77
View File
@@ -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<Arc<state::AppState>>,
Json(payload): Json<model::LoginRequest>,
) -> Json<model::AuthResponse> {
// 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,
}),
}
}
src/api/messages.rs
+39
View File
@@ -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<Arc<state::AppState>>,
headers: axum::http::HeaderMap,
Json(payload): Json<model::MarkReadRequest>,
) -> Json<serde_json::Value> {
// 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"}))
}
}
}
src/api/messages/delete.rs
+141
View File
@@ -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<Arc<state::AppState>>,
headers: axum::http::HeaderMap,
Json(payload): Json<model::DeleteMessageRequest>,
) -> Json<model::DeleteMessageResponse> {
// 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(),
})
}
}
}
}
src/api/profile.rs
+2
View File
@@ -0,0 +1,2 @@
pub mod avatar;
pub mod get;
src/api/profile/avatar.rs
+116
View File
@@ -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<Arc<state::AppState>>,
headers: axum::http::HeaderMap,
mut multipart: Multipart,
) -> Json<model::UploadAvatarResponse> {
// 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(),
})
}
src/api/profile/get.rs
+54
View File
@@ -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<Arc<state::AppState>>,
headers: axum::http::HeaderMap,
) -> Json<model::UserProfile> {
// 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,
}),
}
}
src/api/register.rs
+117
View File
@@ -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<Arc<state::AppState>>,
Json(payload): Json<model::RegisterRequest>,
) -> Json<model::AuthResponse> {
// 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,
})
}
}
}
src/api/search.rs
+48
View File
@@ -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<Arc<state::AppState>>,
Query(params): Query<model::SearchQuery>,
// headers: axum::http::HeaderMap,
) -> Json<Vec<model::ChatPartner>> {
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())
}
}
}
src/api/security.rs
+63
View File
@@ -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<Arc<state::AppState>>,
Json(payload): Json<model::ValidateTokenRequest>,
) -> Json<model::ValidateTokenResponse> {
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<String> = 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()),
}),
}
}
src/api/websock.rs
+174
View File
@@ -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<Arc<state::AppState>>,
) -> Response {
ws.on_upgrade(|socket| handle_websocket(socket, state))
}
async fn handle_websocket(socket: WebSocket, state: Arc<state::AppState>) {
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::<serde_json::Value>(&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::<model::ChatMessage>(&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);
}
src/conf.rs
+71
View File
@@ -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<u32>,
}
#[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<Self, Box<dyn std::error::Error>> {
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()?)
}
}
src/db.rs
+61
View File
@@ -0,0 +1,61 @@
use sqlx::{SqlitePool, sqlite::SqliteConnectOptions};
pub async fn init_db(database_url: &str) -> Result<SqlitePool, Box<dyn std::error::Error>> {
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)
}
src/main.rs
+33 -1508
View File
File diff suppressed because it is too large Load Diff
src/model.rs
+93
View File
@@ -0,0 +1,93 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize)]
pub struct AuthResponse {
pub success: bool,
pub message: String,
pub username: Option<String>,
pub token: Option<String>,
}
#[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<String>,
pub phone: Option<String>,
pub avatar_url: Option<String>,
pub message: Option<String>,
}
#[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<String>,
pub last_message: Option<String>,
pub last_message_time: Option<i64>,
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<String>,
pub message: String,
}
#[derive(Debug, Serialize)]
pub struct UserProfile {
pub username: String,
pub phone: String,
pub avatar_url: Option<String>,
}
#[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,
}
src/state.rs
+14
View File
@@ -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<RwLock<HashMap<String, mpsc::UnboundedSender<Message>>>>;
#[derive(Clone)]
pub struct AppState {
pub users: UserMap,
pub db_pool: SqlitePool,
pub config: conf::AppConfig,
}
src/utils.rs
+164
View File
@@ -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<Option<String>, 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<Vec<u8>, 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<Vec<model::ChatMessage>, 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<String> = 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(())
}