feat: jams — shared listening rooms (#949) · zzstoatzz.io/plyr.fm@90ce8b8

+15

.claude/commands/enable-flag.md

··· 1 + --- 2 + description: Enable a feature flag for a user 3 + argument-hint: <flag_name> for <handle> 4 + --- 5 + 6 + Enable feature flag: $ARGUMENTS 7 + 8 + Use the Neon MCP `run_sql` tool against the dev database (`muddy-flower-98795112`). Look up the DID from the `artists` table by handle, then insert into `feature_flags`: 9 + 10 + ```sql 11 + INSERT INTO feature_flags (user_did, flag, created_at) VALUES ('<did>', '<flag>', NOW()) 12 + ON CONFLICT (user_did, flag) DO NOTHING 13 + ``` 14 + 15 + Known flags: check `KNOWN_FLAGS` in `backend/src/backend/_internal/feature_flags.py`.

+16 -1

STATUS.md

··· 47 47 48 48 ### February 2026 49 49 50 + #### jams — shared listening rooms (Feb 19) 51 + 52 + real-time shared listening rooms. one user creates a jam, gets a shareable code (`plyr.fm/jam/a1b2c3d4`), and anyone with the link can join. all participants control playback — play, pause, seek, next, previous, add/remove tracks. no chat, no host-only lock. 53 + 54 + **backend**: `Jam` and `JamParticipant` models with partial indexes. `JamService` singleton manages lifecycle, WebSocket connections, and Redis Streams fan-out. playback state is server-authoritative — stored as JSONB with monotonic revision counter. commands mutate state, increment revision, commit, and publish to Redis stream. each backend instance runs `XREAD BLOCK` per active jam and fans out to connected WebSockets. 55 + 56 + **frontend**: `JamState` class (`lib/jam.svelte.ts`) with Svelte 5 runes. WebSocket lifecycle with exponential backoff reconnect (1s → 30s). Player.svelte has jam-aware effects — track sync from jam state instead of personal queue, drift correction (seek if >2s off from server timestamp). PlaybackControls routes commands through jam when active. jam page at `/jam/[code]` with track display, controls, participant avatars, share button. 57 + 58 + **sync**: server-timestamp + client interpolation (`progress_ms + (Date.now() - server_time_ms)`). reconnect replays missed events via `XRANGE` from last stream ID, falls back to full DB snapshot if trimmed. personal queue preserved and restored on leave. 59 + 60 + gated behind `jams` feature flag. 19 backend tests covering full lifecycle, all commands, revision monotonicity, flag gating, auto-leave. see `docs/architecture/jams.md`. 61 + 62 + --- 63 + 50 64 #### hidden tag filter autocomplete (PR #945, Feb 18) 51 65 52 66 the homepage hidden tag filter's "add tag" input now has autocomplete. typing a partial tag name fetches matching tags from `GET /tracks/tags?q=...` (same endpoint the portal tag editor uses) with a 200ms debounce. suggestions appear in a compact glass-effect dropdown showing tag name and track count. supports keyboard navigation (arrow keys to cycle, enter to select, escape to close) and mouse selection. tags already in the hidden list are filtered out of suggestions. frontend-only change. ··· 328 342 - ✅ media export with concurrent downloads 329 343 - ✅ supporter-gated content via atprotofans 330 344 - ✅ listen receipts (tracked share links with visitor/listener stats) 345 + - ✅ jams — shared listening rooms with real-time sync via Redis Streams + WebSocket (feature-flagged) 331 346 332 347 **albums** 333 348 - ✅ album CRUD with cover art ··· 438 453 439 454 --- 440 455 441 - this is a living document. last updated 2026-02-18 (hidden tag filter autocomplete). 456 + this is a living document. last updated 2026-02-19 (jams — shared listening rooms). 442 457

+2

backend/alembic/env.py

··· 9 9 from backend.models import ( # noqa: F401 10 10 Artist, 11 11 CopyrightScan, 12 + Jam, 13 + JamParticipant, 12 14 Track, 13 15 TrackComment, 14 16 TrackLike,

+92

backend/alembic/versions/2026_02_19_001258_f4ff6ce7d78b_add_jams_tables.py

··· 1 + """add jams tables 2 + 3 + Revision ID: f4ff6ce7d78b 4 + Revises: e88dbd481272 5 + Create Date: 2026-02-19 00:12:58.334693 6 + 7 + """ 8 + 9 + from collections.abc import Sequence 10 + 11 + import sqlalchemy as sa 12 + from sqlalchemy.dialects import postgresql 13 + 14 + from alembic import op 15 + 16 + # revision identifiers, used by Alembic. 17 + revision: str = "f4ff6ce7d78b" 18 + down_revision: str | Sequence[str] | None = "e88dbd481272" 19 + branch_labels: str | Sequence[str] | None = None 20 + depends_on: str | Sequence[str] | None = None 21 + 22 + 23 + def upgrade() -> None: 24 + """Upgrade schema.""" 25 + op.create_table( 26 + "jams", 27 + sa.Column("id", sa.String(), nullable=False), 28 + sa.Column("code", sa.String(length=12), nullable=False), 29 + sa.Column("host_did", sa.String(), nullable=False), 30 + sa.Column("name", sa.String(length=100), nullable=True), 31 + sa.Column("state", postgresql.JSON(astext_type=sa.Text()), nullable=False), 32 + sa.Column("revision", sa.BigInteger(), nullable=False), 33 + sa.Column("is_active", sa.Boolean(), nullable=False), 34 + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), 35 + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), 36 + sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True), 37 + sa.ForeignKeyConstraint(["host_did"], ["artists.did"]), 38 + sa.PrimaryKeyConstraint("id"), 39 + sa.UniqueConstraint("code"), 40 + ) 41 + op.create_index("ix_jams_code", "jams", ["code"], unique=True) 42 + op.create_index("ix_jams_host_did", "jams", ["host_did"], unique=False) 43 + op.create_index( 44 + "ix_jams_is_active", 45 + "jams", 46 + ["is_active"], 47 + unique=False, 48 + postgresql_where=sa.text("is_active IS true"), 49 + ) 50 + op.create_table( 51 + "jam_participants", 52 + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), 53 + sa.Column("jam_id", sa.String(), nullable=False), 54 + sa.Column("did", sa.String(), nullable=False), 55 + sa.Column("joined_at", sa.DateTime(timezone=True), nullable=False), 56 + sa.Column("left_at", sa.DateTime(timezone=True), nullable=True), 57 + sa.ForeignKeyConstraint(["did"], ["artists.did"]), 58 + sa.ForeignKeyConstraint(["jam_id"], ["jams.id"], ondelete="CASCADE"), 59 + sa.PrimaryKeyConstraint("id"), 60 + ) 61 + op.create_index( 62 + "ix_jam_participants_did_active", 63 + "jam_participants", 64 + ["did"], 65 + unique=False, 66 + postgresql_where=sa.text("left_at IS NULL"), 67 + ) 68 + op.create_index( 69 + "ix_jam_participants_jam_id", 70 + "jam_participants", 71 + ["jam_id"], 72 + unique=False, 73 + ) 74 + 75 + 76 + def downgrade() -> None: 77 + """Downgrade schema.""" 78 + op.drop_index("ix_jam_participants_jam_id", table_name="jam_participants") 79 + op.drop_index( 80 + "ix_jam_participants_did_active", 81 + table_name="jam_participants", 82 + postgresql_where=sa.text("left_at IS NULL"), 83 + ) 84 + op.drop_table("jam_participants") 85 + op.drop_index( 86 + "ix_jams_is_active", 87 + table_name="jams", 88 + postgresql_where=sa.text("is_active IS true"), 89 + ) 90 + op.drop_index("ix_jams_host_did", table_name="jams") 91 + op.drop_index("ix_jams_code", table_name="jams") 92 + op.drop_table("jams")

+2

backend/src/backend/_internal/__init__.py

··· 40 40 update_session_tokens, 41 41 ) 42 42 from backend._internal.constellation import get_like_count_safe 43 + from backend._internal.jams import jam_service 43 44 from backend._internal.feature_flags import ( 44 45 KNOWN_FLAGS, 45 46 disable_flag, ··· 81 82 "get_supported_artists", 82 83 "handle_oauth_callback", 83 84 "has_flag", 85 + "jam_service", 84 86 "list_developer_tokens", 85 87 "notification_service", 86 88 "now_playing_service",

+1

backend/src/backend/_internal/feature_flags.py

··· 12 12 # known flags - add new flags here for documentation 13 13 KNOWN_FLAGS = frozenset( 14 14 { 15 + "jams", # enable shared listening rooms 15 16 "lossless-uploads", # enable AIFF/FLAC upload support 16 17 "pds-audio-uploads", # enable PDS audio blob uploads 17 18 "vibe-search", # enable semantic vibe search in Cmd+K

+723

backend/src/backend/_internal/jams.py

··· 1 + """jam service for shared listening rooms. 2 + 3 + manages jam lifecycle, WebSocket connections, and Redis Streams 4 + for real-time playback sync across participants. 5 + """ 6 + 7 + import asyncio 8 + import contextlib 9 + import json 10 + import logging 11 + import secrets 12 + import string 13 + import time 14 + from datetime import UTC, datetime 15 + from typing import Any 16 + 17 + from fastapi import WebSocket 18 + from sqlalchemy import select, update 19 + from sqlalchemy.orm import selectinload 20 + 21 + from backend.models import Track 22 + from backend.models.jam import Jam, JamParticipant 23 + from backend.schemas import TrackResponse 24 + from backend.utilities.database import db_session 25 + from backend.utilities.redis import get_async_redis_client 26 + 27 + logger = logging.getLogger(__name__) 28 + 29 + CODE_ALPHABET = string.ascii_lowercase + string.digits 30 + CODE_LENGTH = 8 31 + MAX_CODE_ATTEMPTS = 10 32 + 33 + 34 + def _generate_code() -> str: 35 + """generate an 8-char alphanumeric code for jam URLs.""" 36 + return "".join(secrets.choice(CODE_ALPHABET) for _ in range(CODE_LENGTH)) 37 + 38 + 39 + def _empty_state() -> dict[str, Any]: 40 + """return an empty jam playback state.""" 41 + return { 42 + "track_ids": [], 43 + "current_index": 0, 44 + "current_track_id": None, 45 + "is_playing": False, 46 + "progress_ms": 0, 47 + "server_time_ms": int(time.time() * 1000), 48 + } 49 + 50 + 51 + class JamService: 52 + """service for managing jams with Redis Streams + WebSocket fan-out.""" 53 + 54 + def __init__(self) -> None: 55 + self._connections: dict[str, set[WebSocket]] = {} 56 + self._ws_by_did: dict[str, tuple[str, WebSocket]] = {} # did → (jam_id, ws) 57 + self._reader_tasks: dict[str, asyncio.Task] = {} 58 + 59 + async def setup(self) -> None: 60 + """initialize the jam service.""" 61 + logger.info("starting jam service") 62 + 63 + async def shutdown(self) -> None: 64 + """cleanup resources.""" 65 + logger.info("shutting down jam service") 66 + # cancel all reader tasks 67 + for _jam_id, task in self._reader_tasks.items(): 68 + task.cancel() 69 + with contextlib.suppress(asyncio.CancelledError): 70 + await task 71 + self._reader_tasks.clear() 72 + self._connections.clear() 73 + 74 + # ── jam lifecycle ────────────────────────────────────────────── 75 + 76 + async def create_jam( 77 + self, 78 + host_did: str, 79 + name: str | None = None, 80 + track_ids: list[str] | None = None, 81 + current_index: int = 0, 82 + is_playing: bool = False, 83 + progress_ms: int = 0, 84 + ) -> dict[str, Any]: 85 + """create a new jam. auto-leaves any existing jam the host is in.""" 86 + await self._auto_leave(host_did) 87 + 88 + state = _empty_state() 89 + if track_ids: 90 + state["track_ids"] = track_ids 91 + idx = min(current_index, len(track_ids) - 1) if track_ids else 0 92 + state["current_index"] = idx 93 + state["current_track_id"] = track_ids[idx] if track_ids else None 94 + state["is_playing"] = is_playing 95 + state["progress_ms"] = progress_ms 96 + state["server_time_ms"] = int(time.time() * 1000) 97 + 98 + async with db_session() as db: 99 + # generate unique code 100 + code = await self._generate_unique_code(db) 101 + 102 + jam = Jam( 103 + id=secrets.token_hex(16), 104 + code=code, 105 + host_did=host_did, 106 + name=name, 107 + state=state, 108 + revision=1, 109 + is_active=True, 110 + ) 111 + db.add(jam) 112 + 113 + # host joins automatically 114 + participant = JamParticipant( 115 + jam_id=jam.id, 116 + did=host_did, 117 + ) 118 + db.add(participant) 119 + await db.commit() 120 + await db.refresh(jam) 121 + 122 + tracks = await self._hydrate_tracks(db, state.get("track_ids", [])) 123 + 124 + return self._serialize_jam(jam, tracks=tracks) 125 + 126 + async def get_jam_by_code(self, code: str) -> dict[str, Any] | None: 127 + """get jam details by code.""" 128 + async with db_session() as db: 129 + jam = await self._fetch_jam_by_code(db, code) 130 + if not jam: 131 + return None 132 + tracks = await self._hydrate_tracks(db, jam.state.get("track_ids", [])) 133 + participants = await self._get_participants(db, jam.id) 134 + return self._serialize_jam(jam, tracks=tracks, participants=participants) 135 + 136 + async def join_jam(self, code: str, did: str) -> dict[str, Any] | None: 137 + """join a jam by code. auto-leaves any existing jam.""" 138 + await self._auto_leave(did) 139 + 140 + async with db_session() as db: 141 + jam = await self._fetch_jam_by_code(db, code) 142 + if not jam or not jam.is_active: 143 + return None 144 + 145 + # check if already participating 146 + existing = await db.execute( 147 + select(JamParticipant).where( 148 + JamParticipant.jam_id == jam.id, 149 + JamParticipant.did == did, 150 + JamParticipant.left_at.is_(None), 151 + ) 152 + ) 153 + if not existing.scalar_one_or_none(): 154 + participant = JamParticipant(jam_id=jam.id, did=did) 155 + db.add(participant) 156 + await db.commit() 157 + 158 + # publish participant event 159 + await self._publish_event( 160 + jam.id, 161 + { 162 + "type": "participant", 163 + "event": "joined", 164 + "did": did, 165 + "revision": jam.revision, 166 + }, 167 + ) 168 + 169 + tracks = await self._hydrate_tracks(db, jam.state.get("track_ids", [])) 170 + participants = await self._get_participants(db, jam.id) 171 + return self._serialize_jam(jam, tracks=tracks, participants=participants) 172 + 173 + async def leave_jam(self, jam_id: str, did: str) -> bool: 174 + """leave a jam. if last participant, end the jam.""" 175 + async with db_session() as db: 176 + # mark participant as left 177 + cursor = await db.execute( 178 + update(JamParticipant) 179 + .where( 180 + JamParticipant.jam_id == jam_id, 181 + JamParticipant.did == did, 182 + JamParticipant.left_at.is_(None), 183 + ) 184 + .values(left_at=datetime.now(UTC)) 185 + ) 186 + if cursor.rowcount == 0: # type: ignore[union-attr] 187 + return False 188 + 189 + # check remaining participants 190 + remaining = await db.execute( 191 + select(JamParticipant).where( 192 + JamParticipant.jam_id == jam_id, 193 + JamParticipant.left_at.is_(None), 194 + ) 195 + ) 196 + if not remaining.scalars().first(): 197 + # last participant left — end the jam 198 + await self._end_jam(db, jam_id) 199 + 200 + await db.commit() 201 + 202 + # publish participant event 203 + jam = await self._fetch_jam_by_id(db, jam_id) 204 + if jam: 205 + await self._publish_event( 206 + jam_id, 207 + { 208 + "type": "participant", 209 + "event": "left", 210 + "did": did, 211 + "revision": jam.revision, 212 + }, 213 + ) 214 + 215 + return True 216 + 217 + async def end_jam(self, jam_id: str, did: str) -> bool: 218 + """end a jam (host only).""" 219 + async with db_session() as db: 220 + jam = await self._fetch_jam_by_id(db, jam_id) 221 + if not jam or not jam.is_active: 222 + return False 223 + if jam.host_did != did: 224 + return False 225 + await self._end_jam(db, jam_id) 226 + await db.commit() 227 + return True 228 + 229 + async def get_active_jam(self, did: str) -> dict[str, Any] | None: 230 + """get the user's current active jam.""" 231 + async with db_session() as db: 232 + result = await db.execute( 233 + select(JamParticipant) 234 + .where( 235 + JamParticipant.did == did, 236 + JamParticipant.left_at.is_(None), 237 + ) 238 + .order_by(JamParticipant.joined_at.desc()) 239 + .limit(1) 240 + ) 241 + participant = result.scalar_one_or_none() 242 + if not participant: 243 + return None 244 + 245 + jam = await self._fetch_jam_by_id(db, participant.jam_id) 246 + if not jam or not jam.is_active: 247 + return None 248 + 249 + tracks = await self._hydrate_tracks(db, jam.state.get("track_ids", [])) 250 + participants = await self._get_participants(db, jam.id) 251 + return self._serialize_jam(jam, tracks=tracks, participants=participants) 252 + 253 + # ── playback commands ────────────────────────────────────────── 254 + 255 + async def handle_command( 256 + self, jam_id: str, did: str, command: dict[str, Any] 257 + ) -> dict[str, Any] | None: 258 + """process a playback command: mutate state, publish event.""" 259 + async with db_session() as db: 260 + # FOR UPDATE serializes concurrent commands on the same jam 261 + result = await db.execute( 262 + select(Jam).where(Jam.id == jam_id).with_for_update() 263 + ) 264 + jam = result.scalar_one_or_none() 265 + if not jam or not jam.is_active: 266 + return None 267 + 268 + # verify participant 269 + participant = await db.execute( 270 + select(JamParticipant).where( 271 + JamParticipant.jam_id == jam_id, 272 + JamParticipant.did == did, 273 + JamParticipant.left_at.is_(None), 274 + ) 275 + ) 276 + if not participant.scalar_one_or_none(): 277 + return None 278 + 279 + state = dict(jam.state) 280 + cmd_type = command.get("type") 281 + now_ms = int(time.time() * 1000) 282 + 283 + if cmd_type == "play": 284 + state["is_playing"] = True 285 + state["server_time_ms"] = now_ms 286 + elif cmd_type == "pause": 287 + # freeze progress at current interpolated position 288 + if state.get("is_playing"): 289 + elapsed = now_ms - state.get("server_time_ms", now_ms) 290 + state["progress_ms"] = state.get("progress_ms", 0) + elapsed 291 + state["is_playing"] = False 292 + state["server_time_ms"] = now_ms 293 + elif cmd_type == "seek": 294 + state["progress_ms"] = command.get("position_ms", 0) 295 + state["server_time_ms"] = now_ms 296 + elif cmd_type == "next": 297 + track_ids = state.get("track_ids", []) 298 + idx = state.get("current_index", 0) + 1 299 + if idx < len(track_ids): 300 + state["current_index"] = idx 301 + state["current_track_id"] = track_ids[idx] 302 + state["progress_ms"] = 0 303 + state["server_time_ms"] = now_ms 304 + elif cmd_type == "previous": 305 + idx = state.get("current_index", 0) - 1 306 + if idx >= 0: 307 + track_ids = state.get("track_ids", []) 308 + state["current_index"] = idx 309 + state["current_track_id"] = ( 310 + track_ids[idx] if idx < len(track_ids) else None 311 + ) 312 + state["progress_ms"] = 0 313 + state["server_time_ms"] = now_ms 314 + elif cmd_type == "add_tracks": 315 + new_ids = command.get("track_ids", []) 316 + state.setdefault("track_ids", []).extend(new_ids) 317 + # if was empty, set current 318 + if len(state["track_ids"]) == len(new_ids) and new_ids: 319 + state["current_index"] = 0 320 + state["current_track_id"] = new_ids[0] 321 + elif cmd_type == "play_track": 322 + # insert track after current and play it immediately 323 + file_id = command.get("file_id") 324 + if file_id: 325 + track_ids = state.get("track_ids", []) 326 + insert_at = state.get("current_index", 0) + 1 327 + track_ids.insert(insert_at, file_id) 328 + state["track_ids"] = track_ids 329 + state["current_index"] = insert_at 330 + state["current_track_id"] = file_id 331 + state["is_playing"] = True 332 + state["progress_ms"] = 0 333 + state["server_time_ms"] = now_ms 334 + elif cmd_type == "set_index": 335 + idx = command.get("index", 0) 336 + track_ids = state.get("track_ids", []) 337 + if 0 <= idx < len(track_ids): 338 + state["current_index"] = idx 339 + state["current_track_id"] = track_ids[idx] 340 + state["progress_ms"] = 0 341 + state["server_time_ms"] = now_ms 342 + elif cmd_type == "remove_track": 343 + idx = command.get("index") 344 + track_ids = state.get("track_ids", []) 345 + if idx is not None and 0 <= idx < len(track_ids): 346 + track_ids.pop(idx) 347 + state["track_ids"] = track_ids 348 + current = state.get("current_index", 0) 349 + if idx < current: 350 + state["current_index"] = current - 1 351 + elif idx == current: 352 + state["current_index"] = min(current, len(track_ids) - 1) 353 + # update current_track_id 354 + ci = state.get("current_index", 0) 355 + state["current_track_id"] = ( 356 + track_ids[ci] if track_ids and ci < len(track_ids) else None 357 + ) 358 + else: 359 + logger.warning("unknown jam command type: %s", cmd_type) 360 + return None 361 + 362 + # commit state 363 + jam.state = state 364 + jam.revision += 1 365 + jam.updated_at = datetime.now(UTC) 366 + await db.commit() 367 + await db.refresh(jam) 368 + 369 + # determine if tracks changed 370 + tracks_changed = cmd_type in ("add_tracks", "remove_track", "play_track") 371 + 372 + tracks: list[dict[str, Any]] = [] 373 + if tracks_changed: 374 + tracks = await self._hydrate_tracks(db, state.get("track_ids", [])) 375 + 376 + # publish to Redis stream 377 + event: dict[str, Any] = { 378 + "type": "state", 379 + "revision": jam.revision, 380 + "state": state, 381 + "tracks_changed": tracks_changed, 382 + "actor": {"did": did, "type": cmd_type}, 383 + } 384 + if tracks_changed: 385 + event["tracks"] = tracks 386 + await self._publish_event(jam_id, event) 387 + 388 + return { 389 + "state": state, 390 + "revision": jam.revision, 391 + "tracks": tracks, 392 + "tracks_changed": tracks_changed, 393 + } 394 + 395 + # ── WebSocket management ─────────────────────────────────────── 396 + 397 + async def connect_ws(self, jam_id: str, ws: WebSocket, did: str) -> None: 398 + """register a WebSocket connection for a jam.""" 399 + # close any previous socket for this user (e.g. stale from auto-leave) 400 + await self._close_ws_for_did(did) 401 + 402 + if jam_id not in self._connections: 403 + self._connections[jam_id] = set() 404 + self._connections[jam_id].add(ws) 405 + self._ws_by_did[did] = (jam_id, ws) 406 + 407 + # start reader task if first connection for this jam 408 + if jam_id not in self._reader_tasks or self._reader_tasks[jam_id].done(): 409 + self._reader_tasks[jam_id] = asyncio.create_task( 410 + self._stream_reader(jam_id) 411 + ) 412 + logger.info("started stream reader for jam %s", jam_id) 413 + 414 + async def disconnect_ws(self, jam_id: str, ws: WebSocket) -> None: 415 + """unregister a WebSocket connection.""" 416 + # clean up did tracking 417 + dids_to_remove = [ 418 + did 419 + for did, (jid, w) in self._ws_by_did.items() 420 + if jid == jam_id and w is ws 421 + ] 422 + for did in dids_to_remove: 423 + del self._ws_by_did[did] 424 + 425 + if jam_id in self._connections: 426 + self._connections[jam_id].discard(ws) 427 + if not self._connections[jam_id]: 428 + del self._connections[jam_id] 429 + # cancel reader if no more connections 430 + if jam_id in self._reader_tasks: 431 + self._reader_tasks[jam_id].cancel() 432 + with contextlib.suppress(asyncio.CancelledError): 433 + await self._reader_tasks[jam_id] 434 + del self._reader_tasks[jam_id] 435 + logger.info("stopped stream reader for jam %s", jam_id) 436 + 437 + async def _close_ws_for_did(self, did: str) -> None: 438 + """close any existing WebSocket for this DID.""" 439 + entry = self._ws_by_did.pop(did, None) 440 + if not entry: 441 + return 442 + old_jam_id, old_ws = entry 443 + if old_jam_id in self._connections: 444 + self._connections[old_jam_id].discard(old_ws) 445 + with contextlib.suppress(Exception): 446 + await old_ws.close(code=4010, reason="replaced by new connection") 447 + 448 + async def handle_ws_message( 449 + self, jam_id: str, did: str, message: dict[str, Any], ws: WebSocket 450 + ) -> None: 451 + """process an incoming WebSocket message.""" 452 + msg_type = message.get("type") 453 + 454 + if msg_type == "ping": 455 + await ws.send_json({"type": "pong"}) 456 + elif msg_type == "sync": 457 + await self._handle_sync(jam_id, did, message, ws) 458 + elif msg_type == "command": 459 + payload = message.get("payload", {}) 460 + result = await self.handle_command(jam_id, did, payload) 461 + if not result: 462 + await ws.send_json({"type": "error", "message": "command failed"}) 463 + else: 464 + await ws.send_json( 465 + {"type": "error", "message": f"unknown message type: {msg_type}"} 466 + ) 467 + 468 + async def _handle_sync( 469 + self, jam_id: str, did: str, message: dict[str, Any], ws: WebSocket 470 + ) -> None: 471 + """handle sync/reconnect request from a client.""" 472 + last_id = message.get("last_id") 473 + 474 + if not last_id: 475 + # full snapshot from DB 476 + async with db_session() as db: 477 + jam = await self._fetch_jam_by_id(db, jam_id) 478 + if not jam: 479 + await ws.send_json({"type": "error", "message": "jam not found"}) 480 + return 481 + tracks = await self._hydrate_tracks(db, jam.state.get("track_ids", [])) 482 + participants = await self._get_participants(db, jam.id) 483 + 484 + await ws.send_json( 485 + { 486 + "type": "state", 487 + "stream_id": None, 488 + "revision": jam.revision, 489 + "state": jam.state, 490 + "tracks": tracks, 491 + "tracks_changed": True, 492 + "participants": participants, 493 + "actor": {"did": "system", "type": "sync"}, 494 + } 495 + ) 496 + else: 497 + # replay from last_id 498 + try: 499 + redis = get_async_redis_client() 500 + stream_key = f"jam:{jam_id}:events" 501 + messages = await redis.xrange(stream_key, min=f"({last_id}", max="+") 502 + 503 + tracks_changed = False 504 + for msg_id, data in messages: 505 + payload = json.loads(data.get("payload", "{}")) 506 + if payload.get("tracks_changed"): 507 + tracks_changed = True 508 + await ws.send_json( 509 + { 510 + **payload, 511 + "stream_id": msg_id, 512 + } 513 + ) 514 + 515 + # if tracks changed during replay, send full track list 516 + if tracks_changed: 517 + async with db_session() as db: 518 + jam = await self._fetch_jam_by_id(db, jam_id) 519 + if jam: 520 + tracks = await self._hydrate_tracks( 521 + db, jam.state.get("track_ids", []) 522 + ) 523 + await ws.send_json( 524 + { 525 + "type": "state", 526 + "stream_id": None, 527 + "revision": jam.revision, 528 + "state": jam.state, 529 + "tracks": tracks, 530 + "tracks_changed": True, 531 + "actor": {"did": "system", "type": "sync"}, 532 + } 533 + ) 534 + except Exception: 535 + logger.exception("sync replay failed for jam %s", jam_id) 536 + # fall back to full snapshot 537 + await self._handle_sync(jam_id, did, {"last_id": None}, ws) 538 + 539 + # ── Redis Streams ────────────────────────────────────────────── 540 + 541 + async def _publish_event(self, jam_id: str, event: dict[str, Any]) -> None: 542 + """publish an event to the jam's Redis stream.""" 543 + try: 544 + redis = get_async_redis_client() 545 + stream_key = f"jam:{jam_id}:events" 546 + await redis.xadd( 547 + stream_key, 548 + {"payload": json.dumps(event)}, 549 + maxlen=1000, 550 + approximate=True, 551 + ) 552 + except Exception: 553 + logger.exception("failed to publish event for jam %s", jam_id) 554 + 555 + async def _stream_reader(self, jam_id: str) -> None: 556 + """background task that reads from a jam's Redis stream and fans out.""" 557 + redis = get_async_redis_client() 558 + stream_key = f"jam:{jam_id}:events" 559 + last_id = "$" 560 + 561 + while True: 562 + try: 563 + results = await redis.xread({stream_key: last_id}, block=5000, count=10) 564 + for _, messages in results or []: 565 + for msg_id, data in messages: 566 + last_id = msg_id 567 + payload = json.loads(data.get("payload", "{}")) 568 + payload["stream_id"] = msg_id 569 + await self._fan_out(jam_id, payload) 570 + except asyncio.CancelledError: 571 + logger.info("stream reader cancelled for jam %s", jam_id) 572 + break 573 + except Exception: 574 + logger.exception("error in stream reader for jam %s", jam_id) 575 + await asyncio.sleep(1) 576 + 577 + async def _fan_out(self, jam_id: str, payload: dict[str, Any]) -> None: 578 + """send a message to all connected WebSockets for a jam.""" 579 + connections = self._connections.get(jam_id, set()) 580 + dead: list[WebSocket] = [] 581 + 582 + for ws in connections: 583 + try: 584 + await ws.send_json(payload) 585 + except Exception: 586 + dead.append(ws) 587 + 588 + for ws in dead: 589 + connections.discard(ws) 590 + 591 + # ── internal helpers ─────────────────────────────────────────── 592 + 593 + async def _generate_unique_code(self, db: Any) -> str: 594 + """generate a unique jam code, retrying on collision.""" 595 + for _ in range(MAX_CODE_ATTEMPTS): 596 + code = _generate_code() 597 + result = await db.execute(select(Jam).where(Jam.code == code)) 598 + if not result.scalar_one_or_none(): 599 + return code 600 + raise RuntimeError("failed to generate unique jam code") 601 + 602 + async def _fetch_jam_by_code(self, db: Any, code: str) -> Jam | None: 603 + """fetch a jam by its short code.""" 604 + result = await db.execute(select(Jam).where(Jam.code == code)) 605 + return result.scalar_one_or_none() 606 + 607 + async def _fetch_jam_by_id(self, db: Any, jam_id: str) -> Jam | None: 608 + """fetch a jam by ID.""" 609 + result = await db.execute(select(Jam).where(Jam.id == jam_id)) 610 + return result.scalar_one_or_none() 611 + 612 + async def _end_jam(self, db: Any, jam_id: str) -> None: 613 + """mark a jam as ended.""" 614 + now = datetime.now(UTC) 615 + await db.execute( 616 + update(Jam) 617 + .where(Jam.id == jam_id) 618 + .values(is_active=False, ended_at=now, updated_at=now) 619 + ) 620 + # mark all remaining participants as left 621 + await db.execute( 622 + update(JamParticipant) 623 + .where( 624 + JamParticipant.jam_id == jam_id, 625 + JamParticipant.left_at.is_(None), 626 + ) 627 + .values(left_at=now) 628 + ) 629 + # trim Redis stream 630 + try: 631 + redis = get_async_redis_client() 632 + await redis.delete(f"jam:{jam_id}:events") 633 + except Exception: 634 + logger.exception("failed to trim stream for jam %s", jam_id) 635 + 636 + async def _auto_leave(self, did: str) -> None: 637 + """leave any existing active jam for this user.""" 638 + async with db_session() as db: 639 + result = await db.execute( 640 + select(JamParticipant).where( 641 + JamParticipant.did == did, 642 + JamParticipant.left_at.is_(None), 643 + ) 644 + ) 645 + for participant in result.scalars().all(): 646 + participant.left_at = datetime.now(UTC) 647 + await db.commit() 648 + 649 + async def _get_participants(self, db: Any, jam_id: str) -> list[dict[str, Any]]: 650 + """get active participants for a jam with artist info.""" 651 + from backend.models.artist import Artist 652 + 653 + result = await db.execute( 654 + select(JamParticipant, Artist) 655 + .join(Artist, JamParticipant.did == Artist.did) 656 + .where( 657 + JamParticipant.jam_id == jam_id, 658 + JamParticipant.left_at.is_(None), 659 + ) 660 + ) 661 + participants = [] 662 + for participant, artist in result.all(): 663 + participants.append( 664 + { 665 + "did": participant.did, 666 + "handle": artist.handle, 667 + "display_name": artist.display_name, 668 + "avatar_url": artist.avatar_url, 669 + } 670 + ) 671 + return participants 672 + 673 + async def _hydrate_tracks( 674 + self, db: Any, track_ids: list[str] 675 + ) -> list[dict[str, Any]]: 676 + """fetch track metadata for jam display, preserving order.""" 677 + if not track_ids: 678 + return [] 679 + 680 + stmt = ( 681 + select(Track) 682 + .options(selectinload(Track.artist), selectinload(Track.album_rel)) 683 + .where(Track.file_id.in_(track_ids)) 684 + ) 685 + result = await db.execute(stmt) 686 + tracks = result.scalars().all() 687 + track_by_file_id = {track.file_id: track for track in tracks} 688 + 689 + serialized: list[dict[str, Any]] = [] 690 + for file_id in track_ids: 691 + if track := track_by_file_id.get(file_id): 692 + track_response = await TrackResponse.from_track( 693 + track, pds_url=None, liked_track_ids=None, like_counts=None 694 + ) 695 + serialized.append(track_response.model_dump(mode="json")) 696 + 697 + return serialized 698 + 699 + def _serialize_jam( 700 + self, 701 + jam: Jam, 702 + tracks: list[dict[str, Any]] | None = None, 703 + participants: list[dict[str, Any]] | None = None, 704 + ) -> dict[str, Any]: 705 + """serialize a jam for API responses.""" 706 + return { 707 + "id": jam.id, 708 + "code": jam.code, 709 + "host_did": jam.host_did, 710 + "name": jam.name, 711 + "state": jam.state, 712 + "revision": jam.revision, 713 + "is_active": jam.is_active, 714 + "created_at": jam.created_at.isoformat(), 715 + "updated_at": jam.updated_at.isoformat(), 716 + "ended_at": jam.ended_at.isoformat() if jam.ended_at else None, 717 + "tracks": tracks or [], 718 + "participants": participants or [], 719 + } 720 + 721 + 722 + # global instance 723 + jam_service = JamService()

+2

backend/src/backend/api/__init__.py

··· 7 7 from backend.api.audio import router as audio_router 8 8 from backend.api.auth import router as auth_router 9 9 from backend.api.exports import router as exports_router 10 + from backend.api.jams import router as jams_router 10 11 from backend.api.moderation import router as moderation_router 11 12 from backend.api.now_playing import router as now_playing_router 12 13 from backend.api.oembed import router as oembed_router ··· 25 26 "auth_router", 26 27 "discover_router", 27 28 "exports_router", 29 + "jams_router", 28 30 "meta_router", 29 31 "moderation_router", 30 32 "now_playing_router",

+269

backend/src/backend/api/jams.py

··· 1 + """jam api endpoints for shared listening rooms.""" 2 + 3 + import json 4 + import logging 5 + from typing import Annotated, Any 6 + 7 + from fastapi import ( 8 + APIRouter, 9 + Cookie, 10 + Depends, 11 + HTTPException, 12 + WebSocket, 13 + WebSocketDisconnect, 14 + ) 15 + from pydantic import BaseModel, Field 16 + from sqlalchemy import select 17 + from sqlalchemy.ext.asyncio import AsyncSession 18 + 19 + from backend._internal import Session, has_flag, jam_service, require_auth 20 + from backend._internal.auth.session import get_session 21 + from backend.models import get_db 22 + from backend.models.jam import JamParticipant 23 + from backend.utilities.database import db_session 24 + 25 + logger = logging.getLogger(__name__) 26 + 27 + router = APIRouter(prefix="/jams", tags=["jams"]) 28 + 29 + JAMS_FLAG = "jams" 30 + 31 + 32 + # ── request/response models ─────────────────────────────────────── 33 + 34 + 35 + class CreateJamRequest(BaseModel): 36 + name: str | None = None 37 + track_ids: list[str] = Field(default_factory=list) 38 + current_index: int = 0 39 + is_playing: bool = False 40 + progress_ms: int = 0 41 + 42 + 43 + class CommandRequest(BaseModel): 44 + type: str 45 + position_ms: int | None = None 46 + track_ids: list[str] | None = None 47 + index: int | None = None 48 + 49 + 50 + class JamResponse(BaseModel): 51 + id: str 52 + code: str 53 + host_did: str 54 + name: str | None 55 + state: dict[str, Any] 56 + revision: int 57 + is_active: bool 58 + created_at: str 59 + updated_at: str 60 + ended_at: str | None 61 + tracks: list[dict[str, Any]] = Field(default_factory=list) 62 + participants: list[dict[str, Any]] = Field(default_factory=list) 63 + 64 + 65 + # ── flag check helper ────────────────────────────────────────────── 66 + 67 + 68 + async def _require_jams_flag(session: Session, db: AsyncSession) -> None: 69 + """raise 403 if user doesn't have the jams flag.""" 70 + if not await has_flag(db, session.did, JAMS_FLAG): 71 + raise HTTPException(status_code=403, detail="jams feature not enabled") 72 + 73 + 74 + # ── REST endpoints ───────────────────────────────────────────────── 75 + 76 + 77 + @router.post("/", response_model=JamResponse) 78 + async def create_jam( 79 + body: CreateJamRequest, 80 + db: Annotated[AsyncSession, Depends(get_db)], 81 + session: Session = Depends(require_auth), 82 + ) -> JamResponse: 83 + """create a new jam.""" 84 + await _require_jams_flag(session, db) 85 + result = await jam_service.create_jam( 86 + host_did=session.did, 87 + name=body.name, 88 + track_ids=body.track_ids, 89 + current_index=body.current_index, 90 + is_playing=body.is_playing, 91 + progress_ms=body.progress_ms, 92 + ) 93 + return JamResponse(**result) 94 + 95 + 96 + @router.get("/active", response_model=JamResponse | None) 97 + async def get_active_jam( 98 + db: Annotated[AsyncSession, Depends(get_db)], 99 + session: Session = Depends(require_auth), 100 + ) -> JamResponse | None: 101 + """get the user's current active jam.""" 102 + await _require_jams_flag(session, db) 103 + result = await jam_service.get_active_jam(session.did) 104 + if not result: 105 + return None 106 + return JamResponse(**result) 107 + 108 + 109 + @router.get("/{code}", response_model=JamResponse) 110 + async def get_jam( 111 + code: str, 112 + db: Annotated[AsyncSession, Depends(get_db)], 113 + session: Session = Depends(require_auth), 114 + ) -> JamResponse: 115 + """get jam details by code.""" 116 + await _require_jams_flag(session, db) 117 + result = await jam_service.get_jam_by_code(code) 118 + if not result: 119 + raise HTTPException(status_code=404, detail="jam not found") 120 + return JamResponse(**result) 121 + 122 + 123 + @router.post("/{code}/join", response_model=JamResponse) 124 + async def join_jam( 125 + code: str, 126 + db: Annotated[AsyncSession, Depends(get_db)], 127 + session: Session = Depends(require_auth), 128 + ) -> JamResponse: 129 + """join a jam.""" 130 + await _require_jams_flag(session, db) 131 + result = await jam_service.join_jam(code, session.did) 132 + if not result: 133 + raise HTTPException(status_code=404, detail="jam not found or not active") 134 + return JamResponse(**result) 135 + 136 + 137 + @router.post("/{code}/leave") 138 + async def leave_jam( 139 + code: str, 140 + db: Annotated[AsyncSession, Depends(get_db)], 141 + session: Session = Depends(require_auth), 142 + ) -> dict[str, bool]: 143 + """leave a jam.""" 144 + await _require_jams_flag(session, db) 145 + jam = await jam_service.get_jam_by_code(code) 146 + if not jam: 147 + raise HTTPException(status_code=404, detail="jam not found") 148 + success = await jam_service.leave_jam(jam["id"], session.did) 149 + if not success: 150 + raise HTTPException(status_code=400, detail="not in this jam") 151 + return {"ok": True} 152 + 153 + 154 + @router.post("/{code}/end") 155 + async def end_jam( 156 + code: str, 157 + db: Annotated[AsyncSession, Depends(get_db)], 158 + session: Session = Depends(require_auth), 159 + ) -> dict[str, bool]: 160 + """end a jam (host only).""" 161 + await _require_jams_flag(session, db) 162 + jam = await jam_service.get_jam_by_code(code) 163 + if not jam: 164 + raise HTTPException(status_code=404, detail="jam not found") 165 + success = await jam_service.end_jam(jam["id"], session.did) 166 + if not success: 167 + raise HTTPException(status_code=403, detail="only the host can end the jam") 168 + return {"ok": True} 169 + 170 + 171 + @router.post("/{code}/command") 172 + async def jam_command( 173 + code: str, 174 + body: CommandRequest, 175 + db: Annotated[AsyncSession, Depends(get_db)], 176 + session: Session = Depends(require_auth), 177 + ) -> dict[str, Any]: 178 + """send a playback command to the jam.""" 179 + await _require_jams_flag(session, db) 180 + jam = await jam_service.get_jam_by_code(code) 181 + if not jam: 182 + raise HTTPException(status_code=404, detail="jam not found") 183 + 184 + command: dict[str, Any] = {"type": body.type} 185 + if body.position_ms is not None: 186 + command["position_ms"] = body.position_ms 187 + if body.track_ids is not None: 188 + command["track_ids"] = body.track_ids 189 + if body.index is not None: 190 + command["index"] = body.index 191 + 192 + result = await jam_service.handle_command(jam["id"], session.did, command) 193 + if not result: 194 + raise HTTPException(status_code=400, detail="command failed") 195 + return result 196 + 197 + 198 + # ── WebSocket endpoint ───────────────────────────────────────────── 199 + 200 + 201 + async def _get_ws_session(ws: WebSocket) -> Session | None: 202 + """extract session from WebSocket cookies.""" 203 + session_id = ws.cookies.get("session_id") 204 + if not session_id: 205 + return None 206 + return await get_session(session_id) 207 + 208 + 209 + @router.websocket("/{code}/ws") 210 + async def jam_websocket( 211 + ws: WebSocket, 212 + code: str, 213 + session_id: Annotated[str | None, Cookie()] = None, 214 + ) -> None: 215 + """WebSocket endpoint for real-time jam sync.""" 216 + # authenticate via cookie 217 + if not session_id: 218 + await ws.close(code=4001, reason="authentication required") 219 + return 220 + 221 + session = await get_session(session_id) 222 + if not session: 223 + await ws.close(code=4001, reason="invalid session") 224 + return 225 + 226 + # verify flag 227 + async with db_session() as db: 228 + if not await has_flag(db, session.did, JAMS_FLAG): 229 + await ws.close(code=4003, reason="jams feature not enabled") 230 + return 231 + 232 + # look up jam 233 + jam = await jam_service.get_jam_by_code(code) 234 + if not jam or not jam["is_active"]: 235 + await ws.close(code=4004, reason="jam not found or ended") 236 + return 237 + 238 + jam_id = jam["id"] 239 + 240 + # verify participant membership before accepting 241 + async with db_session() as db: 242 + result = await db.execute( 243 + select(JamParticipant).where( 244 + JamParticipant.jam_id == jam_id, 245 + JamParticipant.did == session.did, 246 + JamParticipant.left_at.is_(None), 247 + ) 248 + ) 249 + if not result.scalar_one_or_none(): 250 + await ws.close(code=4003, reason="not a participant") 251 + return 252 + 253 + await ws.accept() 254 + 255 + try: 256 + await jam_service.connect_ws(jam_id, ws, session.did) 257 + while True: 258 + data = await ws.receive_text() 259 + try: 260 + message = json.loads(data) 261 + await jam_service.handle_ws_message(jam_id, session.did, message, ws) 262 + except json.JSONDecodeError: 263 + await ws.send_json({"type": "error", "message": "invalid JSON"}) 264 + except WebSocketDisconnect: 265 + logger.debug("ws disconnected from jam %s: %s", jam_id, session.did) 266 + except Exception: 267 + logger.exception("ws error in jam %s", jam_id) 268 + finally: 269 + await jam_service.disconnect_ws(jam_id, ws)

+8 -1

backend/src/backend/main.py

··· 12 12 from slowapi.errors import RateLimitExceeded 13 13 from slowapi.middleware import SlowAPIMiddleware 14 14 15 - from backend._internal import notification_service, queue_service 15 + from backend._internal import jam_service, notification_service, queue_service 16 16 from backend._internal.background import background_worker_lifespan 17 17 from backend.api import ( 18 18 account_router, ··· 21 21 auth_router, 22 22 discover_router, 23 23 exports_router, 24 + jams_router, 24 25 meta_router, 25 26 moderation_router, 26 27 now_playing_router, ··· 60 61 # setup services 61 62 await notification_service.setup() 62 63 await queue_service.setup() 64 + await jam_service.setup() 63 65 64 66 # start background task worker (docket) 65 67 async with background_worker_lifespan() as docket: ··· 76 78 await asyncio.wait_for(queue_service.shutdown(), timeout=2.0) 77 79 except TimeoutError: 78 80 logging.warning("queue_service.shutdown() timed out") 81 + try: 82 + await asyncio.wait_for(jam_service.shutdown(), timeout=2.0) 83 + except TimeoutError: 84 + logging.warning("jam_service.shutdown() timed out") 79 85 80 86 81 87 app = FastAPI( ··· 131 137 app.include_router(now_playing_router) 132 138 app.include_router(migration_router) 133 139 app.include_router(exports_router) 140 + app.include_router(jams_router) 134 141 app.include_router(pds_backfill_router) 135 142 app.include_router(moderation_router) 136 143 app.include_router(oembed_router)

+3

backend/src/backend/models/__init__.py

··· 7 7 from backend.models.sensitive_image import SensitiveImage 8 8 from backend.models.exchange_token import ExchangeToken 9 9 from backend.models.feature_flag import FeatureFlag 10 + from backend.models.jam import Jam, JamParticipant 10 11 from backend.models.job import Job 11 12 from backend.models.oauth_state import OAuthStateModel 12 13 from backend.models.pending_add_account import PendingAddAccount ··· 30 31 "CopyrightScan", 31 32 "ExchangeToken", 32 33 "FeatureFlag", 34 + "Jam", 35 + "JamParticipant", 33 36 "Job", 34 37 "OAuthStateModel", 35 38 "PendingAddAccount",

+78

backend/src/backend/models/jam.py

··· 1 + """jam models for shared listening rooms.""" 2 + 3 + from datetime import UTC, datetime 4 + 5 + from sqlalchemy import BigInteger, Boolean, DateTime, ForeignKey, Index, Integer, String 6 + from sqlalchemy.dialects.postgresql import JSON 7 + from sqlalchemy.orm import Mapped, mapped_column 8 + 9 + from backend.models.database import Base 10 + 11 + 12 + class Jam(Base): 13 + """shared listening room.""" 14 + 15 + __tablename__ = "jams" 16 + 17 + id: Mapped[str] = mapped_column(String, primary_key=True) 18 + code: Mapped[str] = mapped_column(String(12), unique=True, nullable=False) 19 + host_did: Mapped[str] = mapped_column( 20 + String, ForeignKey("artists.did"), nullable=False 21 + ) 22 + name: Mapped[str | None] = mapped_column(String(100), nullable=True) 23 + state: Mapped[dict] = mapped_column(JSON, nullable=False) 24 + revision: Mapped[int] = mapped_column(BigInteger, nullable=False, default=1) 25 + is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) 26 + created_at: Mapped[datetime] = mapped_column( 27 + DateTime(timezone=True), 28 + default=lambda: datetime.now(UTC), 29 + nullable=False, 30 + ) 31 + updated_at: Mapped[datetime] = mapped_column( 32 + DateTime(timezone=True), 33 + default=lambda: datetime.now(UTC), 34 + onupdate=lambda: datetime.now(UTC), 35 + nullable=False, 36 + ) 37 + ended_at: Mapped[datetime | None] = mapped_column( 38 + DateTime(timezone=True), nullable=True 39 + ) 40 + 41 + __table_args__ = ( 42 + Index("ix_jams_code", "code", unique=True), 43 + Index("ix_jams_host_did", "host_did"), 44 + Index( 45 + "ix_jams_is_active", 46 + "is_active", 47 + postgresql_where=(is_active.is_(True)), 48 + ), 49 + ) 50 + 51 + 52 + class JamParticipant(Base): 53 + """participant in a shared listening room.""" 54 + 55 + __tablename__ = "jam_participants" 56 + 57 + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) 58 + jam_id: Mapped[str] = mapped_column( 59 + String, ForeignKey("jams.id", ondelete="CASCADE"), nullable=False 60 + ) 61 + did: Mapped[str] = mapped_column(String, ForeignKey("artists.did"), nullable=False) 62 + joined_at: Mapped[datetime] = mapped_column( 63 + DateTime(timezone=True), 64 + default=lambda: datetime.now(UTC), 65 + nullable=False, 66 + ) 67 + left_at: Mapped[datetime | None] = mapped_column( 68 + DateTime(timezone=True), nullable=True 69 + ) 70 + 71 + __table_args__ = ( 72 + Index("ix_jam_participants_jam_id", "jam_id"), 73 + Index( 74 + "ix_jam_participants_did_active", 75 + "did", 76 + postgresql_where=(left_at.is_(None)), 77 + ), 78 + )

+583

backend/tests/api/test_jams.py

··· 1 + """tests for jam api endpoints.""" 2 + 3 + from collections.abc import AsyncGenerator 4 + from unittest.mock import AsyncMock 5 + 6 + import pytest 7 + from fastapi import FastAPI 8 + from httpx import ASGITransport, AsyncClient 9 + from sqlalchemy.ext.asyncio import AsyncSession 10 + 11 + from backend._internal import Session 12 + from backend._internal.feature_flags import enable_flag 13 + from backend._internal.jams import JamService 14 + from backend.main import app 15 + from backend.models import Artist 16 + 17 + 18 + class MockSession(Session): 19 + """mock session for auth bypass in tests.""" 20 + 21 + def __init__(self, did: str = "did:test:host"): 22 + self.did = did 23 + self.access_token = "test_token" 24 + self.refresh_token = "test_refresh" 25 + self.session_id = "test_session" 26 + self.handle = "test.host" 27 + self.oauth_session = {} 28 + 29 + 30 + @pytest.fixture 31 + async def test_app(db_session: AsyncSession) -> AsyncGenerator[FastAPI, None]: 32 + """create test app with mocked auth and jams flag.""" 33 + from backend._internal import require_auth 34 + 35 + mock_session = MockSession() 36 + 37 + async def mock_require_auth() -> Session: 38 + return mock_session 39 + 40 + app.dependency_overrides[require_auth] = mock_require_auth 41 + 42 + # create the test artist 43 + artist = Artist( 44 + did="did:test:host", 45 + handle="test.host", 46 + display_name="Test Host", 47 + ) 48 + db_session.add(artist) 49 + await db_session.flush() 50 + 51 + # enable jams flag for the test user 52 + await enable_flag(db_session, "did:test:host", "jams") 53 + await db_session.commit() 54 + 55 + yield app 56 + 57 + app.dependency_overrides.clear() 58 + 59 + 60 + @pytest.fixture 61 + async def second_user(db_session: AsyncSession) -> str: 62 + """create a second test artist with jams flag.""" 63 + artist = Artist( 64 + did="did:test:joiner", 65 + handle="test.joiner", 66 + display_name="Test Joiner", 67 + ) 68 + db_session.add(artist) 69 + await enable_flag(db_session, "did:test:joiner", "jams") 70 + await db_session.commit() 71 + return "did:test:joiner" 72 + 73 + 74 + async def test_create_jam(test_app: FastAPI, db_session: AsyncSession) -> None: 75 + """test POST /jams/ creates a jam.""" 76 + async with AsyncClient( 77 + transport=ASGITransport(app=test_app), base_url="http://test" 78 + ) as client: 79 + response = await client.post( 80 + "/jams/", 81 + json={"name": "test jam", "track_ids": ["track1", "track2"]}, 82 + ) 83 + 84 + assert response.status_code == 200 85 + data = response.json() 86 + assert data["name"] == "test jam" 87 + assert data["host_did"] == "did:test:host" 88 + assert data["is_active"] is True 89 + assert len(data["code"]) == 8 90 + assert data["state"]["track_ids"] == ["track1", "track2"] 91 + assert data["state"]["current_index"] == 0 92 + assert data["state"]["current_track_id"] == "track1" 93 + assert data["revision"] == 1 94 + 95 + 96 + async def test_create_jam_empty(test_app: FastAPI, db_session: AsyncSession) -> None: 97 + """test creating a jam with no tracks.""" 98 + async with AsyncClient( 99 + transport=ASGITransport(app=test_app), base_url="http://test" 100 + ) as client: 101 + response = await client.post("/jams/", json={}) 102 + 103 + assert response.status_code == 200 104 + data = response.json() 105 + assert data["state"]["track_ids"] == [] 106 + assert data["state"]["is_playing"] is False 107 + 108 + 109 + async def test_get_jam_by_code(test_app: FastAPI, db_session: AsyncSession) -> None: 110 + """test GET /jams/{code} returns jam details.""" 111 + async with AsyncClient( 112 + transport=ASGITransport(app=test_app), base_url="http://test" 113 + ) as client: 114 + create_response = await client.post("/jams/", json={"name": "get test"}) 115 + code = create_response.json()["code"] 116 + 117 + response = await client.get(f"/jams/{code}") 118 + 119 + assert response.status_code == 200 120 + data = response.json() 121 + assert data["code"] == code 122 + assert data["name"] == "get test" 123 + 124 + 125 + async def test_get_jam_not_found(test_app: FastAPI, db_session: AsyncSession) -> None: 126 + """test GET /jams/{code} returns 404 for unknown code.""" 127 + async with AsyncClient( 128 + transport=ASGITransport(app=test_app), base_url="http://test" 129 + ) as client: 130 + response = await client.get("/jams/nonexist") 131 + 132 + assert response.status_code == 404 133 + 134 + 135 + async def test_join_jam( 136 + test_app: FastAPI, db_session: AsyncSession, second_user: str 137 + ) -> None: 138 + """test POST /jams/{code}/join adds a participant.""" 139 + from backend._internal import require_auth 140 + 141 + # create jam as host 142 + async with AsyncClient( 143 + transport=ASGITransport(app=test_app), base_url="http://test" 144 + ) as client: 145 + create_response = await client.post("/jams/", json={"name": "join test"}) 146 + code = create_response.json()["code"] 147 + 148 + # switch to second user 149 + async def mock_joiner_auth() -> Session: 150 + return MockSession(did=second_user) 151 + 152 + app.dependency_overrides[require_auth] = mock_joiner_auth 153 + 154 + async with AsyncClient( 155 + transport=ASGITransport(app=test_app), base_url="http://test" 156 + ) as client: 157 + response = await client.post(f"/jams/{code}/join") 158 + 159 + assert response.status_code == 200 160 + data = response.json() 161 + assert len(data["participants"]) == 2 162 + participant_dids = {p["did"] for p in data["participants"]} 163 + assert "did:test:host" in participant_dids 164 + assert second_user in participant_dids 165 + 166 + 167 + async def test_leave_jam(test_app: FastAPI, db_session: AsyncSession) -> None: 168 + """test POST /jams/{code}/leave removes participant.""" 169 + async with AsyncClient( 170 + transport=ASGITransport(app=test_app), base_url="http://test" 171 + ) as client: 172 + create_response = await client.post("/jams/", json={"name": "leave test"}) 173 + code = create_response.json()["code"] 174 + 175 + response = await client.post(f"/jams/{code}/leave") 176 + 177 + assert response.status_code == 200 178 + assert response.json()["ok"] is True 179 + 180 + 181 + async def test_leave_ends_jam_when_last( 182 + test_app: FastAPI, db_session: AsyncSession 183 + ) -> None: 184 + """test that leaving as last participant ends the jam.""" 185 + async with AsyncClient( 186 + transport=ASGITransport(app=test_app), base_url="http://test" 187 + ) as client: 188 + create_response = await client.post("/jams/", json={"name": "last leave test"}) 189 + code = create_response.json()["code"] 190 + 191 + # leave (only participant) 192 + await client.post(f"/jams/{code}/leave") 193 + 194 + # jam should no longer be active 195 + get_response = await client.get(f"/jams/{code}") 196 + 197 + assert get_response.status_code == 200 198 + assert get_response.json()["is_active"] is False 199 + 200 + 201 + async def test_end_jam_host_only( 202 + test_app: FastAPI, db_session: AsyncSession, second_user: str 203 + ) -> None: 204 + """test that only the host can end a jam.""" 205 + from backend._internal import require_auth 206 + 207 + # create jam as host 208 + async with AsyncClient( 209 + transport=ASGITransport(app=test_app), base_url="http://test" 210 + ) as client: 211 + create_response = await client.post("/jams/", json={"name": "end test"}) 212 + code = create_response.json()["code"] 213 + 214 + # switch to second user and try to end 215 + async def mock_joiner_auth() -> Session: 216 + return MockSession(did=second_user) 217 + 218 + app.dependency_overrides[require_auth] = mock_joiner_auth 219 + 220 + async with AsyncClient( 221 + transport=ASGITransport(app=test_app), base_url="http://test" 222 + ) as client: 223 + # join first 224 + await client.post(f"/jams/{code}/join") 225 + # try to end 226 + response = await client.post(f"/jams/{code}/end") 227 + 228 + assert response.status_code == 403 229 + 230 + 231 + async def test_end_jam_by_host(test_app: FastAPI, db_session: AsyncSession) -> None: 232 + """test host can end their jam.""" 233 + async with AsyncClient( 234 + transport=ASGITransport(app=test_app), base_url="http://test" 235 + ) as client: 236 + create_response = await client.post("/jams/", json={"name": "host end test"}) 237 + code = create_response.json()["code"] 238 + 239 + response = await client.post(f"/jams/{code}/end") 240 + 241 + assert response.status_code == 200 242 + assert response.json()["ok"] is True 243 + 244 + 245 + async def test_command_play_pause(test_app: FastAPI, db_session: AsyncSession) -> None: 246 + """test play and pause commands.""" 247 + async with AsyncClient( 248 + transport=ASGITransport(app=test_app), base_url="http://test" 249 + ) as client: 250 + create_response = await client.post( 251 + "/jams/", 252 + json={"track_ids": ["t1"]}, 253 + ) 254 + code = create_response.json()["code"] 255 + 256 + # play 257 + play_response = await client.post( 258 + f"/jams/{code}/command", json={"type": "play"} 259 + ) 260 + assert play_response.status_code == 200 261 + assert play_response.json()["state"]["is_playing"] is True 262 + 263 + # pause 264 + pause_response = await client.post( 265 + f"/jams/{code}/command", json={"type": "pause"} 266 + ) 267 + assert pause_response.status_code == 200 268 + assert pause_response.json()["state"]["is_playing"] is False 269 + 270 + 271 + async def test_command_seek(test_app: FastAPI, db_session: AsyncSession) -> None: 272 + """test seek command.""" 273 + async with AsyncClient( 274 + transport=ASGITransport(app=test_app), base_url="http://test" 275 + ) as client: 276 + create_response = await client.post( 277 + "/jams/", 278 + json={"track_ids": ["t1"]}, 279 + ) 280 + code = create_response.json()["code"] 281 + 282 + response = await client.post( 283 + f"/jams/{code}/command", 284 + json={"type": "seek", "position_ms": 30000}, 285 + ) 286 + 287 + assert response.status_code == 200 288 + assert response.json()["state"]["progress_ms"] == 30000 289 + 290 + 291 + async def test_command_next_previous( 292 + test_app: FastAPI, db_session: AsyncSession 293 + ) -> None: 294 + """test next and previous commands.""" 295 + async with AsyncClient( 296 + transport=ASGITransport(app=test_app), base_url="http://test" 297 + ) as client: 298 + create_response = await client.post( 299 + "/jams/", 300 + json={"track_ids": ["t1", "t2", "t3"]}, 301 + ) 302 + code = create_response.json()["code"] 303 + 304 + # next 305 + next_response = await client.post( 306 + f"/jams/{code}/command", json={"type": "next"} 307 + ) 308 + assert next_response.json()["state"]["current_index"] == 1 309 + assert next_response.json()["state"]["current_track_id"] == "t2" 310 + 311 + # next again 312 + next2_response = await client.post( 313 + f"/jams/{code}/command", json={"type": "next"} 314 + ) 315 + assert next2_response.json()["state"]["current_index"] == 2 316 + 317 + # previous 318 + prev_response = await client.post( 319 + f"/jams/{code}/command", json={"type": "previous"} 320 + ) 321 + assert prev_response.json()["state"]["current_index"] == 1 322 + assert prev_response.json()["state"]["current_track_id"] == "t2" 323 + 324 + 325 + async def test_command_add_tracks(test_app: FastAPI, db_session: AsyncSession) -> None: 326 + """test add_tracks command.""" 327 + async with AsyncClient( 328 + transport=ASGITransport(app=test_app), base_url="http://test" 329 + ) as client: 330 + create_response = await client.post("/jams/", json={"track_ids": ["t1"]}) 331 + code = create_response.json()["code"] 332 + 333 + response = await client.post( 334 + f"/jams/{code}/command", 335 + json={"type": "add_tracks", "track_ids": ["t2", "t3"]}, 336 + ) 337 + 338 + assert response.status_code == 200 339 + assert response.json()["state"]["track_ids"] == ["t1", "t2", "t3"] 340 + assert response.json()["tracks_changed"] is True 341 + 342 + 343 + async def test_command_remove_track( 344 + test_app: FastAPI, db_session: AsyncSession 345 + ) -> None: 346 + """test remove_track command.""" 347 + async with AsyncClient( 348 + transport=ASGITransport(app=test_app), base_url="http://test" 349 + ) as client: 350 + create_response = await client.post( 351 + "/jams/", 352 + json={"track_ids": ["t1", "t2", "t3"]}, 353 + ) 354 + code = create_response.json()["code"] 355 + 356 + response = await client.post( 357 + f"/jams/{code}/command", 358 + json={"type": "remove_track", "index": 1}, 359 + ) 360 + 361 + assert response.status_code == 200 362 + assert response.json()["state"]["track_ids"] == ["t1", "t3"] 363 + assert response.json()["tracks_changed"] is True 364 + 365 + 366 + async def test_revision_monotonicity( 367 + test_app: FastAPI, db_session: AsyncSession 368 + ) -> None: 369 + """test that revision increases monotonically.""" 370 + async with AsyncClient( 371 + transport=ASGITransport(app=test_app), base_url="http://test" 372 + ) as client: 373 + create_response = await client.post( 374 + "/jams/", 375 + json={"track_ids": ["t1"]}, 376 + ) 377 + code = create_response.json()["code"] 378 + rev = create_response.json()["revision"] 379 + assert rev == 1 380 + 381 + for _ in range(5): 382 + cmd_response = await client.post( 383 + f"/jams/{code}/command", json={"type": "play"} 384 + ) 385 + new_rev = cmd_response.json()["revision"] 386 + assert new_rev > rev 387 + rev = new_rev 388 + 389 + 390 + async def test_auto_leave_previous_jam( 391 + test_app: FastAPI, db_session: AsyncSession 392 + ) -> None: 393 + """test that creating a new jam auto-leaves the previous one.""" 394 + async with AsyncClient( 395 + transport=ASGITransport(app=test_app), base_url="http://test" 396 + ) as client: 397 + # create first jam 398 + first_response = await client.post("/jams/", json={"name": "first jam"}) 399 + first_code = first_response.json()["code"] 400 + 401 + # create second jam (should auto-leave first) 402 + second_response = await client.post("/jams/", json={"name": "second jam"}) 403 + second_code = second_response.json()["code"] 404 + 405 + assert first_code != second_code 406 + 407 + # check active jam is the second one 408 + active_response = await client.get("/jams/active") 409 + 410 + assert active_response.status_code == 200 411 + assert active_response.json()["code"] == second_code 412 + 413 + 414 + async def test_flag_gating(test_app: FastAPI, db_session: AsyncSession) -> None: 415 + """test that users without the jams flag get 403.""" 416 + from backend._internal import require_auth 417 + 418 + # create a user without the flag 419 + no_flag_artist = Artist( 420 + did="did:test:noflag", 421 + handle="test.noflag", 422 + display_name="No Flag", 423 + ) 424 + db_session.add(no_flag_artist) 425 + await db_session.commit() 426 + 427 + async def mock_noflag_auth() -> Session: 428 + return MockSession(did="did:test:noflag") 429 + 430 + app.dependency_overrides[require_auth] = mock_noflag_auth 431 + 432 + async with AsyncClient( 433 + transport=ASGITransport(app=test_app), base_url="http://test" 434 + ) as client: 435 + response = await client.post("/jams/", json={}) 436 + 437 + assert response.status_code == 403 438 + assert "not enabled" in response.json()["detail"] 439 + 440 + 441 + async def test_get_active_jam_none(test_app: FastAPI, db_session: AsyncSession) -> None: 442 + """test GET /jams/active returns null when not in a jam.""" 443 + async with AsyncClient( 444 + transport=ASGITransport(app=test_app), base_url="http://test" 445 + ) as client: 446 + response = await client.get("/jams/active") 447 + 448 + assert response.status_code == 200 449 + assert response.json() is None 450 + 451 + 452 + async def test_code_uniqueness(test_app: FastAPI, db_session: AsyncSession) -> None: 453 + """test that each jam gets a unique code.""" 454 + codes = set() 455 + async with AsyncClient( 456 + transport=ASGITransport(app=test_app), base_url="http://test" 457 + ) as client: 458 + for i in range(5): 459 + response = await client.post("/jams/", json={"name": f"jam {i}"}) 460 + assert response.status_code == 200 461 + codes.add(response.json()["code"]) 462 + 463 + assert len(codes) == 5 464 + 465 + 466 + async def test_command_set_index(test_app: FastAPI, db_session: AsyncSession) -> None: 467 + """test set_index command jumps to a specific track.""" 468 + async with AsyncClient( 469 + transport=ASGITransport(app=test_app), base_url="http://test" 470 + ) as client: 471 + create_response = await client.post( 472 + "/jams/", 473 + json={"track_ids": ["t1", "t2", "t3", "t4"]}, 474 + ) 475 + code = create_response.json()["code"] 476 + 477 + # jump to index 2 478 + response = await client.post( 479 + f"/jams/{code}/command", json={"type": "set_index", "index": 2} 480 + ) 481 + assert response.status_code == 200 482 + state = response.json()["state"] 483 + assert state["current_index"] == 2 484 + assert state["current_track_id"] == "t3" 485 + assert state["progress_ms"] == 0 486 + 487 + # jump to index 0 488 + response = await client.post( 489 + f"/jams/{code}/command", json={"type": "set_index", "index": 0} 490 + ) 491 + assert response.status_code == 200 492 + state = response.json()["state"] 493 + assert state["current_index"] == 0 494 + assert state["current_track_id"] == "t1" 495 + 496 + # out-of-bounds index is a no-op (state unchanged, but command succeeds) 497 + response = await client.post( 498 + f"/jams/{code}/command", json={"type": "set_index", "index": 10} 499 + ) 500 + assert response.status_code == 200 501 + assert response.json()["state"]["current_index"] == 0 502 + assert response.json()["state"]["current_track_id"] == "t1" 503 + 504 + 505 + async def test_command_non_participant_rejected( 506 + test_app: FastAPI, db_session: AsyncSession, second_user: str 507 + ) -> None: 508 + """test that non-participants cannot send commands.""" 509 + from backend._internal import require_auth 510 + 511 + # create jam as host 512 + async with AsyncClient( 513 + transport=ASGITransport(app=test_app), base_url="http://test" 514 + ) as client: 515 + create_response = await client.post("/jams/", json={"track_ids": ["t1", "t2"]}) 516 + code = create_response.json()["code"] 517 + 518 + # switch to second user (NOT joined) 519 + async def mock_joiner_auth() -> Session: 520 + return MockSession(did=second_user) 521 + 522 + app.dependency_overrides[require_auth] = mock_joiner_auth 523 + 524 + async with AsyncClient( 525 + transport=ASGITransport(app=test_app), base_url="http://test" 526 + ) as client: 527 + response = await client.post(f"/jams/{code}/command", json={"type": "next"}) 528 + 529 + # command should fail — not a participant 530 + assert response.status_code == 400 531 + 532 + 533 + async def test_sequential_commands_get_distinct_revisions( 534 + test_app: FastAPI, db_session: AsyncSession 535 + ) -> None: 536 + """test that each command gets a distinct revision (verifies no clobbering).""" 537 + async with AsyncClient( 538 + transport=ASGITransport(app=test_app), base_url="http://test" 539 + ) as client: 540 + create_response = await client.post( 541 + "/jams/", json={"track_ids": ["t1", "t2", "t3"]} 542 + ) 543 + code = create_response.json()["code"] 544 + assert create_response.json()["revision"] == 1 545 + 546 + # send two next commands back-to-back 547 + r1 = await client.post(f"/jams/{code}/command", json={"type": "next"}) 548 + r2 = await client.post(f"/jams/{code}/command", json={"type": "next"}) 549 + 550 + assert r1.json()["revision"] == 2 551 + assert r2.json()["revision"] == 3 552 + # both advanced the index correctly 553 + assert r1.json()["state"]["current_index"] == 1 554 + assert r2.json()["state"]["current_index"] == 2 555 + 556 + 557 + async def test_did_socket_replacement() -> None: 558 + """test that connecting a second WS for the same DID closes the first.""" 559 + from starlette.websockets import WebSocket 560 + 561 + service = JamService() 562 + jam_id = "test-jam-123" 563 + 564 + ws1 = AsyncMock(spec=WebSocket) 565 + ws2 = AsyncMock(spec=WebSocket) 566 + 567 + # connect first socket 568 + await service.connect_ws(jam_id, ws1, "did:test:user") 569 + assert jam_id in service._connections 570 + assert ws1 in service._connections[jam_id] 571 + 572 + # connect second socket for same DID — should close first 573 + await service.connect_ws(jam_id, ws2, "did:test:user") 574 + 575 + # first socket should have been closed with code 4010 576 + ws1.close.assert_awaited_once_with(code=4010, reason="replaced by new connection") 577 + 578 + # only second socket should be in connections 579 + assert ws2 in service._connections[jam_id] 580 + assert ws1 not in service._connections[jam_id] 581 + 582 + # DID mapping should point to second socket 583 + assert service._ws_by_did["did:test:user"] == (jam_id, ws2)

+106

docs/architecture/jams-queue-integration.md

··· 1 + # jams + queue integration 2 + 3 + ## big picture 4 + 5 + a jam is "your queue, but shared." when you start a jam, your queue becomes a shared space. anyone in the jam can play a track, skip, reorder, add tracks — same as they would with their own queue. the player footer handles all playback. there's no separate "jam page" or "jam player." 6 + 7 + from the user's perspective, nothing changes about how they interact with the app. they click a track, it plays. they hit next, it skips. the only difference is that everyone in the jam sees and hears the same thing. 8 + 9 + ## the problem 10 + 11 + the current implementation sprinkles `if (jam.active)` conditionals everywhere: 12 + 13 + - `playback.svelte.ts`: `if (jam.active) jam.playTrack() else queue.playNow()` 14 + - `PlaybackControls.svelte`: `if (jam.active) jam.play() else player.togglePlayPause()` 15 + - `Queue.svelte`: `if (jam.active) jam.setIndex() else goToIndex()` 16 + - `Player.svelte`: jam-pause-sync effect, drift correction effect 17 + 18 + every file that touches the queue needs to know about jams. every new feature that interacts with playback needs jam-awareness. this doesn't scale and it's fragile — we already missed `playback.svelte.ts` entirely, which meant clicking a track in the feed did nothing during a jam. 19 + 20 + ## the constraint 21 + 22 + circular imports. `jam.svelte.ts` already imports `queue.svelte.ts` (for `stopPositionSave`/`startPositionSave`). so `queue.svelte.ts` can't import `jam.svelte.ts` back. 23 + 24 + ## the design 25 + 26 + the queue is the single integration point. when a jam is active, the queue routes mutations through the jam's WebSocket instead of local state + REST. callers never know. 27 + 28 + ### bridge pattern 29 + 30 + ``` 31 + ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ 32 + │ playback.ts │────>│ queue │────>│ jam (WS) │ 33 + │ controls │ │ .playNow() │ │ .playTrack()│ 34 + │ Queue.svelte│ │ .next() │ │ .next() │ 35 + │ feed clicks │ │ .addTracks()│ │ .addTracks()│ 36 + └──────────────┘ └──────────────┘ └──────────────┘ 37 + callers single gate transport 38 + (unchanged) (jam-aware) (registered) 39 + ``` 40 + 41 + `jam.svelte.ts` registers a bridge with the queue when a jam starts: 42 + 43 + ```typescript 44 + // jam.svelte.ts — on create/join 45 + queue.setJamBridge({ 46 + playTrack: (fileId) => this.sendCommand({ type: 'play_track', file_id: fileId }), 47 + next: () => this.sendCommand({ type: 'next' }), 48 + previous: () => this.sendCommand({ type: 'previous' }), 49 + addTracks: (ids) => this.sendCommand({ type: 'add_tracks', track_ids: ids }), 50 + removeTrack: (idx) => this.sendCommand({ type: 'remove_track', index: idx }), 51 + setIndex: (idx) => this.sendCommand({ type: 'set_index', index: idx }), 52 + seek: (ms) => this.sendCommand({ type: 'seek', position_ms: ms }), 53 + play: () => this.sendCommand({ type: 'play' }), 54 + pause: () => this.sendCommand({ type: 'pause' }), 55 + }); 56 + 57 + // jam.svelte.ts — on leave/destroy 58 + queue.setJamBridge(null); 59 + ``` 60 + 61 + queue methods check the bridge first: 62 + 63 + ```typescript 64 + // queue.svelte.ts 65 + playNow(track: Track) { 66 + if (this.jamBridge) { 67 + this.jamBridge.playTrack(track.file_id); 68 + return; 69 + } 70 + // ... normal local logic 71 + } 72 + ``` 73 + 74 + ### what this fixes 75 + 76 + - `playback.svelte.ts` — no jam imports, no conditionals. `queue.playNow()` just works. 77 + - `PlaybackControls.svelte` — calls `queue.next()` / `player.togglePlayPause()`. queue handles routing. 78 + - `Queue.svelte` — calls `goToIndex()` / `queue.removeTrack()`. no jam conditionals. 79 + - feed track clicks — already go through `playback.svelte.ts` → `queue.playNow()`. automatically jam-aware. 80 + 81 + ### state flow (jam active) 82 + 83 + incoming jam state (from WebSocket) updates the queue's reactive state directly: 84 + 85 + 1. jam WS message arrives with new state 86 + 2. `jam.handleStateMessage()` updates `jam.tracks`, `jam.currentIndex`, `jam.isPlaying`, etc. 87 + 3. queue reads from jam state when bridge is set (or jam pushes state into queue) 88 + 4. Player.svelte's existing queue-to-player sync picks it up 89 + 90 + ### what stays the same 91 + 92 + - the `<audio>` element and Player.svelte's track loading / play-pause sync 93 + - the backend jam service (commands, Redis Streams, WebSocket fan-out) 94 + - the jam UI in Queue.svelte (rainbow border, participants, share/leave) 95 + - the `/jam/[code]` join redirect page 96 + 97 + ### what changes 98 + 99 + | file | change | 100 + |------|--------| 101 + | `queue.svelte.ts` | add `JamBridge` interface + `setJamBridge()`, check bridge in mutation methods | 102 + | `jam.svelte.ts` | register/unregister bridge on create/join/leave/destroy | 103 + | `playback.svelte.ts` | remove jam import + conditionals (revert to original) | 104 + | `PlaybackControls.svelte` | remove jam conditionals, use queue/player directly | 105 + | `Queue.svelte` | remove jam routing in click/remove handlers, use queue methods | 106 + | `Player.svelte` | simplify jam effects (may still need pause-sync + drift correction) |

+115

docs/architecture/jams-transport-decision.md

··· 1 + # jams transport layer decision: Redis Streams 2 + 3 + ## status 4 + 5 + recommended for implementation in jams v1. 6 + 7 + ## decision 8 + 9 + use **Redis Streams** as the transport for jam state fan-out and reconnect catch-up. 10 + 11 + ## why this decision 12 + 13 + ### 1) reconnect behavior matters most on mobile PWA 14 + 15 + mobile websocket disconnects are normal (backgrounding, network handoffs, weak signal). streams let clients resume from `last_id` and replay only missed updates instead of forcing a separate full-state sync flow every reconnect. 16 + 17 + ### 2) one code path for live + catch-up 18 + 19 + with streams, the same primitive (`XREAD`) handles: 20 + - live updates while connected 21 + - short reconnect gaps 22 + - late join bootstrap from latest snapshot event 23 + 24 + that is simpler to reason about than pub/sub plus a separate state hash sync contract. 25 + 26 + ### 3) queue updates are order-dependent 27 + 28 + play/pause/seek is mostly last-writer-wins, but queue operations (add/remove/reorder) are order-sensitive. streams provide strict per-stream ordering with message ids. 29 + 30 + ### 4) streams are already a production pattern here 31 + 32 + this is the key correction from earlier assumptions: Docket already uses Redis Streams heavily. 33 + 34 + - stream/group setup: `backend/.venv/lib/python3.14/site-packages/docket/docket.py` 35 + - `XREADGROUP` / `XAUTOCLAIM` worker loop: `backend/.venv/lib/python3.14/site-packages/docket/worker.py` 36 + - `XADD` in scheduler pipeline: `backend/.venv/lib/python3.14/site-packages/docket/worker.py` 37 + 38 + so streams are not new operational territory for this stack. pub/sub also exists in Docket, but mainly for cancellation signaling, not the main work queue. 39 + 40 + ## v1 design sketch 41 + 42 + ### key model 43 + 44 + - stream key: `jam:{jam_id}:events` 45 + - trimming: `MAXLEN ~ 1000` (tunable) 46 + - payload strategy: emit **authoritative snapshot events** (not tiny diffs) 47 + 48 + snapshot payload fields (example): 49 + - `revision` (monotonic server integer) 50 + - `track_file_id` 51 + - `progress_ms` 52 + - `is_playing` 53 + - `updated_at` 54 + - queue snapshot or queue revision pointer 55 + - actor metadata (`did`, command type) 56 + 57 + ### websocket protocol 58 + 59 + client hello: 60 + - `{ "type": "sync", "last_id": "<redis-stream-id-or-null>" }` 61 + 62 + server behavior: 63 + 1. if `last_id` is present and still retained, replay `(last_id, +]` in order. 64 + 2. if `last_id` missing or trimmed, send latest snapshot event via `XREVRANGE ... COUNT 1`. 65 + 3. then continue tailing for new events. 66 + 67 + ### publish path 68 + 69 + on authoritative jam command (play/pause/seek/skip/queue mutation): 70 + 1. validate + apply server state mutation 71 + 2. increment `revision` 72 + 3. `XADD jam:{id}:events MAXLEN ~1000 * fields...` 73 + 4. fan out to connected ws clients 74 + 75 + ### read/tail pattern 76 + 77 + - no consumer groups for jam fan-out 78 + - each connection tracks its own offset (`last_id`) 79 + - use blocking `XREAD` loop per active jam reader task on each backend instance 80 + 81 + consumer groups are unnecessary here because this is broadcast, not work distribution. 82 + 83 + ### conflict and idempotency rules 84 + 85 + - `revision` is the source of truth; clients ignore stale revisions 86 + - operations are server-authoritative; client commands are requests, not direct state writes 87 + - reconnect replay can safely include already-applied revisions 88 + 89 + ## trade-offs accepted 90 + 91 + - higher implementation complexity than bare pub/sub 92 + - bounded storage overhead for recent history 93 + - need explicit trimming policy and fallback when requested `last_id` has been trimmed 94 + 95 + these are acceptable given the reconnect reliability gains and the existing Docket stream precedent. 96 + 97 + ## non-goals for v1 98 + 99 + - no long-term jam event history product surface 100 + - no consumer-group coordination 101 + - no cross-jam analytics pipeline from this stream 102 + 103 + ## rollout and safeguards 104 + 105 + 1. ship behind feature flag for jams. 106 + 2. instrument: 107 + - reconnect count per session 108 + - replayed event count per reconnect 109 + - trimmed-last-id fallback count 110 + - end-to-end command-to-fanout latency 111 + 3. set alert thresholds if fallback frequency or fanout latency spikes. 112 + 113 + ## fallback plan 114 + 115 + if stream-based fan-out shows unacceptable operational behavior, fallback is pub/sub + explicit snapshot sync endpoint/message. this keeps product behavior intact while changing transport internals only.

+206

docs/architecture/jams.md

··· 1 + # jams — shared listening rooms 2 + 3 + real-time shared playback rooms. server-side only (no ATProto records in v1). gated behind `jams` feature flag. 4 + 5 + ## overview 6 + 7 + jams let multiple users listen to the same music in sync. one user creates a jam, gets a shareable link, and anyone with the link (and the feature flag) can join. all participants can control playback — it's democratic and chaotic by design. no host-only lock. no chat. 8 + 9 + a jam is "your queue, but shared." the queue panel becomes the jam UI. there's no separate jam page or jam player — the existing player footer handles everything. 10 + 11 + ## playback model 12 + 13 + each client plays audio independently on its own device. the server owns the authoritative state (what track, what position, playing or paused), and clients sync to it. 14 + 15 + **sync is command-driven, not continuous.** the server broadcasts state only when someone does something — play, pause, seek, skip, add a track. between commands, each client free-runs its own audio element. clients may drift apart during uninterrupted playback and that's intentional. the next command re-syncs everyone. 16 + 17 + position interpolation: server stores a snapshot `{progress_ms, server_time_ms, is_playing}` on each state transition. clients compute current position as: 18 + 19 + ``` 20 + if playing: progress_ms + (Date.now() - server_time_ms) 21 + if paused: progress_ms 22 + ``` 23 + 24 + drift correction: if a client's audio element is >2 seconds off from the interpolated server position, it seeks. this only fires when new state arrives via WebSocket, not every frame. 25 + 26 + ## data model 27 + 28 + ### tables 29 + 30 + **`jams`**: room state. `code` is an 8-char alphanumeric for URLs (`plyr.fm/jam/a1b2c3d4`). `state` is a JSONB playback snapshot. `revision` is monotonic — incremented on every mutation. `host_did` is display-only (any participant controls playback). 31 + 32 + **`jam_participants`**: join/leave tracking. `left_at IS NULL` means currently active. partial index `ix_jam_participants_did_active` makes "find user's active jam" fast. 33 + 34 + ### playback state shape (JSONB) 35 + 36 + ```json 37 + { 38 + "track_ids": ["abc", "def"], 39 + "current_index": 0, 40 + "current_track_id": "abc", 41 + "is_playing": true, 42 + "progress_ms": 12500, 43 + "server_time_ms": 1708000000000 44 + } 45 + ``` 46 + 47 + ## transport 48 + 49 + Redis Streams. see `jams-transport-decision.md` for rationale. 50 + 51 + each jam has a stream `jam:{id}:events` (MAXLEN ~1000). backend instances run `XREAD BLOCK` per active jam and fan out to connected WebSockets. no consumer groups — each instance reads independently. 52 + 53 + ## commands 54 + 55 + all 9 commands are server-authoritative. clients send requests via WebSocket, server applies them, increments revision, broadcasts result. 56 + 57 + | command | behavior | 58 + |---------|----------| 59 + | `play` | set `is_playing = true`, update `server_time_ms` | 60 + | `pause` | freeze `progress_ms` at interpolated position, set `is_playing = false` | 61 + | `seek` | set `progress_ms` to requested position | 62 + | `next` | advance `current_index` if not at end | 63 + | `previous` | go back `current_index` if not at start | 64 + | `add_tracks` | append track IDs to queue, auto-init if queue was empty | 65 + | `play_track` | insert track after current, jump to it, auto-play | 66 + | `set_index` | jump to specific track index | 67 + | `remove_track` | remove track, adjust `current_index` if needed | 68 + 69 + ## WebSocket protocol 70 + 71 + client → server: 72 + - `{type: "sync", last_id: string | null}` — initial sync / reconnect 73 + - `{type: "command", payload: {type: "play" | "pause" | "seek" | ..., ...}}` — playback commands 74 + - `{type: "ping"}` — heartbeat 75 + 76 + server → client: 77 + - `{type: "state", stream_id, revision, state, tracks?, tracks_changed?, actor}` — snapshot 78 + - `{type: "participant", event: "joined" | "left", ...}` — presence 79 + - `{type: "pong"}` — heartbeat response 80 + - `{type: "error", message}` — errors 81 + 82 + reconnect: client sends `{type: "sync", last_id: "..."}` with its last stream ID. server replays missed events via `XRANGE`. if stream was trimmed, falls back to full DB snapshot. 83 + 84 + ## API 85 + 86 + | method | route | description | 87 + |--------|-------|-------------| 88 + | POST | `/jams/` | create jam (accepts initial playback state) | 89 + | GET | `/jams/active` | user's current active jam | 90 + | GET | `/jams/{code}` | get jam details | 91 + | POST | `/jams/{code}/join` | join | 92 + | POST | `/jams/{code}/leave` | leave | 93 + | POST | `/jams/{code}/end` | end (host only) | 94 + | POST | `/jams/{code}/command` | playback command (REST fallback) | 95 + | WS | `/jams/{code}/ws` | real-time sync (cookie auth) | 96 + 97 + create accepts `current_index`, `is_playing`, `progress_ms` so starting a jam preserves the host's current playback state. 98 + 99 + ## frontend architecture 100 + 101 + ### bridge pattern 102 + 103 + the queue is the single integration point for all playback mutations. no `if (jam.active)` conditionals in UI components. 104 + 105 + ``` 106 + ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ 107 + │ playback.ts │────>│ queue │────>│ jam (WS) │ 108 + │ controls │ │ .playNow() │ │ .playTrack()│ 109 + │ Queue.svelte│ │ .next() │ │ .next() │ 110 + │ feed clicks │ │ .addTracks()│ │ .addTracks()│ 111 + └──────────────┘ └──────────────┘ └──────────────┘ 112 + callers single gate transport 113 + (unchanged) (jam-aware) (registered) 114 + ``` 115 + 116 + `jam.svelte.ts` registers a `JamBridge` with the queue on create/join, unregisters on leave/destroy. queue methods check the bridge first — if set, route through WebSocket; if not, do normal local mutation. 117 + 118 + **read-path sync**: when jam state arrives via WebSocket, `jam.svelte.ts` pushes `tracks` and `currentIndex` into the queue. this makes queue getters (`hasNext`, `hasPrevious`, `upNext`, `currentTrack`) reflect jam state, so controls and end-of-track logic work correctly without jam-awareness. 119 + 120 + see `jams-queue-integration.md` for design rationale. 121 + 122 + ### bridged queue methods 123 + 124 + these route through the jam when a bridge is set: 125 + 126 + `playNow`, `addTracks`, `goTo`, `next`, `previous`, `removeTrack`, `play`, `pause`, `seek`, `togglePlayPause` 127 + 128 + ### NOT bridged (local-only, broken during jams) 129 + 130 + `moveTrack` (reorder), `toggleShuffle`, `clearUpNext`, `setQueue`, `clear` 131 + 132 + these operate on local queue state only. using them during a jam will desync. 133 + 134 + ### Player.svelte effects 135 + 136 + Player.svelte still imports `jam` directly for incoming state sync. four jam-related effects: 137 + 138 + 1. **track sync** — when `jam.active && jam.currentTrack`, drives player track loading from jam state instead of queue 139 + 2. **pause sync** — watches `jam.isPlaying`, sets `player.paused` accordingly. uses `untrack()` to avoid running every frame 140 + 3. **drift correction** — watches `jam.interpolatedProgressMs`, seeks if audio element is >2s off. uses `untrack()` 141 + 4. **position save skip** — when `jam.active`, skips saving playback position to server (jam owns it) 142 + 143 + ### join flow 144 + 145 + 1. user visits `/jam/[code]` 146 + 2. `onMount` calls `jam.join(code)` — POST to backend, sets up WebSocket, registers bridge 147 + 3. `goto('/')` (SvelteKit navigation, preserves runtime state) 148 + 4. `$effect` in layout detects `jam.active`, auto-opens queue panel 149 + 150 + ### reconnect on page load 151 + 152 + layout `onMount` (after auth init) calls `jam.fetchActive()`. if the user has an active jam, calls `jam.join(code)` to reconnect WebSocket and re-register bridge. 153 + 154 + ### jam UI 155 + 156 + when a jam is active, the queue panel shows: 157 + - jam name and 6-char share code 158 + - connection status dot (green = connected, yellow = reconnecting) 159 + - participant avatars 160 + - copy-link button 161 + - leave button 162 + - rainbow gradient border on the queue panel 163 + 164 + ## edge cases 165 + 166 + - **personal queue**: preserved in Postgres, untouched during jam. restored on leave 167 + - **host leaves**: jam continues. any participant controls playback. jam ends when all leave or host calls `/end` 168 + - **all leave**: jam marked inactive, Redis stream deleted 169 + - **auto-leave**: creating or joining a new jam auto-leaves any previous jam 170 + - **reconnect**: exponential backoff (1s → 30s), sync from last stream ID, drift correction 171 + - **page refresh**: layout detects active jam via `GET /jams/active` and reconnects 172 + - **simultaneous commands**: server-authoritative, serialized via `SELECT ... FOR UPDATE` row locking 173 + - **gated tracks**: each client resolves audio independently. non-supporters get toast + skip 174 + 175 + ## known issues and follow-ups 176 + 177 + ### medium priority 178 + 179 + - **`_auto_leave` cleanup** — leaves previous jams but doesn't close the old WebSocket or unregister the old bridge on the client side. 180 + - **product semantics** — current design is democratic (any participant controls playback). issue #947 describes host-centric control. needs explicit resolution in the issue/PR thread. 181 + 182 + ### nice to have 183 + 184 + - **periodic sync heartbeat** — server could broadcast position every N seconds during playback to reduce drift between commands. not needed unless drift becomes noticeable in practice. 185 + - **queue reorder via drag** — needs a bridge method and backend command to support reordering during jams. 186 + - **shuffle/clear during jams** — currently disabled (no backend commands). could add `shuffle` and `clear_up_next` commands later. 187 + 188 + ## files 189 + 190 + | file | role | 191 + |------|------| 192 + | `backend/src/backend/models/jam.py` | Jam, JamParticipant SQLAlchemy models | 193 + | `backend/src/backend/_internal/jams.py` | JamService — commands, state management, Redis Streams | 194 + | `backend/src/backend/api/jams.py` | REST + WS endpoints, request/response models | 195 + | `backend/tests/api/test_jams.py` | 23 tests covering CRUD, commands, lifecycle, auth, DID socket replacement | 196 + | `frontend/src/lib/jam.svelte.ts` | JamState singleton — WebSocket, bridge registration | 197 + | `frontend/src/lib/queue.svelte.ts` | JamBridge interface, bridge routing in queue methods | 198 + | `frontend/src/lib/components/Queue.svelte` | Jam UI (participants, share, leave, rainbow border) | 199 + | `frontend/src/lib/components/player/Player.svelte` | Jam sync effects (track, pause, drift correction) | 200 + | `frontend/src/lib/playback.svelte.ts` | Jam-aware — blocks `playQueue` during jams with toast | 201 + | `frontend/src/lib/components/player/PlaybackControls.svelte` | No jam awareness — calls queue methods | 202 + | `frontend/src/routes/jam/[code]/+page.svelte` | Join page — calls jam.join(), goto('/') | 203 + | `frontend/src/routes/+layout.svelte` | Startup reconnect via fetchActive(), auto-open queue | 204 + | `docs/architecture/jams-queue-integration.md` | Bridge pattern design rationale | 205 + | `docs/architecture/jams-transport-decision.md` | Redis Streams decision record | 206 + | `backend/alembic/versions/2026_02_19_*_add_jams_tables.py` | Migration |

+1 -1

frontend/src/lib/components/Header.svelte

··· 188 188 189 189 .margin-right { 190 190 position: absolute; 191 - right: var(--queue-width, 0px); 191 + right: 0; 192 192 top: 50%; 193 193 transform: translateY(-50%); 194 194 display: flex;

+279 -39

frontend/src/lib/components/Queue.svelte

··· 1 1 <script lang="ts"> 2 2 import { queue } from '$lib/queue.svelte'; 3 + import { player } from '$lib/player.svelte'; 3 4 import { goToIndex } from '$lib/playback.svelte'; 5 + import { auth } from '$lib/auth.svelte'; 6 + import { jam } from '$lib/jam.svelte'; 7 + import { toast } from '$lib/toast.svelte'; 8 + import { JAMS_FLAG } from '$lib/config'; 4 9 import type { Track } from '$lib/types'; 5 10 6 11 let draggedIndex = $state<number | null>(null); ··· 13 18 let touchDragElement = $state<HTMLElement | null>(null); 14 19 let queueTracksElement = $state<HTMLElement | null>(null); 15 20 16 - const currentTrack = $derived.by<Track | null>(() => queue.tracks[queue.currentIndex] ?? null); 21 + const canJam = $derived(auth.user?.enabled_flags?.includes(JAMS_FLAG) ?? false); 22 + 23 + async function startJam() { 24 + const trackIds = queue.tracks.map((t) => t.file_id); 25 + await jam.create( 26 + undefined, 27 + trackIds, 28 + queue.currentIndex, 29 + !player.paused, 30 + Math.round(player.currentTime * 1000) 31 + ); 32 + } 33 + 34 + async function leaveJam() { 35 + await jam.leave(); 36 + } 37 + 38 + async function shareJam() { 39 + const url = `${window.location.origin}/jam/${jam.code}`; 40 + try { 41 + await navigator.clipboard.writeText(url); 42 + toast.success('link copied'); 43 + } catch { 44 + toast.error('failed to copy link'); 45 + } 46 + } 47 + 48 + // when jam is active, show jam tracks; otherwise show personal queue 49 + const tracks = $derived(jam.active ? jam.tracks : queue.tracks); 50 + const currentIdx = $derived(jam.active ? jam.currentIndex : queue.currentIndex); 51 + const currentTrack = $derived.by<Track | null>(() => tracks[currentIdx] ?? null); 17 52 const upcoming = $derived.by<{ track: Track; index: number }[]>(() => { 18 - return queue.tracks 53 + return tracks 19 54 .map((track, index) => ({ track, index })) 20 - .filter(({ index }) => index > queue.currentIndex); 55 + .filter(({ index }) => index > currentIdx); 21 56 }); 57 + 58 + function handleTrackClick(index: number) { 59 + goToIndex(index); 60 + } 61 + 62 + function handleRemoveTrack(index: number) { 63 + queue.removeTrack(index); 64 + } 22 65 23 66 // desktop drag and drop 24 67 function handleDragStart(event: DragEvent, index: number) { ··· 110 153 } 111 154 </script> 112 155 113 - {#if queue.tracks.length > 0} 114 - <div class="queue"> 156 + {#if tracks.length > 0} 157 + <div class="queue" class:jam-mode={jam.active}> 115 158 <div class="queue-header"> 116 - <h2>queue</h2> 117 - <div class="queue-actions"> 118 - <button 119 - class="shuffle-btn" 120 - class:active={queue.shuffle} 121 - onclick={() => queue.toggleShuffle()} 122 - title={queue.shuffle ? 'disable shuffle' : 'enable shuffle'} 123 - > 124 - <svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"> 125 - <polyline points="16 3 21 3 21 8"></polyline> 126 - <line x1="4" y1="20" x2="21" y2="3"></line> 127 - <polyline points="21 16 21 21 16 21"></polyline> 128 - <line x1="15" y1="15" x2="21" y2="21"></line> 129 - <line x1="4" y1="4" x2="9" y2="9"></line> 130 - </svg> 131 - </button> 132 - {#if upcoming.length > 0} 159 + {#if jam.active} 160 + <div class="jam-info"> 161 + <span class="jam-name">{jam.jam?.name ?? 'jam'}</span> 162 + <div class="jam-meta"> 163 + <span class="connection-dot" class:connected={jam.connected} class:reconnecting={jam.reconnecting}></span> 164 + <span class="jam-code">{jam.code}</span> 165 + </div> 166 + </div> 167 + <div class="queue-actions"> 168 + <button class="share-btn" onclick={shareJam} title="share jam link"> 169 + <svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"> 170 + <circle cx="18" cy="5" r="3"></circle> 171 + <circle cx="6" cy="12" r="3"></circle> 172 + <circle cx="18" cy="19" r="3"></circle> 173 + <line x1="8.59" y1="13.51" x2="15.42" y2="17.49"></line> 174 + <line x1="15.41" y1="6.51" x2="8.59" y2="10.49"></line> 175 + </svg> 176 + </button> 177 + <button class="leave-btn" onclick={leaveJam} title="leave jam"> 178 + <svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"> 179 + <path d="M9 21H5a2 2 0 0 1-2-2V5a2 2 0 0 1 2-2h4"></path> 180 + <polyline points="16 17 21 12 16 7"></polyline> 181 + <line x1="21" y1="12" x2="9" y2="12"></line> 182 + </svg> 183 + </button> 184 + </div> 185 + {:else} 186 + <h2>queue</h2> 187 + <div class="queue-actions"> 188 + {#if canJam} 189 + <button 190 + class="jam-btn" 191 + onclick={startJam} 192 + title="start a jam" 193 + > 194 + <svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"> 195 + <path d="M17 21v-2a4 4 0 0 0-4-4H5a4 4 0 0 0-4 4v2"></path> 196 + <circle cx="9" cy="7" r="4"></circle> 197 + <path d="M23 21v-2a4 4 0 0 0-3-3.87"></path> 198 + <path d="M16 3.13a4 4 0 0 1 0 7.75"></path> 199 + </svg> 200 + </button> 201 + {/if} 133 202 <button 134 - class="clear-btn" 135 - onclick={() => queue.clearUpNext()} 136 - title="clear upcoming tracks" 203 + class="shuffle-btn" 204 + class:active={queue.shuffle} 205 + onclick={() => queue.toggleShuffle()} 206 + title={queue.shuffle ? 'disable shuffle' : 'enable shuffle'} 137 207 > 138 - clear 208 + <svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"> 209 + <polyline points="16 3 21 3 21 8"></polyline> 210 + <line x1="4" y1="20" x2="21" y2="3"></line> 211 + <polyline points="21 16 21 21 16 21"></polyline> 212 + <line x1="15" y1="15" x2="21" y2="21"></line> 213 + <line x1="4" y1="4" x2="9" y2="9"></line> 214 + </svg> 139 215 </button> 140 - {/if} 141 - </div> 216 + {#if upcoming.length > 0} 217 + <button 218 + class="clear-btn" 219 + onclick={() => queue.clearUpNext()} 220 + title="clear upcoming tracks" 221 + > 222 + clear 223 + </button> 224 + {/if} 225 + </div> 226 + {/if} 142 227 </div> 143 228 229 + {#if jam.active && jam.participants.length > 0} 230 + <div class="participants-strip"> 231 + {#each jam.participants as participant (participant.did)} 232 + <div class="participant-chip" title={participant.display_name ?? participant.handle}> 233 + {#if participant.avatar_url} 234 + <img src={participant.avatar_url} alt="" class="participant-avatar" /> 235 + {:else} 236 + <div class="participant-avatar placeholder"> 237 + <svg width="12" height="12" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2"> 238 + <path d="M20 21v-2a4 4 0 0 0-4-4H8a4 4 0 0 0-4 4v2"></path> 239 + <circle cx="12" cy="7" r="4"></circle> 240 + </svg> 241 + </div> 242 + {/if} 243 + </div> 244 + {/each} 245 + </div> 246 + {/if} 247 + 144 248 <div class="queue-body"> 145 249 {#if currentTrack} 146 250 <section class="now-playing"> ··· 166 270 {#if upcoming.length > 0} 167 271 <div 168 272 class="queue-tracks" 273 + role="list" 169 274 bind:this={queueTracksElement} 170 275 ontouchmove={handleTouchMove} 171 276 ontouchend={handleTouchEnd} ··· 174 279 {#each upcoming as { track, index } (`${track.file_id}:${index}`)} 175 280 <div 176 281 class="queue-track" 177 - class:drag-over={dragOverIndex === index && touchDragIndex !== index} 178 - class:is-dragging={touchDragIndex === index || draggedIndex === index} 282 + class:drag-over={!jam.active && dragOverIndex === index && touchDragIndex !== index} 283 + class:is-dragging={!jam.active && (touchDragIndex === index || draggedIndex === index)} 179 284 data-index={index} 180 - draggable="true" 285 + draggable={!jam.active} 181 286 role="button" 182 287 tabindex="0" 183 - ondragstart={(e) => handleDragStart(e, index)} 184 - ondragover={(e) => handleDragOver(e, index)} 185 - ondrop={(e) => handleDrop(e, index)} 186 - ondragend={handleDragEnd} 187 - onclick={() => goToIndex(index)} 188 - onkeydown={(e) => e.key === 'Enter' && goToIndex(index)} 288 + ondragstart={jam.active ? undefined : (e) => handleDragStart(e, index)} 289 + ondragover={jam.active ? undefined : (e) => handleDragOver(e, index)} 290 + ondrop={jam.active ? undefined : (e) => handleDrop(e, index)} 291 + ondragend={jam.active ? undefined : handleDragEnd} 292 + onclick={() => handleTrackClick(index)} 293 + onkeydown={(e) => e.key === 'Enter' && handleTrackClick(index)} 189 294 > 190 -  295 +  296 + {#if !jam.active} 191 297 <button 192 298 class="drag-handle" 193 299 ontouchstart={(e) => handleTouchStart(e, index)} ··· 204 310 <circle cx="11" cy="13" r="1.5"></circle> 205 311 </svg> 206 312 </button> 313 + {/if} 207 314 208 315 <div class="track-info"> 209 316 <div class="track-title">{track.title}</div> ··· 218 325 class="remove-btn" 219 326 onclick={(e) => { 220 327 e.stopPropagation(); 221 - queue.removeTrack(index); 328 + handleRemoveTrack(index); 222 329 }} 223 330 aria-label="remove from queue" 224 331 title="remove from queue" ··· 263 370 gap: 1rem; 264 371 } 265 372 373 + .queue.jam-mode { 374 + border-top: 2px solid transparent; 375 + border-image: linear-gradient(90deg, #ff6b6b, #ffd93d, #6bcb77, #4d96ff, #9b59b6, #ff6b6b) 1; 376 + } 377 + 378 + .jam-info { 379 + display: flex; 380 + flex-direction: column; 381 + gap: 0.2rem; 382 + min-width: 0; 383 + } 384 + 385 + .jam-name { 386 + font-size: var(--text-lg); 387 + font-weight: 600; 388 + color: var(--text-primary); 389 + text-transform: lowercase; 390 + white-space: nowrap; 391 + overflow: hidden; 392 + text-overflow: ellipsis; 393 + } 394 + 395 + .jam-meta { 396 + display: flex; 397 + align-items: center; 398 + gap: 0.375rem; 399 + } 400 + 401 + .connection-dot { 402 + width: 7px; 403 + height: 7px; 404 + border-radius: var(--radius-full); 405 + background: var(--text-tertiary); 406 + flex-shrink: 0; 407 + transition: background 0.3s; 408 + } 409 + 410 + .connection-dot.connected { 411 + background: #22c55e; 412 + } 413 + 414 + .connection-dot.reconnecting { 415 + background: #eab308; 416 + animation: pulse 1.5s ease-in-out infinite; 417 + } 418 + 419 + @keyframes pulse { 420 + 0%, 100% { opacity: 1; } 421 + 50% { opacity: 0.4; } 422 + } 423 + 424 + .jam-code { 425 + font-size: var(--text-xs); 426 + color: var(--text-tertiary); 427 + font-family: monospace; 428 + } 429 + 430 + .share-btn, 431 + .leave-btn { 432 + display: flex; 433 + align-items: center; 434 + justify-content: center; 435 + width: 32px; 436 + height: 32px; 437 + padding: 0; 438 + background: transparent; 439 + border: 1px solid var(--border-subtle); 440 + color: var(--text-tertiary); 441 + border-radius: var(--radius-sm); 442 + cursor: pointer; 443 + transition: all 0.15s ease; 444 + } 445 + 446 + .share-btn:hover { 447 + color: var(--accent); 448 + border-color: var(--accent); 449 + background: color-mix(in srgb, var(--accent) 10%, transparent); 450 + } 451 + 452 + .leave-btn:hover { 453 + color: var(--error); 454 + border-color: var(--error); 455 + background: color-mix(in srgb, var(--error) 10%, transparent); 456 + } 457 + 458 + .participants-strip { 459 + display: flex; 460 + gap: 0.375rem; 461 + padding: 0 0.25rem; 462 + flex-wrap: wrap; 463 + } 464 + 465 + .participant-chip { 466 + flex-shrink: 0; 467 + } 468 + 469 + .participant-avatar { 470 + width: 24px; 471 + height: 24px; 472 + border-radius: var(--radius-full); 473 + object-fit: cover; 474 + border: 1px solid var(--border-subtle); 475 + } 476 + 477 + .participant-avatar.placeholder { 478 + display: flex; 479 + align-items: center; 480 + justify-content: center; 481 + background: var(--bg-tertiary); 482 + color: var(--text-tertiary); 483 + } 484 + 266 485 .queue-header h2 { 267 486 margin: 0; 268 487 font-size: var(--text-lg); ··· 307 526 .shuffle-btn.active { 308 527 color: var(--accent); 309 528 border-color: var(--accent); 529 + } 530 + 531 + .jam-btn { 532 + display: flex; 533 + align-items: center; 534 + justify-content: center; 535 + width: 32px; 536 + height: 32px; 537 + padding: 0; 538 + background: transparent; 539 + border: 1px solid var(--border-subtle); 540 + color: var(--text-tertiary); 541 + border-radius: var(--radius-sm); 542 + cursor: pointer; 543 + transition: all 0.15s ease; 544 + } 545 + 546 + .jam-btn:hover { 547 + color: var(--accent); 548 + border-color: var(--accent); 549 + background: color-mix(in srgb, var(--accent) 10%, transparent); 310 550 } 311 551 312 552 .clear-btn {

+6 -13

frontend/src/lib/components/player/PlaybackControls.svelte

··· 58 58 const RESTART_THRESHOLD = 1; 59 59 60 60 if (player.currentTime > RESTART_THRESHOLD) { 61 - player.currentTime = 0; 61 + queue.seek(0); 62 62 seekValue = 0; 63 - if (player.paused) { 64 - player.paused = false; 65 - } 63 + queue.play(); 66 64 } else if (queue.hasPrevious) { 67 65 queue.previous(); 68 66 } else { 69 - player.currentTime = 0; 67 + queue.seek(0); 70 68 seekValue = 0; 71 - if (player.paused) { 72 - player.paused = false; 73 - } 69 + queue.play(); 74 70 } 75 71 } 76 72 ··· 81 77 } 82 78 83 79 function commitSeek(value: number) { 84 - player.currentTime = value; 85 - if (player.audioElement) { 86 - player.audioElement.currentTime = value; 87 - } 80 + queue.seek(Math.round(value * 1000)); 88 81 seekValue = value; 89 82 } 90 83 ··· 124 117 125 118 <button 126 119 class="control-btn play-pause" 127 - onclick={() => player.togglePlayPause()} 120 + onclick={() => queue.togglePlayPause()} 128 121 title={player.paused ? 'play' : 'pause'} 129 122 > 130 123 {#if !player.paused}

+55 -11

frontend/src/lib/components/player/Player.svelte

··· 1 1 <script lang="ts"> 2 + import { untrack } from 'svelte'; 2 3 import { player } from '$lib/player.svelte'; 3 4 import { queue } from '$lib/queue.svelte'; 5 + import { jam } from '$lib/jam.svelte'; 4 6 import { nowPlaying } from '$lib/now-playing.svelte'; 5 7 import { moderation } from '$lib/moderation.svelte'; 6 8 import { preferences } from '$lib/preferences.svelte'; ··· 53 55 if (!('mediaSession' in navigator)) return; 54 56 55 57 navigator.mediaSession.setActionHandler('play', () => { 56 - player.paused = false; 58 + queue.play(); 57 59 }); 58 60 59 61 navigator.mediaSession.setActionHandler('pause', () => { 60 - player.paused = true; 62 + queue.pause(); 61 63 }); 62 64 63 65 navigator.mediaSession.setActionHandler('previoustrack', () => { ··· 73 75 }); 74 76 75 77 navigator.mediaSession.setActionHandler('seekto', (details) => { 76 - if (details.seekTime !== undefined && player.audioElement) { 77 - player.audioElement.currentTime = details.seekTime; 78 + if (details.seekTime !== undefined) { 79 + queue.seek(details.seekTime * 1000); 78 80 } 79 81 }); 80 82 81 83 navigator.mediaSession.setActionHandler('seekbackward', (details) => { 82 84 if (player.audioElement) { 83 85 const skipTime = details.seekOffset ?? 10; 84 - player.audioElement.currentTime = Math.max(0, player.audioElement.currentTime - skipTime); 86 + const newTime = Math.max(0, player.audioElement.currentTime - skipTime); 87 + queue.seek(newTime * 1000); 85 88 } 86 89 }); 87 90 88 91 navigator.mediaSession.setActionHandler('seekforward', (details) => { 89 92 if (player.audioElement) { 90 93 const skipTime = details.seekOffset ?? 10; 91 - player.audioElement.currentTime = Math.min( 92 - player.duration, 93 - player.audioElement.currentTime + skipTime 94 - ); 94 + const newTime = Math.min(player.duration, player.audioElement.currentTime + skipTime); 95 + queue.seek(newTime * 1000); 95 96 } 96 97 }); 97 98 } ··· 108 109 } 109 110 }); 110 111 111 - // sync playback position to queue for persistence 112 + // sync playback position to queue for persistence (skip in jam mode — server owns position) 112 113 $effect(() => { 113 - queue.progressMs = Math.round(player.currentTime * 1000); 114 + if (!jam.active) { 115 + queue.progressMs = Math.round(player.currentTime * 1000); 116 + } 114 117 }); 115 118 116 119 // track play count when threshold is reached ··· 460 463 let shouldAutoPlay = $state(false); 461 464 462 465 $effect(() => { 466 + // in jam mode, jam state drives the player directly 467 + if (jam.active && jam.currentTrack) { 468 + const trackChanged = jam.currentTrack.id !== player.currentTrack?.id; 469 + if (trackChanged) { 470 + player.currentTrack = jam.currentTrack; 471 + shouldAutoPlay = jam.isPlaying; 472 + } 473 + return; 474 + } 475 + 463 476 if (queue.currentTrack) { 464 477 const trackChanged = queue.currentTrack.id !== player.currentTrack?.id; 465 478 const indexChanged = queue.currentIndex !== previousQueueIndex; ··· 490 503 player.paused = false; 491 504 shouldAutoPlay = false; 492 505 } 506 + }); 507 + 508 + // sync play/pause from jam state (any participant can play/pause) 509 + // only runs when jam play state changes (from WS messages) 510 + $effect(() => { 511 + if (!jam.active) return; 512 + const jamPlaying = jam.isPlaying; 513 + const jamTrackId = jam.currentTrack?.id; 514 + untrack(() => { 515 + if (isLoadingTrack) return; 516 + if (!jamTrackId || jamTrackId !== player.currentTrack?.id) return; 517 + player.paused = !jamPlaying; 518 + }); 519 + }); 520 + 521 + // jam drift correction: seek if >2s off from server 522 + // only runs when jam state changes (progressMs/serverTimeMs from WS), not every frame 523 + $effect(() => { 524 + if (!jam.active) return; 525 + // track jam state as dependencies (these change on WS messages) 526 + const serverPos = jam.interpolatedProgressMs / 1000; 527 + const jamTrackId = jam.currentTrack?.id; 528 + // read player state without tracking to avoid running every frame 529 + untrack(() => { 530 + if (!player.audioElement || !player.duration) return; 531 + if (!jamTrackId || jamTrackId !== player.currentTrack?.id) return; 532 + const drift = Math.abs(player.currentTime - serverPos); 533 + if (drift > 2) { 534 + player.audioElement.currentTime = serverPos; 535 + } 536 + }); 493 537 }); 494 538 495 539 function handleTrackEnded() {

