diff --git a/backend/src/db/migrations.ts b/backend/src/db/migrations.ts index 114cf77..55d92cb 100644 --- a/backend/src/db/migrations.ts +++ b/backend/src/db/migrations.ts @@ -876,6 +876,22 @@ export const MIGRATIONS: Migration[] = [ DO NOTHING; `, }, + { + // Vibe control has to follow the audio. The running session's id rides with + // the playback snapshot, so the device taking the audio can adopt the + // session instead of walking whatever queue it happened to receive — which + // is what used to happen, silently, every time a listener moved the audio to + // their phone mid-Vibe. + // + // No foreign key on purpose: a report from the owning device carries the + // whole session, and losing that to a session row that has since been + // deleted would cost far more than a dangling id. + id: '20260810_playback_state_vibe_session', + sql: ` + ALTER TABLE playback_state + ADD COLUMN IF NOT EXISTS vibe_session_id UUID; + `, + }, { id: '20260810_rehide_resurrected_dislikes', sql: ` diff --git a/backend/src/routes/playback.routes.ts b/backend/src/routes/playback.routes.ts index 1d03cfe..bc9bb13 100644 --- a/backend/src/routes/playback.routes.ts +++ b/backend/src/routes/playback.routes.ts @@ -33,6 +33,12 @@ function parsePatch(body: unknown): PlaybackStatePatch | { error: string } { } patch.trackId = input.trackId as string | null; } + if (input.vibeSessionId !== undefined) { + if (input.vibeSessionId !== null && !(typeof input.vibeSessionId === 'string' && UUID_RE.test(input.vibeSessionId))) { + return { error: 'vibeSessionId must be a UUID or null' }; + } + patch.vibeSessionId = input.vibeSessionId as string | null; + } if (input.queue !== undefined) { if (!Array.isArray(input.queue)) return { error: 'queue must be an array' }; patch.queue = input.queue; @@ -169,6 +175,12 @@ export default async function playbackRoutes( const deviceId = typeof query.deviceId === 'string' && UUID_RE.test(query.deviceId) ? query.deviceId : null; if (!deviceId) return reply.code(400).send({ error: 'deviceId must be a UUID' }); + // The response is written straight to the socket and lives for as long as + // the tab does. Without this Fastify still believes it owes a reply and + // waits on a handler that resolves with nothing. + reply.hijack(); + const releaseStream = playbackSync.claimStream(userId, deviceId); + reply.raw.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache, no-transform', @@ -186,11 +198,21 @@ export default async function playbackRoutes( write(event); }); - const [state, devices] = await Promise.all([ - playbackSync.getState(userId), - playbackSync.listDevices(userId), - ]); - write({ type: 'state', state, devices }); + // The reply is hijacked, so a throw from here reaches no error handler that + // could answer it. Close the stream instead and let the client reopen. + try { + const [state, devices] = await Promise.all([ + playbackSync.getState(userId), + playbackSync.listDevices(userId), + ]); + write({ type: 'state', state, devices }); + } catch (err) { + request.log.error(err); + unsubscribe(); + releaseStream(); + reply.raw.end(); + return; + } const heartbeat = setInterval(() => { reply.raw.write(': ping\n\n'); @@ -200,6 +222,7 @@ export default async function playbackRoutes( request.raw.on('close', () => { clearInterval(heartbeat); unsubscribe(); + releaseStream(); // Ownership deliberately survives a closed stream. A phone changing cell, // locking its screen or dozing drops this connection for a few seconds // while its audio keeps playing; releasing here published an unowned diff --git a/backend/src/services/playback-sync.service.test.ts b/backend/src/services/playback-sync.service.test.ts index 69b0852..0d64c46 100644 --- a/backend/src/services/playback-sync.service.test.ts +++ b/backend/src/services/playback-sync.service.test.ts @@ -9,6 +9,7 @@ import { const USER_ID = 'aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa'; const DESKTOP = '11111111-1111-4111-8111-111111111111'; const PHONE = '22222222-2222-4222-8222-222222222222'; +const SESSION = '33333333-3333-4333-8333-333333333333'; /** * A pool stubbed down to the one row this service reasons about. The queries @@ -18,12 +19,14 @@ const PHONE = '22222222-2222-4222-8222-222222222222'; function poolWith(state: { deviceId?: string | null; trackId?: string | null; + vibeSessionId?: string | null; isPlaying?: boolean; positionMs?: number; }) { const row = { device_id: state.deviceId ?? null, track_id: state.trackId ?? null, + vibe_session_id: state.vibeSessionId ?? null, queue: [], queue_index: -1, position_ms: state.positionMs ?? 0, @@ -40,6 +43,12 @@ function poolWith(state: { if (sql.includes('FROM playback_devices') && sql.includes('ORDER BY')) { return { rows: [{ id: DESKTOP, name: 'Linux · Firefox', last_seen_at: row.updated_at, online: true }], rowCount: 1 }; } + if (sql.startsWith('UPDATE playback_devices SET name')) { + return { rows: [{ id: params?.[0], name: params?.[2], last_seen_at: row.updated_at }], rowCount: 1 }; + } + if (sql.startsWith('INSERT INTO playback_devices')) { + return { rows: [{ id: PHONE, name: params?.[1], last_seen_at: row.updated_at }], rowCount: 1 }; + } if (sql.startsWith('SELECT device_id FROM playback_state')) { return { rows: [{ device_id: row.device_id }], rowCount: 1 }; } @@ -76,11 +85,40 @@ describe('cross-device playback', () => { expect(state.deviceId).toBe(PHONE); }); + it('publishes the Vibe session the owning device is driving', async () => { + // Vibe control follows the audio, and this is how the next owner hears about + // the session it is taking over. + const { pool } = poolWith({ deviceId: DESKTOP, vibeSessionId: SESSION }); + const service = new PlaybackSyncService(pool); + + expect((await service.getState(USER_ID)).vibeSessionId).toBe(SESSION); + }); + + it('keeps a device its stored id across a reload', async () => { + const { pool } = poolWith({ deviceId: DESKTOP }); + const service = new PlaybackSyncService(pool); + + const device = await service.registerDevice(USER_ID, 'Linux · Firefox', DESKTOP); + expect(device.id).toBe(DESKTOP); + }); + + it('gives a second tab a device of its own', async () => { + // Both tabs of a browser ask with the same stored id, and one id shared by + // two pages is one device that runs every command twice. + const { pool } = poolWith({ deviceId: DESKTOP }); + const service = new PlaybackSyncService(pool); + service.claimStream(USER_ID, DESKTOP); + + const device = await service.registerDevice(USER_ID, 'Linux · Firefox', DESKTOP); + expect(device.id).toBe(PHONE); + }); + it('delivers a command to the owning device only', async () => { const { pool } = poolWith({ deviceId: DESKTOP }); const service = new PlaybackSyncService(pool); const seen: PlaybackEvent[] = []; service.subscribe(USER_ID, (event) => seen.push(event)); + service.claimStream(USER_ID, DESKTOP); const result = await service.sendCommand(USER_ID, { type: 'pause' }); @@ -88,6 +126,30 @@ describe('cross-device playback', () => { expect(seen).toContainEqual({ type: 'command', deviceId: DESKTOP, command: { type: 'pause' } }); }); + it('reports no delivery when the owning device has no stream to receive on', async () => { + // Ownership outlives a closed stream, so a killed tab still holds the + // session. Answering 202 there told the presser a lie. + const { pool } = poolWith({ deviceId: DESKTOP }); + const service = new PlaybackSyncService(pool); + + expect(await service.sendCommand(USER_ID, { type: 'pause' })).toEqual({ deliveredTo: null }); + }); + + it('frees a device id once its stream closes, and not before', async () => { + const { pool } = poolWith({ deviceId: DESKTOP }); + const service = new PlaybackSyncService(pool); + + const release = service.claimStream(USER_ID, DESKTOP); + const alsoOpen = service.claimStream(USER_ID, DESKTOP); + release(); + expect(service.hasLiveStream(USER_ID, DESKTOP)).toBe(true); + alsoOpen(); + expect(service.hasLiveStream(USER_ID, DESKTOP)).toBe(false); + // A close arriving twice must not free an id a later stream is holding. + release(); + expect(service.hasLiveStream(USER_ID, DESKTOP)).toBe(false); + }); + it('reports no delivery when nothing holds the audio', async () => { const { pool } = poolWith({ deviceId: null }); const service = new PlaybackSyncService(pool); diff --git a/backend/src/services/playback-sync.service.ts b/backend/src/services/playback-sync.service.ts index 6a19059..e025e52 100644 --- a/backend/src/services/playback-sync.service.ts +++ b/backend/src/services/playback-sync.service.ts @@ -43,6 +43,12 @@ export interface PlaybackDevice { export interface PlaybackSnapshot { deviceId: string | null; trackId: string | null; + /** + * The durable Vibe session the owning device is playing, when it is playing + * one. Whichever device holds the audio drives the session, so this is how the + * next owner learns there is one to adopt. + */ + vibeSessionId: string | null; queue: unknown[]; queueIndex: number; position: number; @@ -53,6 +59,7 @@ export interface PlaybackSnapshot { export interface PlaybackStatePatch { trackId?: string | null; + vibeSessionId?: string | null; queue?: unknown[]; queueIndex?: number; position?: number; @@ -73,6 +80,7 @@ export class NotSessionOwnerError extends Error { type StateRow = { device_id: string | null; track_id: string | null; + vibe_session_id: string | null; queue: unknown[]; queue_index: number; position_ms: number; @@ -85,6 +93,7 @@ function toSnapshot(row: StateRow): PlaybackSnapshot { return { deviceId: row.device_id, trackId: row.track_id, + vibeSessionId: row.vibe_session_id ?? null, queue: Array.isArray(row.queue) ? row.queue : [], queueIndex: row.queue_index, position: row.position_ms / 1000, @@ -101,6 +110,15 @@ function clampPositionMs(position: number | undefined, fallback: number): number export class PlaybackSyncService { private readonly emitter = new EventEmitter(); + /** + * Open streams per device, keyed `userId:deviceId`. A device only exists as + * far as this session is concerned while it is holding one: that is what + * decides whether a command can be delivered, and whether a browser asking to + * reuse a stored device id would be colliding with a tab that already has it. + * Counted rather than a flag, so a reconnect racing its own close cannot leave + * a device permanently marked busy. + */ + private readonly streams = new Map(); constructor(private readonly pgPool: Pool) { // One session with many idle tabs is the normal case, and Node warns at ten @@ -113,6 +131,24 @@ export class PlaybackSyncService { return () => this.emitter.off(userId, listener); } + /** Mark a device's stream open until the returned function is called. */ + claimStream(userId: string, deviceId: string): () => void { + const key = `${userId}:${deviceId}`; + this.streams.set(key, (this.streams.get(key) ?? 0) + 1); + let released = false; + return () => { + if (released) return; + released = true; + const open = (this.streams.get(key) ?? 1) - 1; + if (open > 0) this.streams.set(key, open); + else this.streams.delete(key); + }; + } + + hasLiveStream(userId: string, deviceId: string): boolean { + return (this.streams.get(`${userId}:${deviceId}`) ?? 0) > 0; + } + private async publish(userId: string): Promise { const [state, devices] = await Promise.all([this.getState(userId), this.listDevices(userId)]); this.emitter.emit(userId, { type: 'state', state, devices } satisfies PlaybackEvent); @@ -122,10 +158,15 @@ export class PlaybackSyncService { * Register a device, or refresh one the browser already knows about. The * caller supplies the id it stored locally so a reload keeps its identity and * the device list does not grow one row per page load. + * + * A second tab of the same browser asks with the same stored id, and two tabs + * sharing one device id are one device that runs every command twice and plays + * two copies of the audio. An id whose stream is already open therefore does + * not get reused: the caller is given a device of its own instead. */ async registerDevice(userId: string, name: string, deviceId?: string | null): Promise { const cleanName = (name || 'Unknown device').trim().slice(0, 120) || 'Unknown device'; - if (deviceId) { + if (deviceId && !this.hasLiveStream(userId, deviceId)) { const updated = await this.pgPool.query<{ id: string; name: string; last_seen_at: Date }>( `UPDATE playback_devices SET name = $3, last_seen_at = NOW() WHERE id = $1 AND user_id = $2 @@ -197,7 +238,7 @@ export class PlaybackSyncService { const res = await this.pgPool.query( `INSERT INTO playback_state (user_id) VALUES ($1) ON CONFLICT (user_id) DO UPDATE SET user_id = EXCLUDED.user_id - RETURNING device_id, track_id, queue, queue_index, position_ms, is_playing, version, updated_at`, + RETURNING device_id, track_id, vibe_session_id, queue, queue_index, position_ms, is_playing, version, updated_at`, [userId] ); return toSnapshot(res.rows[0]); @@ -227,10 +268,15 @@ export class PlaybackSyncService { queue_index = COALESCE($6, queue_index), position_ms = COALESCE($7, position_ms), is_playing = COALESCE($8, is_playing), + -- Only the device driving the Vibe reports one, and it reports the + -- absence of one just as explicitly: ending a Vibe and playing an + -- album has to clear this, or the next owner would adopt a session + -- nothing is playing any more. + vibe_session_id = COALESCE($9, CASE WHEN $10 THEN NULL ELSE vibe_session_id END), version = version + 1, updated_at = NOW() WHERE user_id = $1 - RETURNING device_id, track_id, queue, queue_index, position_ms, is_playing, version, updated_at`, + RETURNING device_id, track_id, vibe_session_id, queue, queue_index, position_ms, is_playing, version, updated_at`, [ userId, deviceId, @@ -240,6 +286,8 @@ export class PlaybackSyncService { patch.queueIndex ?? null, patch.position === undefined ? null : clampPositionMs(patch.position, 0), patch.isPlaying ?? null, + patch.vibeSessionId ?? null, + patch.vibeSessionId === null, ] ); @@ -259,10 +307,15 @@ export class PlaybackSyncService { * Aim a command at whichever device holds the audio. Any device may send one, * including the owner itself — that is what makes a phone a remote for the * desktop without either side knowing which is which. + * + * Ownership outlives a closed stream, so the owner is not necessarily + * listening: a tab that was killed keeps the session until the sweep frees it. + * Emitting into that gap answered the presser with a success it did not get, + * so a command is only accepted while the owner has a stream to receive it on. */ async sendCommand(userId: string, command: PlaybackCommand): Promise<{ deliveredTo: string | null }> { const owner = await this.ownerId(userId); - if (!owner) return { deliveredTo: null }; + if (!owner || !this.hasLiveStream(userId, owner)) return { deliveredTo: null }; this.emitter.emit(userId, { type: 'command', deviceId: owner, command } satisfies PlaybackEvent); return { deliveredTo: owner }; } @@ -283,7 +336,7 @@ export class PlaybackSyncService { `UPDATE playback_state SET device_id = $2, version = version + 1, updated_at = NOW() WHERE user_id = $1 - RETURNING device_id, track_id, queue, queue_index, position_ms, is_playing, version, updated_at`, + RETURNING device_id, track_id, vibe_session_id, queue, queue_index, position_ms, is_playing, version, updated_at`, [userId, deviceId] ); await this.publish(userId); diff --git a/frontend/src/components/AudioEngine.test.tsx b/frontend/src/components/AudioEngine.test.tsx index 846ebb7..372fe54 100644 --- a/frontend/src/components/AudioEngine.test.tsx +++ b/frontend/src/components/AudioEngine.test.tsx @@ -1,4 +1,4 @@ -import { render, waitFor } from '@testing-library/react'; +import { act, render, waitFor } from '@testing-library/react'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { Track } from '../types'; import { usePlaybackStore } from '../store/usePlaybackStore'; @@ -28,7 +28,7 @@ describe('AudioEngine', () => { useVibeStore.getState().setActiveSession({ sessionId: 'session-a', seedTrackId: 'song' }); usePlaybackStore.setState({ currentTrack: track('song'), queue: [track('song')], currentIndex: 0, isPlaying: false, - queueOwner: 'vibe', vibeAdvanceHandler: () => undefined, + queueOwner: 'vibe', vibeAdvanceHandler: () => undefined, audioElsewhere: false, }); }); @@ -42,6 +42,24 @@ describe('AudioEngine', () => { Object.defineProperty(audio, 'ended', { value: false, configurable: true }); }; + it('loads no stream while another device holds the audio, and loads one when it comes back', () => { + usePlaybackStore.setState({ + audioElsewhere: true, isPlaying: false, prefetchNext: true, + queue: [track('song'), track('next')], currentIndex: 0, + }); + const { container } = render(); + const [active, idle] = Array.from(container.querySelectorAll('audio')); + + expect(active.getAttribute('src')).toBeNull(); + expect(idle.getAttribute('src')).toBeNull(); + // Mirroring is not playback: this device reports nothing to the session. + expect(reportVibeEvent).not.toHaveBeenCalled(); + + act(() => usePlaybackStore.getState().setAudioElsewhere(false)); + + expect(active.src).toContain('/tracks/song/stream'); + }); + it('buffers the next queued track into the idle element before the current one ends', () => { usePlaybackStore.setState({ queue: [track('song'), track('next')], currentIndex: 0, isPlaying: true, diff --git a/frontend/src/components/AudioEngine.tsx b/frontend/src/components/AudioEngine.tsx index baff593..fe0c217 100644 --- a/frontend/src/components/AudioEngine.tsx +++ b/frontend/src/components/AudioEngine.tsx @@ -213,7 +213,7 @@ export const AudioEngine = () => { /** Warm the predicted next track into the idle element. */ const prefetch = () => { const playback = store(); - if (!playback.prefetchNext) return; + if (!playback.prefetchNext || playback.audioElsewhere) return; const nextId = predictNextTrackId(); if (!nextId || nextId === loadedIdRef.current) return; const idleIdx = 1 - activeIdxRef.current; @@ -352,6 +352,20 @@ export const AudioEngine = () => { if (els.length === 0) return; const applyTrack = (id: string | null) => { + // Another device is playing and this one is only showing what it plays. + // Pointing an element at the stream here would download the track — the + // whole of it, once an earlier prefetch left that element on preload=auto + // — for audio that is never heard. Drop the sources and follow along. + if (usePlaybackStore.getState().audioElsewhere) { + cancelJoin(); + for (const idx of [0, 1]) retire(idx); + loadedIdRef.current = null; + preparedRef.current = null; + endedNaturallyRef.current = false; + crossedThresholdRef.current = false; + earlyAdvanceRef.current = false; + return; + } if (id === loadedIdRef.current) return; // A newer track change supersedes any tail still waiting to hand over. cancelJoin(); @@ -424,8 +438,15 @@ export const AudioEngine = () => { }; // Apply the current value immediately, then subscribe to future changes. + let lastElsewhere = usePlaybackStore.getState().audioElsewhere; applyTrack(usePlaybackStore.getState().currentTrack?.id ?? null); const unsub = usePlaybackStore.subscribe((state) => { + if (state.audioElsewhere !== lastElsewhere) { + lastElsewhere = state.audioElsewhere; + // The audio just came back to this device. Nothing is loaded, and the + // track id has not changed, so say so and let applyTrack load it. + if (!lastElsewhere) loadedIdRef.current = null; + } applyTrack(state.currentTrack?.id ?? null); }); return unsub; @@ -438,6 +459,8 @@ export const AudioEngine = () => { const apply = (isPlaying: boolean) => { const audio = els[activeIdxRef.current]; + // Nothing is loaded while another device holds the audio. + if (usePlaybackStore.getState().audioElsewhere) return; if (isPlaying) { // Never resume a finished element. Between `ended` and the next source // being loaded, isPlaying is still true, and resuming here replays the @@ -488,6 +511,9 @@ export const AudioEngine = () => { const apply = (position: number) => { const audio = els[activeIdxRef.current]; + // A mirrored position belongs to another device's playhead, and there is + // no source loaded here to move anyway. + if (usePlaybackStore.getState().audioElsewhere) return; if (Math.abs(audio.currentTime - position) > SEEK_THRESHOLD) { audio.currentTime = position; } diff --git a/frontend/src/components/NowPlayingPanel.tsx b/frontend/src/components/NowPlayingPanel.tsx index 9702f0a..1fc7c10 100644 --- a/frontend/src/components/NowPlayingPanel.tsx +++ b/frontend/src/components/NowPlayingPanel.tsx @@ -35,7 +35,7 @@ export function NowPlayingPanel({ onClose }: { onClose: () => void }) { previous?.focus(); }; }, []); - const { currentTrack, queue, isPlaying, position, duration } = usePlaybackStore(); + const { currentTrack, queue, position, duration } = usePlaybackStore(); const transport = useTransport(); const currentIdx = currentTrack ? queue.findIndex((t) => t.id === currentTrack.id) : -1; @@ -110,11 +110,11 @@ export function NowPlayingPanel({ onClose }: { onClose: () => void }) { diff --git a/frontend/src/components/PlaybackBar.tsx b/frontend/src/components/PlaybackBar.tsx index 901fb65..1f96435 100644 --- a/frontend/src/components/PlaybackBar.tsx +++ b/frontend/src/components/PlaybackBar.tsx @@ -16,7 +16,7 @@ interface PlaybackBarProps { } export function PlaybackBar({ queueOpen, lyricsOpen, onToggleQueue, onToggleLyrics }: PlaybackBarProps) { - const { currentTrack, isPlaying, position, duration, volume, shuffle, repeat, setVolume, toggleShuffle, cycleRepeat } = usePlaybackStore(); + const { currentTrack, position, duration, volume, shuffle, repeat, setVolume, toggleShuffle, cycleRepeat } = usePlaybackStore(); const dislikeTrack = useDislikeTrack(); const transport = useTransport(); @@ -90,9 +90,9 @@ export function PlaybackBar({ queueOpen, lyricsOpen, onToggleQueue, onToggleLyri onClick={transport.toggle} disabled={!currentTrack} className="transport-btn" - aria-label={isPlaying ? 'Pause' : 'Play'} + aria-label={transport.playing ? 'Pause' : 'Play'} > - {isPlaying ? : } + {transport.playing ? : }