import asyncio import random from datetime import datetime, timedelta, timezone from jam_session.models import QueueState, Track ADJECTIVES = [ "Amazing", "Bold", "Cosmic", "Dapper", "Electric", "Funky", "Golden", "Happy", "Icy", "Jazzy", "Kind", "Loud", "Mellow", "Neon", "Orange", "Purple", "Quiet", "Rusty", "Sunny", "Turbo", ] NOUNS = [ "Badger", "Cobra", "Dolphin", "Eagle", "Falcon", "Grizzly", "Hawk", "Iguana", "Jaguar", "Koala", "Lion", "Moose", "Narwhal", "Otter", "Panda", "Quail", "Raven", "Shark", "Tiger", "Wolf", ] def generate_nickname() -> str: adj = random.choice(ADJECTIVES) noun = random.choice(NOUNS) num = random.randint(1, 99) return f"{adj}-{noun}-{num}" class NicknameTakenError(Exception): pass class SessionManager: def __init__(self): self._queue: list[Track] = [] self._current_track: Track | None = None self.paused: bool = False self.playback_time: float = 0.0 self._lock: asyncio.Lock = asyncio.Lock() self._nicknames: dict[str, str] = {} self._nickname_index: dict[str, str] = {} self._held_nicknames: dict[str, datetime] = {} self._votes: dict[str, set[str]] = {} self._manual_order: bool = False self._skip_votes: set[str] = set() self._skip_triggered: bool = False self._skip_cooldown_until: float = 0.0 self._skip_denominator: int = 0 self._cooldown_handle: asyncio.TimerHandle | None = None self._host_token: str | None = None self._host_conn_id: str | None = None @staticmethod def _compute_skip_threshold(nickname_count: int, threshold_pct: int) -> int: if nickname_count == 1: return 1 if nickname_count == 2: return 2 if nickname_count == 3: return 2 return int(nickname_count * threshold_pct / 100) + 1 async def add_track(self, track: Track) -> Track: async with self._lock: self._queue.append(track) return track async def remove_track(self, track_id: str) -> Track | None: async with self._lock: for votes in self._votes.values(): votes.discard(track_id) for i, t in enumerate(self._queue): if t.id == track_id: return self._queue.pop(i) return None async def reorder(self, new_order: list[str]) -> list[Track]: async with self._lock: ordered = [t for tid in new_order for t in self._queue if t.id == tid] if len(ordered) != len(self._queue): raise ValueError("new_order does not match queue content") self._queue = ordered return self._queue def get_state(self) -> QueueState: return QueueState( queue=list(self._queue), current_track=self._current_track, paused=self.paused, playback_time=self.playback_time, nicknames=dict(self._nicknames), held_nicknames={ n: exp.isoformat() for n, exp in self._held_nicknames.items() }, ) async def set_nickname(self, conn_id: str, nickname: str) -> str: async with self._lock: lower = nickname.lower() if lower in self._nickname_index: existing_conn = self._nickname_index[lower] if existing_conn != conn_id: raise NicknameTakenError( f"Nickname '{nickname}' is already taken" ) old_nickname = self._nicknames.get(conn_id) if old_nickname: old_lower = old_nickname.lower() if old_lower in self._nickname_index and self._nickname_index[old_lower] == conn_id: del self._nickname_index[old_lower] self._nicknames[conn_id] = nickname self._nickname_index[lower] = conn_id self._held_nicknames.pop(nickname, None) return nickname def get_nickname(self, conn_id: str) -> str | None: return self._nicknames.get(conn_id) def get_all_nicknames(self) -> dict[str, str]: return dict(self._nicknames) async def hold_nickname(self, conn_id: str): async with self._lock: nickname = self._nicknames.pop(conn_id, None) if nickname: self._held_nicknames[nickname] = datetime.now(timezone.utc) + timedelta(seconds=30) lower = nickname.lower() self._nickname_index.pop(lower, None) async def release_expired_holds(self): async with self._lock: now = datetime.now(timezone.utc) expired = [n for n, exp in self._held_nicknames.items() if now >= exp] for nickname in expired: del self._held_nicknames[nickname] async def toggle_vote(self, conn_id: str, track_id: str) -> tuple[int, bool]: async with self._lock: track_found = False for t in self._queue: if t.id == track_id: track_found = True break if not track_found: raise ValueError(f"Track {track_id} not found in queue") if conn_id not in self._votes: self._votes[conn_id] = set() was_upvoted = False if track_id in self._votes[conn_id]: self._votes[conn_id].discard(track_id) was_upvoted = False else: self._votes[conn_id].add(track_id) was_upvoted = True if not self._manual_order: self._sort_by_votes() count = sum(1 for votes in self._votes.values() if track_id in votes) return count, was_upvoted def _sort_by_votes(self): self._queue.sort(key=lambda t: ( -sum(1 for votes in self._votes.values() if t.id in votes), t.added_at )) def get_votes_for(self, conn_id: str) -> list[str]: return list(self._votes.get(conn_id, set())) def get_vote_counts(self) -> dict[str, int]: return { t.id: sum(1 for votes in self._votes.values() if t.id in votes) for t in self._queue } def set_manual_order(self, active: bool): self._manual_order = active def is_manual_order(self) -> bool: return self._manual_order def set_host_token(self, token: str): self._host_token = token def get_host_token(self) -> str | None: return self._host_token def validate_host_token(self, token: str) -> bool: return self._host_token is not None and token == self._host_token def set_host_conn_id(self, conn_id: str): self._host_conn_id = conn_id def get_host_conn_id(self) -> str | None: return self._host_conn_id def is_host(self, conn_id: str) -> bool: return self._host_conn_id is not None and self._host_conn_id == conn_id async def advance(self) -> Track | None: async with self._lock: if self._queue: self._current_track = self._queue.pop(0) else: self._current_track = None self.playback_time = 0.0 return self._current_track async def toggle_skip_vote(self, conn_id: str, threshold_pct: int) -> dict: async with self._lock: now = asyncio.get_running_loop().time() if now < self._skip_cooldown_until: return { "vote_count": len(self._skip_votes), "threshold": self._compute_skip_threshold(self._skip_denominator, threshold_pct), "triggered": False, "cooldown_remaining": int(self._skip_cooldown_until - now), } if conn_id in self._skip_votes: self._skip_votes.discard(conn_id) else: self._skip_votes.add(conn_id) threshold = self._compute_skip_threshold(self._skip_denominator, threshold_pct) triggered = len(self._skip_votes) >= threshold and not self._skip_triggered if triggered: self._skip_triggered = True return { "vote_count": len(self._skip_votes), "threshold": threshold, "triggered": triggered, "cooldown_remaining": 0, } def get_skip_state(self, conn_id: str, threshold_pct: int) -> dict: now = asyncio.get_running_loop().time() cooldown_remaining = 0 if now < self._skip_cooldown_until: cooldown_remaining = int(self._skip_cooldown_until - now) return { "my_skip_vote": conn_id in self._skip_votes, "skip_vote_count": len(self._skip_votes), "skip_threshold": self._compute_skip_threshold(self._skip_denominator, threshold_pct), "skip_cooldown_remaining": cooldown_remaining, } def reset_skip_votes(self): self._skip_votes.clear() self._skip_triggered = False self._skip_denominator = len(self._nicknames) def start_skip_cooldown(self, seconds: int): if self._cooldown_handle is not None: self._cooldown_handle.cancel() self._cooldown_handle = None loop = asyncio.get_running_loop() self._skip_cooldown_until = loop.time() + seconds def _on_cooldown_end(): self._cooldown_handle = None self._cooldown_handle = loop.call_later(seconds, _on_cooldown_end)