+1

frontend/src/lib/config.ts

··· 4 4 5 5 export const PDS_AUDIO_UPLOADS_FLAG = 'pds-audio-uploads'; 6 6 export const VIBE_SEARCH_FLAG = 'vibe-search'; 7 + export const JAMS_FLAG = 'jams'; 7 8 8 9 /** 9 10 * generate atprotofans support URL for an artist.

+404

frontend/src/lib/jam.svelte.ts

··· 1 + import { browser } from '$app/environment'; 2 + import { API_URL } from './config'; 3 + import { queue, type JamBridge } from './queue.svelte'; 4 + import type { JamInfo, JamParticipant, JamPlaybackState, Track } from './types'; 5 + 6 + const RECONNECT_BASE_MS = 1000; 7 + const RECONNECT_MAX_MS = 30000; 8 + 9 + class JamState { 10 + active = $state(false); 11 + jam = $state<JamInfo | null>(null); 12 + participants = $state<JamParticipant[]>([]); 13 + tracks = $state<Track[]>([]); 14 + currentIndex = $state(0); 15 + isPlaying = $state(false); 16 + progressMs = $state(0); 17 + serverTimeMs = $state(0); 18 + revision = $state(0); 19 + connected = $state(false); 20 + reconnecting = $state(false); 21 + 22 + private ws: WebSocket | null = null; 23 + private lastStreamId: string | null = null; 24 + private reconnectTimer: number | null = null; 25 + private reconnectDelay = RECONNECT_BASE_MS; 26 + private visibilityHandler: (() => void) | null = null; 27 + private currentCode: string | null = null; 28 + 29 + get currentTrack(): Track | null { 30 + if (this.tracks.length === 0) return null; 31 + return this.tracks[this.currentIndex] ?? null; 32 + } 33 + 34 + get interpolatedProgressMs(): number { 35 + if (!this.isPlaying) return this.progressMs; 36 + return this.progressMs + (Date.now() - this.serverTimeMs); 37 + } 38 + 39 + get code(): string | null { 40 + return this.jam?.code ?? null; 41 + } 42 + 43 + private createBridge(): JamBridge { 44 + return { 45 + playTrack: (fileId) => this.sendCommand({ type: 'play_track', file_id: fileId }), 46 + addTracks: (fileIds) => this.sendCommand({ type: 'add_tracks', track_ids: fileIds }), 47 + removeTrack: (index) => this.sendCommand({ type: 'remove_track', index }), 48 + setIndex: (index) => this.sendCommand({ type: 'set_index', index }), 49 + next: () => this.sendCommand({ type: 'next' }), 50 + previous: () => this.sendCommand({ type: 'previous' }), 51 + play: () => this.sendCommand({ type: 'play' }), 52 + pause: () => this.sendCommand({ type: 'pause' }), 53 + seek: (ms) => this.sendCommand({ type: 'seek', position_ms: ms }), 54 + }; 55 + } 56 + 57 + // ── lifecycle ───────────────────────────────────────────────── 58 + 59 + async create( 60 + name?: string, 61 + trackIds?: string[], 62 + currentIndex?: number, 63 + isPlaying?: boolean, 64 + progressMs?: number 65 + ): Promise<string | null> { 66 + if (!browser) return null; 67 + 68 + try { 69 + const response = await fetch(`${API_URL}/jams/`, { 70 + method: 'POST', 71 + credentials: 'include', 72 + headers: { 'Content-Type': 'application/json' }, 73 + body: JSON.stringify({ 74 + name: name ?? null, 75 + track_ids: trackIds ?? [], 76 + current_index: currentIndex ?? 0, 77 + is_playing: isPlaying ?? false, 78 + progress_ms: progressMs ?? 0 79 + }) 80 + }); 81 + 82 + if (!response.ok) return null; 83 + 84 + const data: JamInfo = await response.json(); 85 + this.applyJamData(data); 86 + this.connect(data.code); 87 + queue.stopPositionSave(); 88 + queue.setJamBridge(this.createBridge()); 89 + return data.code; 90 + } catch (error) { 91 + console.error('failed to create jam:', error); 92 + return null; 93 + } 94 + } 95 + 96 + async join(code: string): Promise<boolean> { 97 + if (!browser) return false; 98 + 99 + try { 100 + const response = await fetch(`${API_URL}/jams/${code}/join`, { 101 + method: 'POST', 102 + credentials: 'include' 103 + }); 104 + 105 + if (!response.ok) return false; 106 + 107 + const data: JamInfo = await response.json(); 108 + this.applyJamData(data); 109 + this.connect(code); 110 + queue.stopPositionSave(); 111 + queue.setJamBridge(this.createBridge()); 112 + return true; 113 + } catch (error) { 114 + console.error('failed to join jam:', error); 115 + return false; 116 + } 117 + } 118 + 119 + async leave(): Promise<void> { 120 + if (!browser || !this.jam) return; 121 + 122 + const code = this.jam.code; 123 + try { 124 + await fetch(`${API_URL}/jams/${code}/leave`, { 125 + method: 'POST', 126 + credentials: 'include' 127 + }); 128 + } catch (error) { 129 + console.error('failed to leave jam:', error); 130 + } 131 + 132 + this.closeWs(); 133 + this.reset(); 134 + queue.setJamBridge(null); 135 + queue.startPositionSave(); 136 + queue.fetchQueue(); 137 + } 138 + 139 + async fetchJam(code: string): Promise<JamInfo | null> { 140 + if (!browser) return null; 141 + 142 + try { 143 + const response = await fetch(`${API_URL}/jams/${code}`, { 144 + credentials: 'include' 145 + }); 146 + if (!response.ok) return null; 147 + return await response.json(); 148 + } catch (error) { 149 + console.error('failed to fetch jam:', error); 150 + return null; 151 + } 152 + } 153 + 154 + async fetchActive(): Promise<JamInfo | null> { 155 + if (!browser) return null; 156 + 157 + try { 158 + const response = await fetch(`${API_URL}/jams/active`, { 159 + credentials: 'include' 160 + }); 161 + if (!response.ok) return null; 162 + const data = await response.json(); 163 + return data ?? null; 164 + } catch { 165 + return null; 166 + } 167 + } 168 + 169 + // ── commands (via WebSocket) ─────────────────────────────────── 170 + 171 + play(): void { 172 + this.sendCommand({ type: 'play' }); 173 + } 174 + 175 + pause(): void { 176 + this.sendCommand({ type: 'pause' }); 177 + } 178 + 179 + seek(ms: number): void { 180 + this.sendCommand({ type: 'seek', position_ms: ms }); 181 + } 182 + 183 + next(): void { 184 + this.sendCommand({ type: 'next' }); 185 + } 186 + 187 + previous(): void { 188 + this.sendCommand({ type: 'previous' }); 189 + } 190 + 191 + addTracks(ids: string[]): void { 192 + this.sendCommand({ type: 'add_tracks', track_ids: ids }); 193 + } 194 + 195 + removeTrack(index: number): void { 196 + this.sendCommand({ type: 'remove_track', index }); 197 + } 198 + 199 + playTrack(fileId: string): void { 200 + this.sendCommand({ type: 'play_track', file_id: fileId }); 201 + } 202 + 203 + setIndex(index: number): void { 204 + this.sendCommand({ type: 'set_index', index }); 205 + } 206 + 207 + private sendCommand(payload: Record<string, unknown>): void { 208 + if (!this.ws || this.ws.readyState !== WebSocket.OPEN) return; 209 + this.ws.send(JSON.stringify({ type: 'command', payload })); 210 + } 211 + 212 + // ── WebSocket lifecycle ──────────────────────────────────────── 213 + 214 + private connect(code: string): void { 215 + this.closeWs(); 216 + this.currentCode = code; 217 + 218 + const wsProtocol = API_URL.startsWith('https') ? 'wss' : 'ws'; 219 + const wsBase = API_URL.replace(/^https?/, wsProtocol); 220 + const url = `${wsBase}/jams/${code}/ws`; 221 + 222 + this.ws = new WebSocket(url); 223 + this.reconnecting = false; 224 + 225 + // reconnect when app resumes from background 226 + this.visibilityHandler = () => { 227 + if (document.visibilityState === 'visible' && this.active && !this.connected) { 228 + this.connect(code); 229 + } 230 + }; 231 + document.addEventListener('visibilitychange', this.visibilityHandler); 232 + 233 + this.ws.onopen = () => { 234 + this.connected = true; 235 + this.reconnectDelay = RECONNECT_BASE_MS; 236 + // request sync 237 + this.ws?.send( 238 + JSON.stringify({ type: 'sync', last_id: this.lastStreamId }) 239 + ); 240 + }; 241 + 242 + this.ws.onmessage = (event) => { 243 + try { 244 + const data = JSON.parse(event.data); 245 + this.handleMessage(data); 246 + } catch (error) { 247 + console.error('failed to parse jam ws message:', error); 248 + } 249 + }; 250 + 251 + this.ws.onclose = (event) => { 252 + this.connected = false; 253 + // terminal codes: server rejected us, don't retry 254 + if (event.code === 4003 || event.code === 4010) { 255 + this.closeWs(); 256 + this.reset(); 257 + queue.setJamBridge(null); 258 + queue.startPositionSave(); 259 + queue.fetchQueue(); 260 + return; 261 + } 262 + if (this.active) { 263 + this.scheduleReconnect(code); 264 + } 265 + }; 266 + 267 + this.ws.onerror = () => { 268 + // onclose will fire after onerror 269 + }; 270 + } 271 + 272 + private scheduleReconnect(code: string): void { 273 + if (this.reconnectTimer !== null) return; 274 + this.reconnecting = true; 275 + 276 + this.reconnectTimer = window.setTimeout(() => { 277 + this.reconnectTimer = null; 278 + if (this.active) { 279 + this.connect(code); 280 + } 281 + }, this.reconnectDelay); 282 + 283 + this.reconnectDelay = Math.min(this.reconnectDelay * 2, RECONNECT_MAX_MS); 284 + } 285 + 286 + private closeWs(): void { 287 + if (this.visibilityHandler) { 288 + document.removeEventListener('visibilitychange', this.visibilityHandler); 289 + this.visibilityHandler = null; 290 + } 291 + if (this.reconnectTimer !== null) { 292 + window.clearTimeout(this.reconnectTimer); 293 + this.reconnectTimer = null; 294 + } 295 + if (this.ws) { 296 + this.ws.onclose = null; 297 + this.ws.onerror = null; 298 + this.ws.onmessage = null; 299 + this.ws.onopen = null; 300 + this.ws.close(); 301 + this.ws = null; 302 + } 303 + this.connected = false; 304 + this.reconnecting = false; 305 + } 306 + 307 + // ── message handling ─────────────────────────────────────────── 308 + 309 + private handleMessage(data: Record<string, unknown>): void { 310 + const msgType = data.type as string; 311 + 312 + if (msgType === 'state') { 313 + this.handleStateMessage(data); 314 + } else if (msgType === 'participant') { 315 + this.handleParticipantMessage(data); 316 + } else if (msgType === 'pong') { 317 + // heartbeat response, no-op 318 + } else if (msgType === 'error') { 319 + console.error('jam error:', data.message); 320 + } 321 + } 322 + 323 + private handleStateMessage(data: Record<string, unknown>): void { 324 + const state = data.state as JamPlaybackState; 325 + const rev = data.revision as number; 326 + const streamId = data.stream_id as string | null; 327 + 328 + if (rev < this.revision) return; 329 + 330 + this.revision = rev; 331 + if (streamId) this.lastStreamId = streamId; 332 + 333 + this.currentIndex = state.current_index; 334 + this.isPlaying = state.is_playing; 335 + this.progressMs = state.progress_ms; 336 + this.serverTimeMs = state.server_time_ms; 337 + 338 + if (data.tracks_changed && Array.isArray(data.tracks)) { 339 + this.tracks = data.tracks as Track[]; 340 + } 341 + 342 + if (Array.isArray(data.participants)) { 343 + this.participants = data.participants as JamParticipant[]; 344 + } 345 + 346 + this.syncToQueue(); 347 + } 348 + 349 + private async handleParticipantMessage(_data: Record<string, unknown>): Promise<void> { 350 + // participant events only carry DID — fetch full list with metadata 351 + if (!this.jam) return; 352 + const fresh = await this.fetchJam(this.jam.code); 353 + if (fresh) { 354 + this.participants = fresh.participants ?? []; 355 + } 356 + } 357 + 358 + // ── state management ─────────────────────────────────────────── 359 + 360 + private applyJamData(data: JamInfo): void { 361 + this.jam = data; 362 + this.active = true; 363 + this.tracks = data.tracks ?? []; 364 + this.participants = data.participants ?? []; 365 + this.revision = data.revision; 366 + 367 + const state = data.state; 368 + this.currentIndex = state.current_index; 369 + this.isPlaying = state.is_playing; 370 + this.progressMs = state.progress_ms; 371 + this.serverTimeMs = state.server_time_ms; 372 + 373 + this.syncToQueue(); 374 + } 375 + 376 + /** push jam tracks/index into queue so read-path getters (hasNext, hasPrevious, etc.) work */ 377 + private syncToQueue(): void { 378 + queue.tracks = this.tracks; 379 + queue.currentIndex = this.currentIndex; 380 + } 381 + 382 + private reset(): void { 383 + this.active = false; 384 + this.jam = null; 385 + this.participants = []; 386 + this.tracks = []; 387 + this.currentIndex = 0; 388 + this.isPlaying = false; 389 + this.progressMs = 0; 390 + this.serverTimeMs = 0; 391 + this.revision = 0; 392 + this.lastStreamId = null; 393 + } 394 + 395 + destroy(): void { 396 + this.closeWs(); 397 + this.reset(); 398 + queue.setJamBridge(null); 399 + queue.startPositionSave(); 400 + queue.fetchQueue(); 401 + } 402 + } 403 + 404 + export const jam = new JamState();

