diff --git a/backend/app/main.py b/backend/app/main.py index 34c21b0..8744fb8 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -9,6 +9,7 @@ from .routers import auth, rooms, tracks, websocket, messages from .database import async_session from .models.room import Room from .services.sync import manager +from .services.ping_task import ping_users_task from .config import get_settings # Setup logging @@ -62,15 +63,22 @@ async def periodic_sync(): @asynccontextmanager async def lifespan(app: FastAPI): - # Start background sync task + # Start background tasks sync_task = asyncio.create_task(periodic_sync()) + ping_task = asyncio.create_task(ping_users_task()) + logger.info("Background tasks started: periodic_sync, ping_users_task") yield # Cleanup sync_task.cancel() + ping_task.cancel() try: await sync_task except asyncio.CancelledError: pass + try: + await ping_task + except asyncio.CancelledError: + pass app = FastAPI(title="EnigFM", description="Listen to music together with friends", lifespan=lifespan) diff --git a/backend/app/routers/rooms.py b/backend/app/routers/rooms.py index 629949a..f63c8dc 100644 --- a/backend/app/routers/rooms.py +++ b/backend/app/routers/rooms.py @@ -106,7 +106,25 @@ async def delete_room( if room.owner_id != current_user.id: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not room owner") + # Уведомляем всех подключенных пользователей о удалении комнаты + await manager.broadcast_to_room( + room_id, + {"type": "room_deleted", "message": "Room has been deleted by owner"}, + ) + + # Отключаем всех пользователей от WebSocket + if room_id in manager.active_connections: + connections = list(manager.active_connections[room_id]) + for websocket, user_id in connections: + try: + await websocket.close(code=1000, reason="Room deleted") + except Exception: + pass + manager.disconnect(websocket, room_id, user_id) + + # Удаляем комнату (cascade delete удалит participants, queue, messages) await db.delete(room) + return {"status": "deleted"} diff --git a/backend/app/routers/tracks.py b/backend/app/routers/tracks.py index e8e1478..655a74e 100644 --- a/backend/app/routers/tracks.py +++ b/backend/app/routers/tracks.py @@ -3,12 +3,13 @@ from urllib.parse import quote from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, File, Form, Request from fastapi.responses import StreamingResponse from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy import select, func +from sqlalchemy import select, func, update from mutagen.mp3 import MP3 from io import BytesIO from ..database import get_db from ..models.user import User from ..models.track import Track +from ..models.room import Room from ..schemas.track import TrackResponse, TrackWithUrl from ..services.auth import get_current_user from ..services.s3 import upload_file, delete_file, generate_presigned_url, can_upload_file, get_file_size, stream_file_chunks @@ -184,10 +185,17 @@ async def delete_track( if track.uploaded_by != current_user.id: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not track owner") + # Clear current_track_id from any rooms using this track + await db.execute( + update(Room) + .where(Room.current_track_id == track_id) + .values(current_track_id=None, is_playing=False) + ) + # Delete from S3 await delete_file(track.s3_key) - # Delete from DB + # Delete from DB (queue entries will be cascade deleted automatically) await db.delete(track) return {"status": "deleted"} diff --git a/backend/app/routers/websocket.py b/backend/app/routers/websocket.py index cc512db..e7e549f 100644 --- a/backend/app/routers/websocket.py +++ b/backend/app/routers/websocket.py @@ -63,6 +63,11 @@ async def room_websocket(websocket: WebSocket, room_id: UUID): await websocket.send_json({"type": "pong"}) continue + # Handle pong response (from server ping) + if message.get("type") == "pong": + manager.update_pong(room_id, user.id) + continue + async with async_session() as db: if message["type"] == "player_action": await handle_player_action(db, room_id, user, message) @@ -72,11 +77,50 @@ async def room_websocket(websocket: WebSocket, room_id: UUID): await handle_sync_request(db, room_id, websocket) except WebSocketDisconnect: - manager.disconnect(websocket, room_id, user.id) - await manager.broadcast_to_room( - room_id, - {"type": "user_left", "user_id": str(user.id)}, + await handle_user_disconnect(websocket, room_id, user.id) + + +async def handle_user_disconnect(websocket: WebSocket, room_id: UUID, user_id: UUID): + """Обработка отключения пользователя от комнаты""" + manager.disconnect(websocket, room_id, user_id) + + # Удаляем пользователя из participants в БД + async with async_session() as db: + await db.execute( + select(RoomParticipant) + .where(RoomParticipant.room_id == room_id) + .where(RoomParticipant.user_id == user_id) ) + await db.execute( + RoomParticipant.__table__.delete().where( + RoomParticipant.room_id == room_id, + RoomParticipant.user_id == user_id + ) + ) + await db.commit() + + # Проверяем, остались ли участники в комнате + room_user_count = manager.get_room_user_count(room_id) + + # Если комната пустая - ставим трек на паузу + if room_user_count == 0: + result = await db.execute(select(Room).where(Room.id == room_id)) + room = result.scalar_one_or_none() + if room and room.is_playing: + # Сохраняем текущую позицию + if room.playback_started_at: + elapsed = (datetime.utcnow() - room.playback_started_at).total_seconds() * 1000 + room.playback_position = int((room.playback_position or 0) + elapsed) + + room.is_playing = False + room.playback_started_at = None + await db.commit() + + # Уведомляем остальных участников + await manager.broadcast_to_room( + room_id, + {"type": "user_left", "user_id": str(user_id)}, + ) async def handle_player_action(db: AsyncSession, room_id: UUID, user: User, message: dict): diff --git a/backend/app/services/ping_task.py b/backend/app/services/ping_task.py new file mode 100644 index 0000000..743600c --- /dev/null +++ b/backend/app/services/ping_task.py @@ -0,0 +1,88 @@ +import asyncio +import time +from uuid import UUID +from datetime import datetime +from sqlalchemy import select, delete +from app.database import async_session +from app.models.room import Room, RoomParticipant +from app.services.sync import manager + + +async def ping_users_task(): + """Фоновая задача для периодического пинга пользователей и отключения неактивных""" + while True: + try: + await asyncio.sleep(manager.ping_interval) + + current_time = time.time() + disconnected_users = [] + + # Проходим по всем активным соединениям + for room_id, connections in list(manager.get_all_connections().items()): + for websocket, user_id in list(connections): + # Отправляем ping + try: + await websocket.send_json({"type": "ping"}) + except Exception: + # Если не смогли отправить - помечаем на отключение + disconnected_users.append((websocket, room_id, user_id)) + continue + + # Проверяем время последнего pong + last_pong_time = manager.last_pong.get((room_id, user_id)) + if last_pong_time is None or (current_time - last_pong_time) > manager.ping_timeout: + # Пользователь не отвечает слишком долго + disconnected_users.append((websocket, room_id, user_id)) + + # Отключаем неактивных пользователей + for websocket, room_id, user_id in disconnected_users: + await disconnect_inactive_user(websocket, room_id, user_id) + + except Exception as e: + print(f"Error in ping_users_task: {e}") + await asyncio.sleep(5) # Подождём немного при ошибке + + +async def disconnect_inactive_user(websocket, room_id: UUID, user_id: UUID): + """Отключить неактивного пользователя""" + try: + # Закрываем WebSocket соединение + await websocket.close(code=1000, reason="Ping timeout") + except Exception: + pass + + # Удаляем из менеджера соединений + manager.disconnect(websocket, room_id, user_id) + + # Удаляем из БД + async with async_session() as db: + await db.execute( + delete(RoomParticipant).where( + RoomParticipant.room_id == room_id, + RoomParticipant.user_id == user_id + ) + ) + await db.commit() + + # Проверяем, остались ли участники + room_user_count = manager.get_room_user_count(room_id) + + # Если комната пустая - ставим трек на паузу + if room_user_count == 0: + result = await db.execute(select(Room).where(Room.id == room_id)) + room = result.scalar_one_or_none() + if room and room.is_playing: + # Сохраняем текущую позицию + if room.playback_started_at: + elapsed = (datetime.utcnow() - room.playback_started_at).total_seconds() * 1000 + room.playback_position = int((room.playback_position or 0) + elapsed) + + room.is_playing = False + room.playback_started_at = None + await db.commit() + + # Уведомляем остальных участников + await manager.broadcast_to_room( + room_id, + {"type": "user_left", "user_id": str(user_id), "reason": "timeout"}, + ) diff --git a/backend/app/services/sync.py b/backend/app/services/sync.py index 7e6d20a..e9e44a7 100644 --- a/backend/app/services/sync.py +++ b/backend/app/services/sync.py @@ -2,24 +2,39 @@ from typing import Dict, Set from fastapi import WebSocket from uuid import UUID import json +import time +import asyncio class ConnectionManager: def __init__(self): # room_id -> set of (websocket, user_id) self.active_connections: Dict[UUID, Set[tuple[WebSocket, UUID]]] = {} + # (room_id, user_id) -> last pong timestamp + self.last_pong: Dict[tuple[UUID, UUID], float] = {} + # Ping interval and timeout in seconds + self.ping_interval = 30 # Отправлять ping каждые 30 секунд + self.ping_timeout = 60 # Отключать если нет ответа 60 секунд async def connect(self, websocket: WebSocket, room_id: UUID, user_id: UUID): await websocket.accept() if room_id not in self.active_connections: self.active_connections[room_id] = set() self.active_connections[room_id].add((websocket, user_id)) + # Инициализируем время последнего pong + self.last_pong[(room_id, user_id)] = time.time() def disconnect(self, websocket: WebSocket, room_id: UUID, user_id: UUID): if room_id in self.active_connections: self.active_connections[room_id].discard((websocket, user_id)) if not self.active_connections[room_id]: del self.active_connections[room_id] + # Удаляем запись о последнем pong + self.last_pong.pop((room_id, user_id), None) + + def update_pong(self, room_id: UUID, user_id: UUID): + """Обновить время последнего pong от пользователя""" + self.last_pong[(room_id, user_id)] = time.time() async def broadcast_to_room(self, room_id: UUID, message: dict, exclude_user: UUID = None): if room_id not in self.active_connections: @@ -39,10 +54,35 @@ class ConnectionManager: for conn in disconnected: self.active_connections[room_id].discard(conn) + async def send_to_user(self, room_id: UUID, user_id: UUID, message: dict): + """Отправить сообщение конкретному пользователю в комнате""" + if room_id not in self.active_connections: + return False + + message_json = json.dumps(message, default=str) + for websocket, uid in self.active_connections[room_id]: + if uid == user_id: + try: + await websocket.send_text(message_json) + return True + except Exception: + return False + return False + def get_room_user_count(self, room_id: UUID) -> int: if room_id not in self.active_connections: return 0 return len(self.active_connections[room_id]) + def get_room_users(self, room_id: UUID) -> Set[UUID]: + """Получить список user_id в комнате""" + if room_id not in self.active_connections: + return set() + return {user_id for _, user_id in self.active_connections[room_id]} + + def get_all_connections(self) -> Dict[UUID, Set[tuple[WebSocket, UUID]]]: + """Получить все активные соединения (для фоновой задачи пинга)""" + return self.active_connections + manager = ConnectionManager() diff --git a/frontend/src/components/room/RoomCard.vue b/frontend/src/components/room/RoomCard.vue index 2146e54..4da81e5 100644 --- a/frontend/src/components/room/RoomCard.vue +++ b/frontend/src/components/room/RoomCard.vue @@ -38,6 +38,15 @@ + @@ -51,12 +60,23 @@ diff --git a/frontend/src/stores/activeRoom.js b/frontend/src/stores/activeRoom.js index f30dc3c..798c9e5 100644 --- a/frontend/src/stores/activeRoom.js +++ b/frontend/src/stores/activeRoom.js @@ -181,6 +181,17 @@ export const useActiveRoomStore = defineStore('activeRoom', () => { function handleMessage(msg) { switch (msg.type) { + case 'ping': + // Отвечаем на серверный ping + send({ type: 'pong' }) + break + case 'pong': + // Ответ на наш клиентский ping (игнорируем) + break + case 'room_deleted': + // Комната удалена владельцем + handleRoomDeleted(msg.message) + break case 'player_state': case 'sync_state': syncToState(msg) @@ -209,6 +220,16 @@ export const useActiveRoomStore = defineStore('activeRoom', () => { } } + function handleRoomDeleted(message) { + // Отключаемся от комнаты + disconnect() + + // Перенаправляем на главную страницу + if (typeof window !== 'undefined') { + window.location.href = '/' + } + } + function syncToState(state) { if (!audio) { initAudio() diff --git a/frontend/src/views/HomeView.vue b/frontend/src/views/HomeView.vue index cdf1081..c8902c7 100644 --- a/frontend/src/views/HomeView.vue +++ b/frontend/src/views/HomeView.vue @@ -43,6 +43,7 @@ :key="room.id" :room="room" @click="goToRoom(room.id)" + @delete="confirmDeleteRoom(room)" /> @@ -85,6 +86,38 @@ + + + + + + mdi-alert + Удалить комнату? + + + +

Вы уверены, что хотите удалить комнату "{{ roomToDelete?.name }}"?

+

+ Это действие нельзя отменить. Все участники будут отключены, а очередь и сообщения будут удалены. +

+ + + + + + Отмена + + + Удалить + + + + @@ -103,6 +136,9 @@ const loading = ref(true) const showCreateModal = ref(false) const newRoomName = ref('') const creating = ref(false) +const showDeleteDialog = ref(false) +const roomToDelete = ref(null) +const deleting = ref(false) onMounted(async () => { await roomStore.fetchRooms() @@ -128,6 +164,27 @@ function goToRoom(roomId) { router.push('/login') } } + +function confirmDeleteRoom(room) { + roomToDelete.value = room + showDeleteDialog.value = true +} + +async function deleteRoom() { + if (!roomToDelete.value) return + + try { + deleting.value = true + await roomStore.deleteRoom(roomToDelete.value.id) + showDeleteDialog.value = false + roomToDelete.value = null + } catch (e) { + console.error('Failed to delete room:', e) + alert('Ошибка при удалении комнаты') + } finally { + deleting.value = false + } +}