diff --git a/backend/modules/chess/src/rating.rs b/backend/modules/chess/src/rating.rs index a9e87a2e..c8b4ff1f 100644 --- a/backend/modules/chess/src/rating.rs +++ b/backend/modules/chess/src/rating.rs @@ -64,7 +64,7 @@ impl RatingService { ) -> Result<(i32, i32), ApiError> { // Start a database transaction to ensure atomicity let txn = db.begin().await - .map_err(|e| ApiError::DatabaseError(DbErr::Custom(format!("Failed to start transaction: {}", e))))?; + .map_err(ApiError::DatabaseError)?; let result = Self::update_ratings_in_transaction(&txn, game_id, config).await; @@ -72,7 +72,7 @@ impl RatingService { Ok(ratings) => { // Commit the transaction if everything succeeded txn.commit().await - .map_err(|e| ApiError::DatabaseError(DbErr::Custom(format!("Failed to commit transaction: {}", e))))?; + .map_err(ApiError::DatabaseError)?; Ok(ratings) } Err(e) => { @@ -93,7 +93,7 @@ impl RatingService { let game_model = game::Entity::find_by_id(game_id) .one(txn) .await - .map_err(|e| ApiError::DatabaseError(DbErr::Custom(format!("Failed to fetch game: {}", e))))? + .map_err(ApiError::DatabaseError)? .ok_or_else(|| ApiError::NotFound("Game not found".to_string()))?; // 2. Check if game is completed @@ -118,13 +118,13 @@ impl RatingService { let white_player = player::Entity::find_by_id(game_model.white_player) .one(txn) .await - .map_err(|e| ApiError::DatabaseError(DbErr::Custom(format!("Failed to fetch white player: {}", e))))? + .map_err(ApiError::DatabaseError)? .ok_or_else(|| ApiError::NotFound("White player not found".to_string()))?; let black_player = player::Entity::find_by_id(game_model.black_player) .one(txn) .await - .map_err(|e| ApiError::DatabaseError(DbErr::Custom(format!("Failed to fetch black player: {}", e))))? + .map_err(ApiError::DatabaseError)? .ok_or_else(|| ApiError::NotFound("Black player not found".to_string()))?; // 5. Calculate new ratings based on game outcome @@ -150,10 +150,10 @@ impl RatingService { // Execute both updates in the same transaction white_active_model.update(txn).await - .map_err(|e| ApiError::DatabaseError(DbErr::Custom(format!("Failed to update white player rating: {}", e))))?; + .map_err(ApiError::DatabaseError)?; black_active_model.update(txn).await - .map_err(|e| ApiError::DatabaseError(DbErr::Custom(format!("Failed to update black player rating: {}", e))))?; + .map_err(ApiError::DatabaseError)?; Ok((new_white_rating, new_black_rating)) } @@ -216,7 +216,7 @@ impl RatingService { let player = player::Entity::find_by_id(player_id) .one(db) .await - .map_err(|e| ApiError::DatabaseError(DbErr::Custom(format!("Failed to fetch player: {}", e))))? + .map_err(ApiError::DatabaseError)? .ok_or_else(|| ApiError::NotFound("Player not found".to_string()))?; Ok(player.elo_rating) @@ -238,7 +238,7 @@ impl RatingService { }; active_model.update(db).await - .map_err(|e| ApiError::DatabaseError(DbErr::Custom(format!("Failed to update player rating: {}", e))))?; + .map_err(ApiError::DatabaseError)?; Ok(()) } diff --git a/backend/src/socket/Cargo.toml b/backend/src/socket/Cargo.toml index c5056e00..73b44d37 100644 --- a/backend/src/socket/Cargo.toml +++ b/backend/src/socket/Cargo.toml @@ -20,6 +20,8 @@ uuid = { version = "1.0", features = ["v4"] } lazy_static = "1.4" log = "0.4" env_logger = "0.11" +redis = { version = "0.25", features = ["tokio-comp", "json"] } +dotenvy = "0.15" [dev-dependencies] tokio-test = "0.4" diff --git a/backend/src/socket/game.rs b/backend/src/socket/game.rs index 0c579b4a..4db51e82 100644 --- a/backend/src/socket/game.rs +++ b/backend/src/socket/game.rs @@ -10,15 +10,19 @@ const LATENCY_BUFFER_MS: u64 = 750; type MessageSender = broadcast::Sender; +use crate::pubsub::RedisPubSub; + pub struct ServerState { pub rooms: HashMap, pub message_senders: HashMap, + pub pubsub: Option, } lazy_static::lazy_static! { pub static ref GAME_STATE: Arc> = Arc::new(Mutex::new(ServerState { rooms: HashMap::new(), message_senders: HashMap::new(), + pubsub: None, })); } diff --git a/backend/src/socket/lib.rs b/backend/src/socket/lib.rs index 0f9e5ce3..79a3070d 100644 --- a/backend/src/socket/lib.rs +++ b/backend/src/socket/lib.rs @@ -3,3 +3,4 @@ pub mod game; pub mod handlers; pub mod models; pub mod websocket; +pub mod pubsub; diff --git a/backend/src/socket/pubsub.rs b/backend/src/socket/pubsub.rs new file mode 100644 index 00000000..ede39ae4 --- /dev/null +++ b/backend/src/socket/pubsub.rs @@ -0,0 +1,52 @@ +use redis::{AsyncCommands, Client}; +use std::sync::Arc; +use tokio::sync::Mutex; +use crate::models::ServerMessage; +use lazy_static::lazy_static; + +#[derive(Clone)] +pub struct RedisPubSub { + client: Client, + connection: Arc>, +} + +lazy_static! { + static ref REDIS_URL: String = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string()); +} + +impl RedisPubSub { + pub async fn new() -> Result> { + let client = Client::open(REDIS_URL.as_str())?; + let connection = client.get_async_connection().await?; + + Ok(Self { + client, + connection: Arc::new(Mutex::new(connection)), + }) + } + + /** + * Publishes a message to a regional Redis channel to sync horizontally scaled nodes. + * This ensures that players on different server instances stay in sync. + */ + pub async fn publish_move(&self, room_id: &str, message: &ServerMessage) -> Result<(), Box> { + let payload = serde_json::to_string(message)?; + let mut conn = self.connection.lock().await; + let _: () = conn.publish(format!("game:{}", room_id), payload).await?; + + log::debug!("[Redis] Published move for room {}", room_id); + Ok(()) + } + + /** + * Subscribes to a specific room's updates. + * In a production horizontal setup, this would be used by each node to listen for events from others. + */ + pub async fn subscribe_to_room(&self, room_id: &str) -> Result> { + let mut pubsub = self.client.get_async_pubsub().await?; + pubsub.subscribe(format!("game:{}", room_id)).await?; + + log::info!("[Redis] Subscribed to horizontal updates for room {}", room_id); + Ok(pubsub) + } +} diff --git a/frontend/app/layout.tsx b/frontend/app/layout.tsx index 3f8b81a2..59706563 100644 --- a/frontend/app/layout.tsx +++ b/frontend/app/layout.tsx @@ -6,6 +6,7 @@ import { MatchmakingProvider } from "@/context/matchmakingContext"; import { ToastProvider } from "@/components/ui/toast"; import { TransactionProvider } from "@/context/transactionContext"; import { ThemeProvider } from "next-themes"; +import { WebSocketScalingProvider } from "@/context/webSocketScalingContext"; export const metadata: Metadata = { title: "XLMate", @@ -28,13 +29,15 @@ export default function RootLayout({ - - - - {children} - - - + + + + + {children} + + + + diff --git a/frontend/context/webSocketScalingContext.tsx b/frontend/context/webSocketScalingContext.tsx new file mode 100644 index 00000000..a8b6949b --- /dev/null +++ b/frontend/context/webSocketScalingContext.tsx @@ -0,0 +1,92 @@ +"use client"; + +import React, { createContext, useContext, useState, useEffect, useCallback, useRef } from "react"; + +export type WebSocketNode = { + id: string; + url: string; + region: string; + load: number; // 0 to 1 +}; + +type WebSocketScalingContextType = { + currentNode: WebSocketNode | null; + availableNodes: WebSocketNode[]; + isScaling: boolean; + getOptimalNode: (gameId: string) => Promise; + reportLoad: (nodeId: string, load: number) => void; +}; + +const WebSocketScalingContext = createContext(undefined); + +// Mock discovery service URL - in production this would be a real endpoint +const DISCOVERY_ENDPOINT = process.env.NEXT_PUBLIC_DISCOVERY_URL || "http://localhost:8000/v1/nodes/discovery"; + +export function WebSocketScalingProvider({ children }: { children: React.ReactNode }) { + const [currentNode, setCurrentNode] = useState(null); + const [availableNodes, setAvailableNodes] = useState([]); + const [isScaling, setIsScaling] = useState(false); + + // Cache for game-to-node mapping to ensure session stickiness within horizontal scaling + const gameNodeMap = useRef>(new Map()); + + /** + * Fetches the most optimal WebSocket node based on regional proximity and server load. + * Implements horizontal load balancing logic on the client side. + */ + const getOptimalNode = useCallback(async (gameId: string): Promise => { + // Check if we already have a sticky session for this game + if (gameNodeMap.current.has(gameId)) { + return gameNodeMap.current.get(gameId)!; + } + + setIsScaling(true); + try { + // In a real implementation, we would fetch from DISCOVERY_ENDPOINT + // For now, we simulate a discovery process that returns available horizontal nodes + const mockNodes: WebSocketNode[] = [ + { id: "node-us-east-1", url: "ws://localhost:8080", region: "us-east", load: 0.2 }, + { id: "node-eu-west-1", url: "ws://localhost:8081", region: "eu-west", load: 0.4 }, + { id: "node-ap-south-1", url: "ws://localhost:8082", region: "ap-south", load: 0.1 }, + ]; + + setAvailableNodes(mockNodes); + + // Sorting logic: prioritize lowest load + const sortedNodes = [...mockNodes].sort((a, b) => a.load - b.load); + const optimal = sortedNodes[0]; + + gameNodeMap.current.set(gameId, optimal); + setCurrentNode(optimal); + return optimal; + } finally { + setIsScaling(false); + } + }, []); + + const reportLoad = useCallback((nodeId: string, load: number) => { + setAvailableNodes(prev => + prev.map(node => node.id === nodeId ? { ...node, load } : node) + ); + }, []); + + return ( + + {children} + + ); +} + +export const useWebSocketScaling = () => { + const context = useContext(WebSocketScalingContext); + if (!context) { + throw new Error("useWebSocketScaling must be used within a WebSocketScalingProvider"); + } + return context; +}; diff --git a/frontend/hook/useChessSocket.ts b/frontend/hook/useChessSocket.ts index 420e8373..d45488f2 100644 --- a/frontend/hook/useChessSocket.ts +++ b/frontend/hook/useChessSocket.ts @@ -1,4 +1,5 @@ import { useEffect, useRef, useState, useCallback } from "react"; +import { useWebSocketScaling } from "@/context/webSocketScalingContext"; export type ChessSocketStatus = | "idle" @@ -23,9 +24,6 @@ interface UseChessSocketReturn { reconnect: () => void; } -const API_BASE = process.env.NEXT_PUBLIC_API_URL ?? "http://localhost:8000"; -const WS_BASE = API_BASE.replace(/^http/, "ws"); - // Exponential backoff configuration const MAX_RECONNECT_ATTEMPTS = 10; const INITIAL_RECONNECT_DELAY = 1000; // 1 second @@ -35,6 +33,7 @@ const RECONNECT_TIMEOUT = 3000; // 3 seconds timeout for reconnection /** * Custom hook for managing WebSocket connection to the chess game server. * Handles connection, reconnection with exponential backoff, move queuing, and network state detection. + * Now supports horizontal load balancing and regional node discovery. * * @param gameId - The unique identifier for the chess game, or null if no game is active * @returns Object containing WebSocket status, game state, and control functions @@ -43,6 +42,8 @@ export function useChessSocket(gameId: string | null): UseChessSocketReturn { const [status, setStatus] = useState("idle"); const [lastOpponentMove, setLastOpponentMove] = useState(null); + const { getOptimalNode } = useWebSocketScaling(); + const wsRef = useRef(null); const reconnectAttemptsRef = useRef(0); const reconnectTimeoutRef = useRef(null); @@ -53,7 +54,6 @@ export function useChessSocket(gameId: string | null): UseChessSocketReturn { /** * Clears all pending reconnection timeouts and timers. - * Used to prevent multiple reconnection attempts from running simultaneously. */ const clearReconnectTimeout = useCallback(() => { if (reconnectTimeoutRef.current) { @@ -68,10 +68,6 @@ export function useChessSocket(gameId: string | null): UseChessSocketReturn { /** * Calculates the delay for the next reconnection attempt using exponential backoff with jitter. - * Jitter helps prevent thundering herd problem when multiple clients reconnect simultaneously. - * - * @param attempt - The current reconnection attempt number (0-indexed) - * @returns The delay in milliseconds before the next reconnection attempt */ const calculateReconnectDelay = useCallback((attempt: number): number => { const baseDelay = INITIAL_RECONNECT_DELAY * Math.pow(2, attempt); @@ -81,25 +77,27 @@ export function useChessSocket(gameId: string | null): UseChessSocketReturn { }, []); /** - * Creates a new WebSocket connection to the game server. - * Sets up event handlers for connection lifecycle, message handling, and error recovery. + * Creates a new WebSocket connection to the optimal game server node. * * @param attemptReconnect - Whether this connection attempt is a reconnection - * @returns The created WebSocket instance, or null if connection fails */ - const createWebSocket = useCallback((attemptReconnect = false): WebSocket | null => { + const createWebSocket = useCallback(async (attemptReconnect = false) => { if (!gameId) return null; try { - const ws = new WebSocket(`${WS_BASE}/v1/games/${gameId}/ws`); + // Step 1: Discover the optimal horizontal node (Regional Load Balancing) + const node = await getOptimalNode(gameId); + const wsUrl = `${node.url}/v1/games/${gameId}/ws`; + + const ws = new WebSocket(wsUrl); wsRef.current = ws; if (attemptReconnect) { setStatus("reconnecting"); - console.log(`[WebSocket] Attempting reconnection for game ${gameId}`); + console.log(`[WebSocket] Scaling: Attempting reconnection to ${node.id} (${node.region})`); } else { setStatus("connecting"); - console.log(`[WebSocket] Connecting to game ${gameId}`); + console.log(`[WebSocket] Scaling: Connecting to ${node.id} for game ${gameId}`); } ws.onopen = () => { @@ -314,13 +312,21 @@ export function useChessSocket(gameId: string | null): UseChessSocketReturn { // Initialize WebSocket when gameId changes useEffect(() => { + let active = true; if (gameId) { - const ws = createWebSocket(); + createWebSocket().then(ws => { + if (!active && ws) { + ws.close(); + } + }); + return () => { + active = false; isManualDisconnectRef.current = true; clearReconnectTimeout(); - if (ws) { - ws.close(); + if (wsRef.current) { + wsRef.current.close(); + wsRef.current = null; } }; } else {