+6

frontend/src/lib/playback.svelte.ts

··· 7 7 8 8 import { browser } from '$app/environment'; 9 9 import { queue } from './queue.svelte'; 10 + import { jam } from './jam.svelte'; 10 11 import { toast } from './toast.svelte'; 11 12 import { API_URL, getAtprotofansSupportUrl } from './config'; 12 13 import type { Track } from './types'; ··· 133 134 */ 134 135 export async function playQueue(tracks: Track[], startIndex = 0): Promise<boolean> { 135 136 if (!browser || tracks.length === 0) return false; 137 + 138 + if (jam.active) { 139 + toast.info('leave the jam to play a different collection'); 140 + return false; 141 + } 136 142 137 143 const startTrack = tracks[startIndex]; 138 144 if (!startTrack) return false;

+93

frontend/src/lib/queue.svelte.ts

··· 3 3 import { API_URL } from './config'; 4 4 import { APP_BROADCAST_PREFIX } from './branding'; 5 5 import { auth } from './auth.svelte'; 6 + import { player } from './player.svelte'; 6 7 7 8 const SYNC_DEBOUNCE_MS = 250; 8 9 10 + /** bridge for routing queue mutations through a jam's WebSocket transport */ 11 + export interface JamBridge { 12 + playTrack(fileId: string): void; 13 + addTracks(fileIds: string[]): void; 14 + removeTrack(index: number): void; 15 + setIndex(index: number): void; 16 + next(): void; 17 + previous(): void; 18 + play(): void; 19 + pause(): void; 20 + seek(ms: number): void; 21 + } 22 + 9 23 // global queue state using Svelte 5 runes 10 24 class Queue { 25 + jamBridge = $state<JamBridge | null>(null); 26 + 27 + setJamBridge(bridge: JamBridge | null): void { 28 + this.jamBridge = bridge; 29 + } 11 30 tracks = $state<Track[]>([]); 12 31 currentIndex = $state(0); 13 32 shuffle = $state(false); ··· 406 425 } 407 426 } 408 427 428 + // ── playback control (routed through bridge when jam active) ── 429 + 430 + play(): void { 431 + if (this.jamBridge) { 432 + this.jamBridge.play(); 433 + } else { 434 + player.paused = false; 435 + } 436 + } 437 + 438 + pause(): void { 439 + if (this.jamBridge) { 440 + this.jamBridge.pause(); 441 + } else { 442 + player.paused = true; 443 + } 444 + } 445 + 446 + togglePlayPause(): void { 447 + if (player.paused) this.play(); 448 + else this.pause(); 449 + } 450 + 451 + seek(ms: number): void { 452 + if (this.jamBridge) { 453 + this.jamBridge.seek(ms); 454 + } else if (player.audioElement) { 455 + player.audioElement.currentTime = ms / 1000; 456 + } 457 + } 458 + 459 + // ── track mutations (routed through bridge when jam active) ── 460 + 409 461 addTracks(tracks: Track[], playNow = false) { 410 462 if (tracks.length === 0) return; 411 463 464 + if (this.jamBridge) { 465 + const fileIds = tracks.map((t) => t.file_id); 466 + if (playNow && fileIds.length > 0) { 467 + this.jamBridge.playTrack(fileIds[0]); 468 + if (fileIds.length > 1) this.jamBridge.addTracks(fileIds.slice(1)); 469 + } else { 470 + this.jamBridge.addTracks(fileIds); 471 + } 472 + return; 473 + } 474 + 412 475 this.lastUpdateWasLocal = true; 413 476 this.tracks = [...this.tracks, ...tracks]; 414 477 this.originalOrder = [...this.originalOrder, ...tracks]; ··· 421 484 } 422 485 423 486 setQueue(tracks: Track[], startIndex = 0) { 487 + if (this.jamBridge) return; // no set_queue command — block during jams 424 488 if (tracks.length === 0) { 425 489 this.clear(); 426 490 return; ··· 434 498 } 435 499 436 500 playNow(track: Track, autoPlay = true) { 501 + if (this.jamBridge) { 502 + this.jamBridge.playTrack(track.file_id); 503 + return; 504 + } 505 + 437 506 this.lastUpdateWasLocal = autoPlay; 438 507 const upNext = this.tracks.slice(this.currentIndex + 1); 439 508 this.tracks = [track, ...upNext]; ··· 452 521 453 522 goTo(index: number) { 454 523 if (index < 0 || index >= this.tracks.length) return; 524 + 525 + if (this.jamBridge) { 526 + this.jamBridge.setIndex(index); 527 + return; 528 + } 529 + 455 530 this.lastUpdateWasLocal = true; 456 531 this.currentIndex = index; 457 532 this.schedulePush(); ··· 460 535 next() { 461 536 if (this.tracks.length === 0) return; 462 537 538 + if (this.jamBridge) { 539 + this.jamBridge.next(); 540 + return; 541 + } 542 + 463 543 if (this.currentIndex < this.tracks.length - 1) { 464 544 this.lastUpdateWasLocal = true; 465 545 this.currentIndex += 1; ··· 470 550 previous(forceSkip = false) { 471 551 if (this.tracks.length === 0) return; 472 552 553 + if (this.jamBridge) { 554 + this.jamBridge.previous(); 555 + return true; 556 + } 557 + 473 558 if (this.currentIndex > 0 || forceSkip) { 474 559 this.lastUpdateWasLocal = true; 475 560 if (this.currentIndex > 0) { ··· 482 567 } 483 568 484 569 toggleShuffle() { 570 + if (this.jamBridge) return; // no shuffle command — block during jams 485 571 // shuffle is an action, not a mode - shuffle upcoming tracks every time 486 572 if (this.tracks.length <= 1) { 487 573 return; ··· 526 612 } 527 613 528 614 moveTrack(fromIndex: number, toIndex: number) { 615 + if (this.jamBridge) return; // no move_track command — block during jams 529 616 if (fromIndex === toIndex) return; 530 617 if (fromIndex < 0 || fromIndex >= this.tracks.length) return; 531 618 if (toIndex < 0 || toIndex >= this.tracks.length) return; ··· 556 643 if (index < 0 || index >= this.tracks.length) return; 557 644 if (index === this.currentIndex) return; 558 645 646 + if (this.jamBridge) { 647 + this.jamBridge.removeTrack(index); 648 + return; 649 + } 650 + 559 651 this.lastUpdateWasLocal = true; 560 652 const updated = [...this.tracks]; 561 653 const [removed] = updated.splice(index, 1); ··· 579 671 } 580 672 581 673 clearUpNext() { 674 + if (this.jamBridge) return; // no clear command — block during jams 582 675 if (this.tracks.length === 0) return; 583 676 584 677 this.lastUpdateWasLocal = true;

+31

frontend/src/lib/types.ts

··· 141 141 tracks: Track[]; 142 142 } 143 143 144 + export interface JamInfo { 145 + id: string; 146 + code: string; 147 + name: string | null; 148 + host_did: string; 149 + state: JamPlaybackState; 150 + revision: number; 151 + is_active: boolean; 152 + created_at: string; 153 + updated_at: string; 154 + ended_at: string | null; 155 + tracks: Track[]; 156 + participants: JamParticipant[]; 157 + } 158 + 159 + export interface JamParticipant { 160 + did: string; 161 + handle: string; 162 + display_name?: string; 163 + avatar_url?: string; 164 + } 165 + 166 + export interface JamPlaybackState { 167 + track_ids: string[]; 168 + current_index: number; 169 + current_track_id: string | null; 170 + is_playing: boolean; 171 + progress_ms: number; 172 + server_time_ms: number; 173 + } 174 +

