# -*- coding: utf-8 -*- """Live Room: private audio broadcast fan-out (relay side). The HOST's PC encodes its playback with the bundled mpv.exe (24 kbps MP3) and pushes chunks here; listeners receive them as a live HTTP stream in their browser. The relay is a BYTE PIPE only: * no subprocesses, no encoding, no audio processing; * a rolling in-memory buffer (~512 KB) per room - nothing is stored, a listener who joins mid-stream starts at the live edge; * rooms exist only while the host pushes (LIVE_ROOM_TTL after the last push, then gone); the room token is a random 32-hex string unrelated to any control-session code. Resource caps (the whole safety model): MAX_ROOMS rooms, MAX_LISTENERS fan-out per room, buffer bytes capped per room. Worst case for the VPS is MAX_ROOMS * MAX_LISTENERS * 12 KB/s of trivial socket writes. """ import threading import time from collections import deque from typing import Dict, Optional LIVE_ROOM_TTL = 5 * 60 # room dies 5 min after the last push MAX_ROOMS = 8 MAX_LISTENERS = 20 # concurrent listeners per room BUFFER_CHUNKS = 32 # rolling ring: ~1-2 min of 24 kbps audio CHUNK_SOFT_CAP = 131072 # one push bigger than this is refused # ~96 kbps MP3 ~= 12 KB/s; BUFFER_CHUNKS keeps ~40 s of back-buffer. class LiveRoom: def __init__(self, token: str, title: str = "", by: str = "", owner: str = ""): self.token = token self.title = title # "now playing" text for the room self.by = by self.owner = owner # identity of the host ("admin" or "lr:") self.created = time.time() self.last_push = time.time() self.listeners = 0 self.bytes_total = 0 self.chunks: deque = deque(maxlen=BUFFER_CHUNKS) self.chunk_seq = 0 # monotonically increasing chunk id self.lock = threading.Lock() def push(self, data: bytes) -> bool: if len(data) > CHUNK_SOFT_CAP or not data: return False with self.lock: self.chunks.append(data) self.chunk_seq += 1 self.bytes_total += len(data) self.last_push = time.time() return True def snapshot_from(self, min_chunks: int): """(chunks, start_seq) for a listener: everything buffered, or wait until *min_chunks* exist (a fresh room before the first bytes). Cheap: a list slice of the deque.""" with self.lock: if len(self.chunks) < min_chunks: return None return (list(self.chunks), self.chunk_seq - len(self.chunks)) def age(self) -> float: return time.time() - self.last_push class LiveRooms: def __init__(self): self._lock = threading.Lock() self._rooms: Dict[str, LiveRoom] = {} self._last_sweep = time.time() # -- host --------------------------------------------------------------- def create(self, token: str, title: str = "", by: str = "", owner: str = "") -> str: """Returns 'ok', 'exists' or 'full'. The owner identity doubles as the per-user cap: a stable identity (a paired session's token hash, or the operator key) always maps to the SAME room token, so re-hosting refreshes the existing room instead of allowing a second one - one room per host, free.""" with self._lock: self._sweep_locked() if token in self._rooms: # Host re-announcing: refresh metadata, keep buffer. self._rooms[token].last_push = time.time() self._rooms[token].owner = owner or self._rooms[token].owner return "exists" if len(self._rooms) >= MAX_ROOMS: return "full" self._rooms[token] = LiveRoom(token, title, by, owner) return "ok" def close(self, token: str) -> bool: with self._lock: return self._rooms.pop(token, None) is not None def close_by_owner(self, owner: str) -> int: """Kill every room hosted by *owner* (admin revocation path). Returns the number of rooms closed.""" with self._lock: dead = [t for t, room in self._rooms.items() if room.owner == owner] for t in dead: del self._rooms[t] return len(dead) def push(self, token: str, data: bytes) -> str: """Returns 'ok', 'missing' or 'full' (room expired).""" with self._lock: room = self._rooms.get(token) if room is None: return "missing" if room.age() > LIVE_ROOM_TTL: self.close(token) return "full" return "ok" if room.push(data) else "bad" def set_title(self, token: str, title: str, by: str = "") -> None: with self._lock: room = self._rooms.get(token) if room: room.title = str(title or "")[:200] room.by = str(by or "")[:24] room.last_push = time.time() def meta(self, token: str) -> Optional[dict]: """Public, read-only info for the listener page.""" with self._lock: room = self._rooms.get(token) if not room or room.age() > LIVE_ROOM_TTL: return None return {"title": room.title, "listeners": room.listeners} def stats(self, token: str) -> Optional[dict]: with self._lock: room = self._rooms.get(token) if not room: return None return {"listeners": room.listeners, "bytes": room.bytes_total, "title": room.title, "age": round(room.age(), 1)} # -- listener ----------------------------------------------------------- def open_stream(self, token: str): """Returns the room while registering a listener slot, or None when the room is gone or full. Caller MUST release_stream().""" with self._lock: room = self._rooms.get(token) if room is None or room.age() > LIVE_ROOM_TTL: return None if room.listeners >= MAX_LISTENERS: return None room.listeners += 1 return room def release_stream(self, room: LiveRoom) -> None: with room.lock: room.listeners = max(0, room.listeners - 1) # -- housekeeping --------------------------------------------------------- def sweep(self) -> None: with self._lock: self._sweep_locked() def _sweep_locked(self) -> None: now = time.time() if now - self._last_sweep < 60: return self._last_sweep = now dead = [t for t, room in self._rooms.items() if room.age() > LIVE_ROOM_TTL] for t in dead: del self._rooms[t]