"""Double Elimination Tournament Engine for ServeSync""" from sqlalchemy.orm import Session from app.models.tournament import Tournament, TournamentEntry, TournamentMatch, TournamentStatus, BracketType, TournamentMatchStatus from app.models.player import Player from app.models.court import Court from app.models.match import Match, MatchPlayer, MatchStage, MatchStatus, MatchType from app.services.matchmaking import find_available_court from datetime import datetime from typing import Optional, List import math def create_tournament(db: Session, name: str, max_participants: int = 8) -> Tournament: """Create a new tournament""" t = Tournament(name=name, max_participants=max_participants) db.add(t) db.commit() db.refresh(t) return t def register_player(db: Session, tournament_id: int, player_id: int) -> Optional[TournamentEntry]: """Register a player for the tournament""" t = db.query(Tournament).filter(Tournament.id == tournament_id).first() if not t or t.status != TournamentStatus.REGISTRATION: return None current_count = db.query(TournamentEntry).filter(TournamentEntry.tournament_id == tournament_id).count() if current_count >= t.max_participants: return None existing = db.query(TournamentEntry).filter( TournamentEntry.tournament_id == tournament_id, TournamentEntry.player_id == player_id ).first() if existing: return existing entry = TournamentEntry(tournament_id=tournament_id, player_id=player_id) db.add(entry) db.commit() db.refresh(entry) return entry def start_tournament(db: Session, tournament_id: int) -> Optional[Tournament]: """Start tournament and generate initial bracket""" t = db.query(Tournament).filter(Tournament.id == tournament_id).first() if not t or t.status != TournamentStatus.REGISTRATION: return None entries = db.query(TournamentEntry).filter( TournamentEntry.tournament_id == tournament_id ).join(Player).order_by(Player.elo_rating.desc()).all() if len(entries) < 4: return None # Seed players by ELO for i, entry in enumerate(entries): entry.seed = i + 1 # Generate bracket matches _generate_winners_bracket(db, t, entries) t.status = TournamentStatus.IN_PROGRESS t.started_at = datetime.utcnow() db.commit() db.refresh(t) return t def _generate_winners_bracket(db: Session, tournament: Tournament, entries: List[TournamentEntry]): """Generate the initial winners bracket matches""" n = len(entries) # Pair up by seeding (1 vs n, 2 vs n-1, etc.) match_num = 1 pairs = [] for i in range(n // 2): p1 = entries[i] p2 = entries[n - 1 - i] pairs.append((p1, p2)) for p1, p2 in pairs: tm = TournamentMatch( tournament_id=tournament.id, round_number=1, match_number=match_num, bracket_type=BracketType.WINNERS, status=TournamentMatchStatus.PENDING, player1_id=p1.player_id, player2_id=p2.player_id, ) db.add(tm) match_num += 1 db.flush() def complete_tournament_match( db: Session, tournament_match_id: int, team1_score: int, team2_score: int ) -> Optional[TournamentMatch]: """Complete a tournament match and advance bracket""" tm = db.query(TournamentMatch).filter(TournamentMatch.id == tournament_match_id).first() if not tm: return None tm.team1_score = team1_score tm.team2_score = team2_score tm.winner_team = 1 if team1_score > team2_score else 2 tm.status = TournamentMatchStatus.COMPLETED tm.completed_at = datetime.utcnow() winner_player_id = tm.player1_id if tm.winner_team == 1 else tm.player2_id loser_player_id = tm.player2_id if tm.winner_team == 1 else tm.player1_id # Update entry stats winner_entry = db.query(TournamentEntry).filter( TournamentEntry.tournament_id == tm.tournament_id, TournamentEntry.player_id == winner_player_id ).first() loser_entry = db.query(TournamentEntry).filter( TournamentEntry.tournament_id == tm.tournament_id, TournamentEntry.player_id == loser_player_id ).first() if winner_entry: winner_entry.wins += 1 if loser_entry: loser_entry.losses += 1 # Double elimination logic if loser_entry and loser_entry.losses < 2: # Move to losers bracket loser_entry.is_in_losers = True _create_losers_match(db, tm, loser_player_id) elif loser_entry: # Eliminated loser_entry.is_eliminated = True _assign_final_ranks(db, tm.tournament_id) db.commit() db.refresh(tm) return tm def _create_losers_match(db: Session, completed_match: TournamentMatch, loser_id: int): """Route loser to losers bracket""" # Find existing pending losers match in next round next_round = completed_match.round_number + 1 existing = db.query(TournamentMatch).filter( TournamentMatch.tournament_id == completed_match.tournament_id, TournamentMatch.bracket_type == BracketType.LOSERS, TournamentMatch.round_number == next_round, TournamentMatch.player2_id.is_(None), ).first() if existing: existing.player2_id = loser_id else: # Create a new losers match match_count = db.query(TournamentMatch).filter( TournamentMatch.tournament_id == completed_match.tournament_id, TournamentMatch.bracket_type == BracketType.LOSERS, TournamentMatch.round_number == next_round, ).count() tm = TournamentMatch( tournament_id=completed_match.tournament_id, round_number=next_round, match_number=match_count + 1, bracket_type=BracketType.LOSERS, status=TournamentMatchStatus.PENDING, player1_id=loser_id, ) db.add(tm) def _assign_final_ranks(db: Session, tournament_id: int): """Assign final ranks to eliminated players""" entries = db.query(TournamentEntry).filter( TournamentEntry.tournament_id == tournament_id, TournamentEntry.is_eliminated == True, TournamentEntry.final_rank.is_(None) ).order_by(TournamentEntry.losses.desc()).all() remaining = db.query(TournamentEntry).filter( TournamentEntry.tournament_id == tournament_id, TournamentEntry.is_eliminated == False ).count() base_rank = remaining + 1 for i, entry in enumerate(entries): entry.final_rank = base_rank + i def get_bracket_data(db: Session, tournament_id: int) -> dict: """Get structured bracket data for frontend visualization""" tournament = db.query(Tournament).filter(Tournament.id == tournament_id).first() if not tournament: return {} matches = db.query(TournamentMatch).filter( TournamentMatch.tournament_id == tournament_id ).order_by(TournamentMatch.round_number, TournamentMatch.match_number).all() entries = db.query(TournamentEntry).filter( TournamentEntry.tournament_id == tournament_id ).all() players_map = {} for entry in entries: player = db.query(Player).filter(Player.id == entry.player_id).first() if player: players_map[player.id] = { "id": player.id, "name": player.name, "elo": player.elo_rating, "seed": entry.seed, "wins": entry.wins, "losses": entry.losses, "is_eliminated": entry.is_eliminated, "is_in_losers": entry.is_in_losers, "final_rank": entry.final_rank, } winners_bracket = [] losers_bracket = [] grand_final = [] for m in matches: match_data = { "id": m.id, "round": m.round_number, "match_number": m.match_number, "status": m.status.value, "player1": players_map.get(m.player1_id), "player2": players_map.get(m.player2_id), "team1_score": m.team1_score, "team2_score": m.team2_score, "winner_team": m.winner_team, } if m.bracket_type == BracketType.WINNERS: winners_bracket.append(match_data) elif m.bracket_type == BracketType.LOSERS: losers_bracket.append(match_data) else: grand_final.append(match_data) # Build leaderboard leaderboard = sorted( [ { "player": players_map[e.player_id], "wins": e.wins, "losses": e.losses, "is_eliminated": e.is_eliminated, "is_in_losers": e.is_in_losers, "final_rank": e.final_rank, } for e in entries if e.player_id in players_map ], key=lambda x: (x["final_rank"] or 999, -x["wins"]), ) return { "tournament": { "id": tournament.id, "name": tournament.name, "status": tournament.status.value, "current_round": tournament.current_round, }, "winners_bracket": winners_bracket, "losers_bracket": losers_bracket, "grand_final": grand_final, "leaderboard": leaderboard, }