+28 -15

frontend/src/routes/+layout.svelte

··· 13 13 import FeedbackModal from '$lib/components/FeedbackModal.svelte'; 14 14 import TermsOverlay from '$lib/components/TermsOverlay.svelte'; 15 15 import LikersSheet from '$lib/components/LikersSheet.svelte'; 16 - import { onMount, onDestroy } from 'svelte'; 16 + import { onMount, onDestroy, untrack } from 'svelte'; 17 17 import { page } from '$app/stores'; 18 18 import { afterNavigate } from '$app/navigation'; 19 19 import { auth } from '$lib/auth.svelte'; ··· 21 21 import { moderation } from '$lib/moderation.svelte'; 22 22 import { player } from '$lib/player.svelte'; 23 23 import { queue } from '$lib/queue.svelte'; 24 + import { jam } from '$lib/jam.svelte'; 25 + import { JAMS_FLAG } from '$lib/config'; 24 26 import { search } from '$lib/search.svelte'; 25 27 import { browser } from '$app/environment'; 26 28 let { children } = $props<{ children: any }>(); ··· 65 67 if (queue.revision === null) { 66 68 void queue.fetchQueue(); 67 69 } 70 + 71 + // reconnect to active jam on page load/refresh 72 + if (!jam.active && auth.user?.enabled_flags?.includes(JAMS_FLAG)) { 73 + const activeJam = await jam.fetchActive(); 74 + if (activeJam) { 75 + await jam.join(activeJam.code); 76 + } 77 + } 78 + } 79 + }); 80 + 81 + // auto-open queue when joining a jam (only on jam.active transition, not on queue toggle) 82 + $effect(() => { 83 + if (!browser) return; 84 + if (jam.active) { 85 + untrack(() => { 86 + if (!showQueue) { 87 + showQueue = true; 88 + localStorage.setItem('showQueue', 'true'); 89 + } 90 + }); 68 91 } 69 92 }); 70 93 ··· 203 226 switch (event.key) { 204 227 case ' ': // space - play/pause 205 228 event.preventDefault(); 206 - player.togglePlayPause(); 229 + queue.togglePlayPause(); 207 230 break; 208 231 209 232 case 'ArrowLeft': // seek backward ··· 242 265 if (!player.audioElement || !player.duration) return; 243 266 244 267 const newTime = Math.max(0, Math.min(player.duration, player.currentTime + seconds)); 245 - player.currentTime = newTime; 246 - player.audioElement.currentTime = newTime; 268 + queue.seek(newTime * 1000); 247 269 } 248 270 249 271 function handlePreviousTrack() { 250 272 const RESTART_THRESHOLD = 3; // restart if more than 3 seconds in 251 273 252 274 if (player.currentTime > RESTART_THRESHOLD) { 253 - // restart current track 254 - player.currentTime = 0; 255 - if (player.audioElement) { 256 - player.audioElement.currentTime = 0; 257 - } 275 + queue.seek(0); 258 276 } else if (queue.hasPrevious) { 259 - // go to previous track 260 277 queue.previous(); 261 278 } else { 262 - // restart from beginning 263 - player.currentTime = 0; 264 - if (player.audioElement) { 265 - player.audioElement.currentTime = 0; 266 - } 279 + queue.seek(0); 267 280 } 268 281 } 269 282

