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

vorking with chats

Browse Source
This commit is contained in:
wt committed 2025-11-23 19:01:04 +07:00
1 parent e03af7466b
commit d9d3dd2de9
2 files changed
+1081 -503
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/main.rs
+627 -58
View File
@@ -1,10 +1,10 @@
use axum::{
extract::{
ws::{Message, WebSocket, WebSocketUpgrade},
State, Json,
State, Json, Query,
},
response::Response,
routing::{get, post},
routing::{get, post, delete},
Router,
};
use serde::{Deserialize, Serialize};
@@ -82,6 +82,38 @@ struct ValidateTokenResponse {
message: Option<String>,
}
#[derive(Debug, Serialize)]
struct ChatPartner {
username: String,
phone: String,
last_message: Option<String>,
last_message_time: Option<i64>,
unread_count: i64,
}
#[derive(Debug, Deserialize)]
struct SearchQuery {
q: String,
}
#[derive(Debug, Deserialize)]
struct MarkReadRequest {
from_user: String,
to_user: String,
}
#[derive(Debug, Deserialize)]
struct DeleteMessageRequest {
message_id: String,
for_everyone: bool,
}
#[derive(Debug, Serialize)]
struct DeleteMessageResponse {
success: bool,
message: String,
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Initialize logging
@@ -102,11 +134,16 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
.route("/api/register", post(register_handler))
.route("/api/login", post(login_handler))
.route("/api/validate-token", post(validate_token_handler))
.route("/api/chats", get(get_chats_handler))
.route("/api/search-users", get(search_users_handler))
.route("/api/debug-messages", get(debug_messages_handler))
.route("/api/mark-read", post(mark_read_handler))
.route("/api/delete-message", post(delete_message_handler))
.route("/api/profile", get(profile_handler))
.nest_service("/", ServeDir::new("web"))
.with_state(state);
let addr = SocketAddr::from(([127, 0, 0, 1], 3333));
let addr = SocketAddr::from(([0, 0, 0, 0], 3333));
println!("Server running on http://{}", addr);
let listener = tokio::net::TcpListener::bind(addr).await?;
@@ -142,7 +179,11 @@ async fn init_db() -> Result<SqlitePool, Box<dyn std::error::Error>> {
to_user TEXT NOT NULL,
content TEXT NOT NULL,
timestamp INTEGER NOT NULL,
message_type TEXT NOT NULL
message_type TEXT NOT NULL,
read BOOLEAN DEFAULT FALSE,
deleted_for_sender BOOLEAN DEFAULT FALSE,
deleted_for_receiver BOOLEAN DEFAULT FALSE,
deleted_for_everyone BOOLEAN DEFAULT FALSE
)
"#,
)
@@ -395,9 +436,510 @@ async fn validate_token_handler(
}
}
// Debug endpoint to see all messages
async fn debug_messages_handler(
State(state): State<Arc<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 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)}))
}
}
}
// УПРОЩЕННАЯ версия получения чатов
async fn get_chats_handler(
State(state): State<Arc<AppState>>,
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());
};
// 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<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 FROM users WHERE username = ?"
)
.bind(&partner_username)
.fetch_optional(&state.db_pool)
.await;
match user_result {
Ok(Some(user_row)) => {
let partner_name: String = user_row.get("username");
let phone: String = user_row.get("phone");
// Получаем последнее сообщение
let last_message_result = sqlx::query(
r#"
SELECT content, timestamp
FROM messages
WHERE ((from_user = ? AND to_user = ?) OR (from_user = ? AND to_user = ?))
AND deleted_for_everyone = FALSE
AND ((from_user = ? AND deleted_for_sender = FALSE) OR (to_user = ? AND deleted_for_receiver = FALSE))
ORDER BY timestamp DESC
LIMIT 1
"#
)
.bind(&username) // текущий пользователь
.bind(&partner_name) // собеседник
.bind(&partner_name) // собеседник
.bind(&username) // текущий пользователь
.bind(&username) // для проверки deleted_for_sender
.bind(&username) // для проверки deleted_for_receiver
.fetch_optional(&state.db_pool)
.await;
let (last_message, last_message_time) = match last_message_result {
Ok(Some(row)) => {
let content: String = row.get("content");
let timestamp: i64 = row.get("timestamp");
println!("💬 Last message with {}: '{}' at {}", partner_name, content, timestamp);
(Some(content), Some(timestamp))
}
Ok(None) => {
println!("❌ No last message found for chat with {}", partner_name);
(None, None)
}
Err(e) => {
eprintln!("❌ Error fetching last message: {}", e);
(None, None)
}
};
// Получаем количество непрочитанных сообщений
let unread_count_result = sqlx::query(
"SELECT COUNT(*) as count FROM messages WHERE from_user = ? AND to_user = ? AND read = FALSE AND deleted_for_everyone = FALSE AND deleted_for_receiver = FALSE"
)
.bind(&partner_name)
.bind(&username)
.fetch_one(&state.db_pool)
.await;
let unread_count = match unread_count_result {
Ok(row) => row.get::<i64, _>("count"),
Err(e) => {
eprintln!("❌ Error counting unread messages: {}", e);
0
}
};
let chat = ChatPartner {
username: partner_name.clone(),
phone,
last_message,
last_message_time,
unread_count,
};
println!("✅ Added chat with {} (phone: {}), last_message: {:?}, unread: {}",
chat.username, chat.phone, chat.last_message, chat.unread_count);
chats.push(chat);
}
Ok(None) => {
eprintln!("❌ User not found: {}", partner_username);
}
Err(e) => {
eprintln!("❌ Error fetching user details for {}: {}", partner_username, e);
}
}
}
// Сортируем по времени последнего сообщения (новые сверху)
chats.sort_by(|a, b| {
b.last_message_time.unwrap_or(0).cmp(&a.last_message_time.unwrap_or(0))
});
println!("🎉 Final chat list for {}: {} chats", username, chats.len());
for chat in &chats {
println!(" - {}: {:?} (unread: {})", chat.username, chat.last_message, chat.unread_count);
}
Json(chats)
}
// New endpoint to search users
async fn search_users_handler(
State(state): State<Arc<AppState>>,
Query(params): Query<SearchQuery>,
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());
};
// Get current user from token
let current_user = match verify_token(&state.db_pool, &token).await {
Ok(Some(user)) => user,
_ => return Json(Vec::new()),
};
let search_pattern = format!("%{}%", params.q);
let users = sqlx::query(
r#"
SELECT username, phone
FROM users
WHERE (username LIKE ? OR phone LIKE ?)
AND username != ?
LIMIT 20
"#
)
.bind(&search_pattern)
.bind(&search_pattern)
.bind(&current_user)
.fetch_all(&state.db_pool)
.await;
match users {
Ok(rows) => {
let mut results = Vec::new();
for row in rows {
results.push(ChatPartner {
username: row.get("username"),
phone: row.get("phone"),
last_message: None,
last_message_time: None,
unread_count: 0,
});
}
Json(results)
}
Err(e) => {
eprintln!("Error searching users: {}", e);
Json(Vec::new())
}
}
}
// Mark messages as read endpoint
async fn mark_read_handler(
State(state): State<Arc<AppState>>,
headers: axum::http::HeaderMap,
Json(payload): Json<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 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<Arc<AppState>>,
headers: axum::http::HeaderMap,
Json(payload): Json<DeleteMessageRequest>,
) -> Json<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(DeleteMessageResponse {
success: false,
message: "No token".to_string(),
});
};
// Get current user from token
let username = match verify_token(&state.db_pool, &token).await {
Ok(Some(user)) => user,
_ => return Json(DeleteMessageResponse {
success: false,
message: "Invalid token".to_string(),
}),
};
// Получаем информацию о сообщении
let message_result = sqlx::query(
"SELECT from_user, to_user FROM messages WHERE id = ?"
)
.bind(&payload.message_id)
.fetch_optional(&state.db_pool)
.await;
let message = match message_result {
Ok(Some(row)) => row,
Ok(None) => {
return Json(DeleteMessageResponse {
success: false,
message: "Message not found".to_string(),
});
}
Err(e) => {
eprintln!("Error fetching message: {}", e);
return Json(DeleteMessageResponse {
success: false,
message: "Database error".to_string(),
});
}
};
let from_user: String = message.get("from_user");
let to_user: String = message.get("to_user");
// Проверяем права доступа
if from_user != username && to_user != username {
return Json(DeleteMessageResponse {
success: false,
message: "You don't have permission to delete this message".to_string(),
});
}
if payload.for_everyone {
// Удаление для всех
if from_user != username {
return Json(DeleteMessageResponse {
success: false,
message: "Only the sender can delete message for everyone".to_string(),
});
}
let result = sqlx::query(
"UPDATE messages SET deleted_for_everyone = TRUE WHERE id = ?"
)
.bind(&payload.message_id)
.execute(&state.db_pool)
.await;
match result {
Ok(_) => {
// Отправляем уведомление через WebSocket
let users = state.users.read().await;
if let Some(recipient_tx) = users.get(&to_user) {
let delete_msg = serde_json::json!({
"type": "message_deleted",
"message_id": payload.message_id,
"for_everyone": true
});
let _ = recipient_tx.send(Message::Text(delete_msg.to_string()));
}
Json(DeleteMessageResponse {
success: true,
message: "Message deleted for everyone".to_string(),
})
}
Err(e) => {
eprintln!("Error deleting message for everyone: {}", e);
Json(DeleteMessageResponse {
success: false,
message: "Database error".to_string(),
})
}
}
} else {
// Удаление только для себя
let field_to_update = if username == from_user {
"deleted_for_sender"
} else {
"deleted_for_receiver"
};
let query = format!("UPDATE messages SET {} = TRUE WHERE id = ?", field_to_update);
let result = sqlx::query(&query)
.bind(&payload.message_id)
.execute(&state.db_pool)
.await;
match result {
Ok(_) => Json(DeleteMessageResponse {
success: true,
message: "Message deleted for you".to_string(),
}),
Err(e) => {
eprintln!("Error deleting message for user: {}", e);
Json(DeleteMessageResponse {
success: false,
message: "Database error".to_string(),
})
}
}
}
}
async fn profile_handler() -> Json<UserProfile> {
// This would normally verify the token and return user profile
// For now, return a placeholder
Json(UserProfile {
username: "user".to_string(),
phone: "+1234567890".to_string(),
@@ -413,23 +955,18 @@ async fn websocket_handler(
}
async fn handle_websocket(socket: WebSocket, state: Arc<AppState>) {
// Split the socket into sender and receiver
let (mut sender, mut receiver) = socket.split();
// Create channel for sending messages to this client
let (tx, mut rx) = mpsc::unbounded_channel();
// Handle authentication - receive first message with token
// Handle authentication
let username = match receiver.next().await {
Some(Ok(Message::Text(auth_msg))) => {
match serde_json::from_str::<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" {
// Verify token and get username
match verify_token(&state.db_pool, token).await {
Ok(Some(user_username)) => {
// Check if username is already connected
if state.users.read().await.contains_key(&user_username) {
let _ = sender.send(Message::Text(
serde_json::json!({
@@ -440,10 +977,8 @@ async fn handle_websocket(socket: WebSocket, state: Arc<AppState>) {
return;
}
// Register user
state.users.write().await.insert(user_username.clone(), tx);
// Send success message
let _ = sender.send(Message::Text(
serde_json::json!({
"type": "system",
@@ -451,11 +986,9 @@ async fn handle_websocket(socket: WebSocket, state: Arc<AppState>) {
}).to_string(),
)).await;
// Send user list
broadcast_user_list(&state.users).await;
// Send message history
if let Ok(messages) = get_message_history(&state.db_pool, &user_username).await {
if let Ok(messages) = get_message_history(&state.db_pool, &user_username, "").await {
for msg in messages {
let message_json = serde_json::to_string(&msg).unwrap();
let _ = sender.send(Message::Text(message_json)).await;
@@ -483,12 +1016,8 @@ async fn handle_websocket(socket: WebSocket, state: Arc<AppState>) {
return;
}
}
} else {
return;
}
} else {
return;
}
} else { return; }
} else { return; }
}
Err(_) => return,
}
@@ -498,11 +1027,9 @@ async fn handle_websocket(socket: WebSocket, state: Arc<AppState>) {
println!("User {} connected", username);
// Clone state for cleanup
let state_clone = state.clone();
let username_clone = username.clone();
// Spawn a task to handle sending messages to the client
let mut send_sender = sender;
let send_task = tokio::spawn(async move {
while let Some(message) = rx.recv().await {
@@ -512,66 +1039,56 @@ async fn handle_websocket(socket: WebSocket, state: Arc<AppState>) {
}
});
// Handle receiving messages from the client in the main task
let recv_state = state.clone();
let recv_task = tokio::spawn(async move {
while let Some(result) = receiver.next().await {
match result {
Ok(Message::Text(text)) => {
println!("Received message from {}: {}", username, text);
if let Ok(mut chat_msg) = serde_json::from_str::<ChatMessage>(&text) {
// Set the from_user to the authenticated username
chat_msg.from_user = username.clone();
println!("Processing message from {} to {}", chat_msg.from_user, chat_msg.to_user);
// Save message to database
if let Err(e) = save_message(&recv_state.db_pool, &chat_msg).await {
eprintln!("Failed to save message: {}", e);
continue;
}
// Send to recipient if online
let users = recv_state.users.read().await;
if let Some(recipient_tx) = users.get(&chat_msg.to_user) {
let message_json = serde_json::to_string(&chat_msg).unwrap();
println!("Sending message to {}: {}", chat_msg.to_user, message_json);
let _ = recipient_tx.send(Message::Text(message_json));
} else {
println!("Recipient {} not found online", chat_msg.to_user);
}
// Also send the message back to the sender for confirmation
if let Some(sender_tx) = users.get(&username) {
let message_json = serde_json::to_string(&chat_msg).unwrap();
let _ = sender_tx.send(Message::Text(message_json));
}
} else {
println!("Failed to parse message as ChatMessage: {}", text);
} else if let Ok(delete_msg) = serde_json::from_str::<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(_)) => {
println!("WebSocket closed by client");
break;
}
Err(e) => {
eprintln!("WebSocket error: {}", e);
break;
}
Ok(Message::Close(_)) => break,
Err(_) => break,
_ => {}
}
}
});
// Wait for both tasks to finish
tokio::select! {
_ = send_task => {},
_ = recv_task => {},
}
// Cleanup
state_clone.users.write().await.remove(&username_clone);
broadcast_user_list(&state_clone.users).await;
println!("User {} disconnected", username_clone);
@@ -625,14 +1142,53 @@ async fn save_message(pool: &SqlitePool, msg: &ChatMessage) -> Result<(), sqlx::
Ok(())
}
async fn get_message_history(pool: &SqlitePool, username: &str) -> Result<Vec<ChatMessage>, sqlx::Error> {
let rows = sqlx::query(
"SELECT id, from_user, to_user, content, timestamp, message_type FROM messages WHERE from_user = ? OR to_user = ? ORDER BY timestamp"
)
.bind(username)
.bind(username)
.fetch_all(pool)
.await?;
async fn get_message_history(pool: &SqlitePool, username: &str, other_user: &str) -> Result<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
WHERE (from_user = ? OR to_user = ?)
AND deleted_for_everyone = FALSE
AND ((from_user = ? AND deleted_for_sender = FALSE) OR (to_user = ? AND deleted_for_receiver = FALSE))
ORDER BY timestamp
"#
} else {
// Получаем сообщения только с конкретным пользователем (исключая удаленные)
r#"
SELECT id, from_user, to_user, content, timestamp, message_type
FROM messages
WHERE ((from_user = ? AND to_user = ?) OR (from_user = ? AND to_user = ?))
AND deleted_for_everyone = FALSE
AND ((from_user = ? AND deleted_for_sender = FALSE) OR (to_user = ? AND deleted_for_receiver = FALSE))
ORDER BY timestamp
"#
};
let rows = if other_user.is_empty() {
sqlx::query(query)
.bind(username)
.bind(username)
.bind(username)
.bind(username)
.fetch_all(pool)
.await?
} else {
sqlx::query(query)
.bind(username)
.bind(other_user)
.bind(other_user)
.bind(username)
.bind(username)
.bind(username)
.fetch_all(pool)
.await?
};
let mut messages = Vec::new();
for row in rows {
@@ -648,4 +1204,17 @@ async fn get_message_history(pool: &SqlitePool, username: &str) -> Result<Vec<Ch
}
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(())
}
web/index.html
+454 -445
View File
File diff suppressed because it is too large. Load diff