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

fix ci/cd
Continuous integration / Check (push) Successful in 32s
Details
Continuous integration / Rustfmt (push) Successful in 26s
Details
Continuous integration / build (push) Successful in 29s
Details
Continuous integration / build-release (push) Successful in 57s
Details
Continuous integration / Check (pull_request) Successful in 33s
Details
Continuous integration / Rustfmt (pull_request) Successful in 30s
Details
Continuous integration / build (pull_request) Successful in 31s
Details
Continuous integration / build-release (pull_request) Successful in 56s
Details

Browse Source
This commit is contained in:
wt
2025-11-27 18:20:12 +07:00
parent 5a26ac8690
commit ab25ef7ba8
1 changed files with 290 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.rs
+290 -234
View File
@@ -1,29 +1,25 @@
use axum::{
Router,
extract::{
Json, Multipart, Query, State,
ws::{Message, WebSocket, WebSocketUpgrade},
State, Json, Query, Multipart,
},
response::Response,
routing::{get, post, delete},
Router,
routing::{get, post},
};
use bcrypt::{DEFAULT_COST, hash, verify};
use chrono::{Duration, Utc};
use config::{Config as ConfigLoader, File, FileFormat};
use serde::{Deserialize, Serialize};
use sqlx::{sqlite::SqlitePool, Row};
use std::{
collections::HashMap,
net::SocketAddr,
sync::Arc,
};
use tokio::sync::{mpsc, RwLock};
use tower_http::services::ServeDir;
use futures_util::{StreamExt, SinkExt};
use bcrypt::{hash, verify, DEFAULT_COST};
use futures_util::{SinkExt, StreamExt};
use rand::distributions::Alphanumeric;
use rand::{thread_rng, Rng};
use chrono::{Utc, Duration};
use uuid::Uuid;
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 tower_http::services::ServeDir;
use uuid::Uuid;
// Types for shared state
type UserMap = Arc<RwLock<HashMap<String, mpsc::UnboundedSender<Message>>>>;
@@ -58,7 +54,7 @@ struct FilesConfig {
#[derive(Debug, Deserialize, Clone)]
struct LimitsConfig {
max_message_length: usize,
// max_message_length: usize,
max_username_length: usize,
min_password_length: usize,
}
@@ -114,10 +110,10 @@ struct ChatMessage {
message_type: String,
}
#[derive(Debug, Serialize, Deserialize)]
struct UserList {
users: Vec<String>,
}
// #[derive(Debug, Serialize, Deserialize)]
// struct UserList {
// users: Vec<String>,
// }
// Структуры для аутентификации
#[derive(Debug, Deserialize)]
@@ -243,17 +239,17 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let addr = format!("{}:{}", state.config.server.host, state.config.server.port)
.parse::<SocketAddr>()?;
println!("🚀 Server running on http://{}", addr);
println!("📁 Web directory: {}", state.config.files.web_dir);
println!("💾 Database: {}", state.config.database.url);
let listener = tokio::net::TcpListener::bind(addr).await?;
if let Some(max_connections) = state.config.server.max_connections {
println!("🔒 Max connections: {}", max_connections);
}
axum::serve(listener, app).await?;
Ok(())
@@ -331,10 +327,15 @@ async fn register_handler(
});
}
if payload.username.len() < 3 || payload.username.len() > state.config.limits.max_username_length {
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),
message: format!(
"Username must be between 3 and {} characters",
state.config.limits.max_username_length
),
username: None,
token: None,
});
@@ -343,7 +344,10 @@ async fn register_handler(
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),
message: format!(
"Password must be at least {} characters",
state.config.limits.min_password_length
),
username: None,
token: None,
});
@@ -363,24 +367,22 @@ async fn register_handler(
};
// 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;
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 (?, ?, ?, ?)"
"INSERT INTO sessions (token, username, phone, expires_at) VALUES (?, ?, ?, ?)",
)
.bind(&token)
.bind(&payload.username)
@@ -397,9 +399,15 @@ async fn register_handler(
})
}
Err(e) => {
let error_message = if e.to_string().contains("UNIQUE constraint failed: users.phone") {
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") {
} else if e
.to_string()
.contains("UNIQUE constraint failed: users.username")
{
"Username already taken".to_string()
} else {
"Registration failed".to_string()
@@ -420,12 +428,10 @@ async fn login_handler(
Json(payload): Json<LoginRequest>,
) -> Json<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;
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)) => {
@@ -438,7 +444,7 @@ async fn login_handler(
// 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 (?, ?, ?, ?)"
@@ -493,12 +499,10 @@ async fn validate_token_handler(
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;
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)) => {
@@ -568,11 +572,11 @@ async fn debug_messages_handler(
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 = ?
FROM messages
WHERE from_user = ? OR to_user = ?
ORDER BY timestamp DESC
LIMIT 100
"#
"#,
)
.bind(&username)
.bind(&username)
@@ -582,15 +586,15 @@ async fn debug_messages_handler(
// Get chat partners count
let partners_result = sqlx::query(
r#"
SELECT COUNT(DISTINCT
CASE
WHEN from_user = ? THEN to_user
ELSE from_user
SELECT COUNT(DISTINCT
CASE
WHEN from_user = ? THEN to_user
ELSE from_user
END
) as partner_count
FROM messages
FROM messages
WHERE from_user = ? OR to_user = ?
"#
"#,
)
.bind(&username)
.bind(&username)
@@ -617,12 +621,12 @@ async fn debug_messages_handler(
"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(),
@@ -630,9 +634,7 @@ async fn debug_messages_handler(
"messages": result
}))
}
Err(e) => {
Json(serde_json::json!({"error": format!("Database error: {}", e)}))
}
Err(e) => Json(serde_json::json!({"error": format!("Database error: {}", e)})),
}
}
@@ -660,12 +662,12 @@ async fn get_chats_handler(
let chat_partners = sqlx::query(
r#"
-- Находим всех уникальных собеседников
SELECT DISTINCT
CASE
WHEN from_user = ? THEN to_user
ELSE from_user
SELECT DISTINCT
CASE
WHEN from_user = ? THEN to_user
ELSE from_user
END as partner_username
FROM messages
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))
@@ -681,7 +683,10 @@ async fn get_chats_handler(
let partner_usernames = match chat_partners {
Ok(rows) => {
let partners: Vec<String> = rows.iter().map(|row| row.get::<String, _>("partner_username")).collect();
let partners: Vec<String> = rows
.iter()
.map(|row| row.get::<String, _>("partner_username"))
.collect();
println!("🤝 Found partner usernames: {:?}", partners);
partners
}
@@ -693,7 +698,7 @@ async fn get_chats_handler(
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"
@@ -702,49 +707,51 @@ async fn get_chats_handler(
.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);
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;
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
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
ORDER BY timestamp DESC
LIMIT 1
"#
)
@@ -761,7 +768,10 @@ async fn get_chats_handler(
Ok(Some(row)) => {
let content: String = row.get("content");
let timestamp: i64 = row.get("timestamp");
println!("💬 Last message with {}: '{}' at {}", partner_name, content, timestamp);
println!(
"💬 Last message with {}: '{}' at {}",
partner_name, content, timestamp
);
(Some(content), Some(timestamp))
}
Ok(None) => {
@@ -799,30 +809,40 @@ async fn get_chats_handler(
last_message_time,
unread_count,
};
println!("✅ Added chat with {} (phone: {}), last_message: {:?}, unread: {}",
chat.username, chat.phone, chat.last_message, chat.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);
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))
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);
println!(
" - {}: {:?} (unread: {})",
chat.username, chat.last_message, chat.unread_count
);
}
Json(chats)
}
@@ -830,35 +850,35 @@ async fn get_chats_handler(
async fn search_users_handler(
State(state): State<Arc<AppState>>,
Query(params): Query<SearchQuery>,
headers: axum::http::HeaderMap,
// headers: axum::http::HeaderMap,
) -> Json<Vec<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());
};
// 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 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 ?)
AND username != ?
SELECT username, phone, avatar_url
FROM users
WHERE (username LIKE ? OR phone LIKE ?)
LIMIT 20
"#
"#,
)
// AND username != ?
.bind(&search_pattern)
.bind(&search_pattern)
.bind(&current_user)
// .bind(&current_user)
.fetch_all(&state.db_pool)
.await;
@@ -936,19 +956,19 @@ async fn delete_message_handler(
// 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(),
}),
_ => {
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_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,
@@ -987,12 +1007,10 @@ async fn delete_message_handler(
});
}
let result = sqlx::query(
"UPDATE messages SET deleted_for_everyone = TRUE WHERE id = ?"
)
.bind(&payload.message_id)
.execute(&state.db_pool)
.await;
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(_) => {
@@ -1028,8 +1046,11 @@ async fn delete_message_handler(
"deleted_for_receiver"
};
let query = format!("UPDATE messages SET {} = TRUE WHERE id = ?", field_to_update);
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)
@@ -1071,26 +1092,31 @@ async fn upload_avatar_handler(
// 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(),
}),
_ => {
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),
message: format!(
"Image too large (max {} bytes)",
state.config.files.max_avatar_size
),
});
}
@@ -1100,7 +1126,7 @@ async fn upload_avatar_handler(
// 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);
@@ -1110,18 +1136,16 @@ async fn upload_avatar_handler(
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;
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 {
@@ -1168,7 +1192,7 @@ async fn process_avatar(data: &[u8]) -> Result<Vec<u8>, 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) {
@@ -1195,29 +1219,28 @@ async fn profile_handler(
// 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,
}),
_ => {
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;
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"),
})
}
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(),
@@ -1227,77 +1250,95 @@ async fn profile_handler(
}
// WebSocket handlers
async fn websocket_handler(
ws: WebSocketUpgrade,
State(state): State<Arc<AppState>>,
) -> Response {
async fn websocket_handler(ws: WebSocketUpgrade, State(state): State<Arc<AppState>>) -> Response {
ws.on_upgrade(|socket| handle_websocket(socket, state))
}
async fn handle_websocket(socket: WebSocket, state: Arc<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 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;
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;
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 {
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;
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;
let _ = sender
.send(Message::Text(
serde_json::json!({
"type": "error",
"content": "Authentication error"
})
.to_string(),
))
.await;
return;
}
}
} else { return; }
} else { return; }
} else {
return;
}
} else {
return;
}
}
Err(_) => return,
}
@@ -1326,7 +1367,7 @@ async fn handle_websocket(socket: WebSocket, state: Arc<AppState>) {
Ok(Message::Text(text)) => {
if let Ok(mut chat_msg) = serde_json::from_str::<ChatMessage>(&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;
@@ -1342,20 +1383,22 @@ async fn handle_websocket(socket: WebSocket, state: Arc<AppState>) {
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::<serde_json::Value>(&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
}
}
}
} // else if let Ok(delete_msg) = serde_json::from_str::<serde_json::Value>(&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,
@@ -1385,7 +1428,7 @@ fn generate_token() -> String {
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')"
"SELECT username FROM sessions WHERE token = ? AND expires_at > datetime('now')",
)
.bind(token)
.fetch_optional(pool)
@@ -1422,17 +1465,21 @@ async fn save_message(pool: &SqlitePool, msg: &ChatMessage) -> Result<(), sqlx::
Ok(())
}
async fn get_message_history(pool: &SqlitePool, username: &str, other_user: &str) -> Result<Vec<ChatMessage>, sqlx::Error> {
async fn get_message_history(
pool: &SqlitePool,
username: &str,
other_user: &str,
) -> Result<Vec<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
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))
@@ -1441,8 +1488,8 @@ async fn get_message_history(pool: &SqlitePool, username: &str, other_user: &str
} else {
// Получаем сообщения только с конкретным пользователем (исключая удаленные)
r#"
SELECT id, from_user, to_user, content, timestamp, message_type
FROM messages
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))
@@ -1486,7 +1533,11 @@ async fn get_message_history(pool: &SqlitePool, username: &str, other_user: &str
Ok(messages)
}
async fn mark_messages_as_read(pool: &SqlitePool, from_user: &str, to_user: &str) -> Result<(), sqlx::Error> {
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"
)
@@ -1494,7 +1545,12 @@ async fn mark_messages_as_read(pool: &SqlitePool, from_user: &str, to_user: &str
.bind(to_user)
.execute(pool)
.await?;
println!("✅ Marked {} messages as read from {} to {}", result.rows_affected(), from_user, to_user);
println!(
"✅ Marked {} messages as read from {} to {}",
result.rows_affected(),
from_user,
to_user
);
Ok(())
}
}