+70

frontend/src/routes/jam/[code]/+page.svelte

··· 1 + <script lang="ts"> 2 + import { onMount } from 'svelte'; 3 + import { goto } from '$app/navigation'; 4 + import { jam } from '$lib/jam.svelte'; 5 + import { toast } from '$lib/toast.svelte'; 6 + import { APP_NAME } from '$lib/branding'; 7 + import type { PageData } from './+page'; 8 + 9 + let { data }: { data: PageData } = $props(); 10 + 11 + let error = $state<string | null>(null); 12 + 13 + onMount(async () => { 14 + const ok = await jam.join(data.code); 15 + if (ok) { 16 + // SvelteKit navigation preserves runtime — jam state survives 17 + goto('/'); 18 + } else { 19 + error = 'could not join jam — it may have ended or the code is invalid'; 20 + toast.error('failed to join jam'); 21 + } 22 + }); 23 + </script> 24 + 25 + <svelte:head> 26 + <title>joining jam - {APP_NAME}</title> 27 + </svelte:head> 28 + 29 + <div class="join-page"> 30 + {#if error} 31 + <div class="error-state"> 32 + <p>{error}</p> 33 + <a href="/">go home</a> 34 + </div> 35 + {:else} 36 + <p class="joining">joining jam...</p> 37 + {/if} 38 + </div> 39 + 40 + <style> 41 + .join-page { 42 + display: flex; 43 + align-items: center; 44 + justify-content: center; 45 + min-height: 60vh; 46 + padding: 2rem; 47 + } 48 + 49 + .joining { 50 + color: var(--text-tertiary); 51 + font-size: var(--text-base); 52 + } 53 + 54 + .error-state { 55 + text-align: center; 56 + display: flex; 57 + flex-direction: column; 58 + gap: 1rem; 59 + color: var(--text-secondary); 60 + } 61 + 62 + .error-state a { 63 + color: var(--accent); 64 + text-decoration: none; 65 + } 66 + 67 + .error-state a:hover { 68 + text-decoration: underline; 69 + } 70 + </style>

+9

frontend/src/routes/jam/[code]/+page.ts

··· 1 + import type { LoadEvent } from '@sveltejs/kit'; 2 + 3 + export interface PageData { 4 + code: string; 5 + } 6 + 7 + export function load({ params }: LoadEvent): PageData { 8 + return { code: params.code as string }; 9 + }

+1 -1

frontend/src/routes/playlist/[id]/+page.svelte

··· 919 919 <button 920 920 class="play-button" 921 921 class:is-playing={isPlaylistPlaying} 922 - onclick={() => isPlaylistActive ? player.togglePlayPause() : playNow()} 922 + onclick={() => isPlaylistActive ? queue.togglePlayPause() : playNow()} 923 923 > 924 924 {#if isPlaylistPlaying} 925 925 <svg width="20" height="20" viewBox="0 0 24 24" fill="currentColor">

+4 -6

frontend/src/routes/track/[id]/+page.svelte

··· 160 160 async function handlePlay() { 161 161 if (player.currentTrack?.id === track.id) { 162 162 // this track is already loaded - just toggle play/pause 163 - player.togglePlayPause(); 163 + queue.togglePlayPause(); 164 164 } else { 165 165 // different track or no track - start this one 166 166 // use playTrack for gated content checks ··· 249 249 250 250 async function seekToTimestamp(ms: number) { 251 251 const doSeek = () => { 252 - if (player.audioElement) { 253 - player.audioElement.currentTime = ms / 1000; 254 - } 252 + queue.seek(ms); 255 253 }; 256 254 257 255 // if this track is already loaded, seek immediately ··· 455 453 player.audioElement && 456 454 player.audioElement.readyState >= 1 457 455 ) { 458 - const seekTo = pendingSeekMs / 1000; 456 + const seekMs = pendingSeekMs; 459 457 pendingSeekMs = null; 460 - player.audioElement.currentTime = seekTo; 458 + queue.seek(seekMs); 461 459 // don't auto-play - browser policy blocks it without user interaction 462 460 // user will click play themselves 463 461 }

+1 -1

frontend/src/routes/u/[handle]/album/[slug]/+page.svelte

··· 557 557 <button 558 558 class="play-button" 559 559 class:is-playing={isAlbumPlaying} 560 - onclick={() => isAlbumActive ? player.togglePlayPause() : playNow()} 560 + onclick={() => isAlbumActive ? queue.togglePlayPause() : playNow()} 561 561 > 562 562 {#if isAlbumPlaying} 563 563 <svg width="20" height="20" viewBox="0 0 24 24" fill="currentColor">

+12 -4

loq.toml

··· 22 22 max_lines = 612 23 23 24 24 [[rules]] 25 + path = "backend/src/backend/_internal/jams.py" 26 + max_lines = 730 27 + 28 + [[rules]] 25 29 path = "backend/src/backend/api/albums.py" 26 30 max_lines = 784 27 31 ··· 64 68 [[rules]] 65 69 path = "backend/tests/api/test_list_record_sync.py" 66 70 max_lines = 554 71 + 72 + [[rules]] 73 + path = "backend/tests/api/test_jams.py" 74 + max_lines = 590 67 75 68 76 [[rules]] 69 77 path = "backend/tests/api/test_track_comments.py" ··· 103 111 104 112 [[rules]] 105 113 path = "frontend/src/lib/components/Queue.svelte" 106 - max_lines = 599 114 + max_lines = 870 107 115 108 116 [[rules]] 109 117 path = "frontend/src/lib/components/SearchModal.svelte" ··· 123 131 124 132 [[rules]] 125 133 path = "frontend/src/lib/components/player/Player.svelte" 126 - max_lines = 583 134 + max_lines = 640 127 135 128 136 [[rules]] 129 137 path = "frontend/src/lib/queue.svelte.ts" 130 - max_lines = 627 138 + max_lines = 725 131 139 132 140 [[rules]] 133 141 path = "frontend/src/routes/+layout.svelte" 134 - max_lines = 680 142 + max_lines = 710 135 143 136 144 [[rules]] 137 145 path = "frontend/src/routes/costs/+page.svelte"