From b83a6c7e77eccfeaf72d28e7af989fcc4ab55bbc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Thu, 30 Jul 2026 05:27:42 +0000 Subject: [PATCH] move the whole local-audio pipeline into the music sidecar MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cliamp playback was implemented entirely in officer: it located the cliamp binary, validated the requested path against the owner's home, faked a PTY with `script`, injected PULSE_SINK and an ALSA config shipped inside the API tree, spawned parec to capture the sink, and set up the pulseaudio daemon and the virtual_out null sink at every boot — about 356 lines of audio-pipeline knowledge in a process that is meant to be a proxy, and none of it owned by the sidecar whose whole job is music. all of it now lives in sidecar/music: cliamp-ws.ts serves both sockets (/cliamp/ws for the player, /cliamp/audio/ws for the PCM capture) on the loopback server it already runs, pulse-audio.ts does the daemon + sink setup at sidecar startup instead of at officer's, and the asoundrc moved next to the code that passes it. officer keeps the part that is actually its job — authenticating the browser — and relays frames both ways without reading them (api/cliamp/relay.ts, same dumb-pipe shape as the vault notifications relay). the browser's frame contract is unchanged, so the frontend is not touched. two things fixed on the way: the traversal check now requires a separator after the home path, so a sibling directory whose name merely starts with it can no longer pass; and the music proxy no longer special-cases /reindex and /reindex/stream by name to extend the idle timeout — it extends the whole prefix, because a proxy should not know which of the sidecar's routes are slow. the music-specific `files` query param is out of the shared WS envelope too: upgradeWs now carries the raw query string, which any relayed provider can use. Co-Authored-By: Claude Opus 4.8 --- src/server.tsx | 57 +---- src/servers/api/cliamp/audio-ws.ts | 91 ------- src/servers/api/cliamp/relay.ts | 110 +++++++++ src/servers/api/cliamp/websocket.ts | 201 --------------- src/servers/api/music/router.ts | 17 +- src/servers/api/music/sidecar-server.ts | 5 + .../{api/cliamp => sidecar/music}/asoundrc | 0 src/servers/sidecar/music/cliamp-ws.test.ts | 66 +++++ src/servers/sidecar/music/cliamp-ws.ts | 232 ++++++++++++++++++ src/servers/sidecar/music/index.ts | 16 ++ src/servers/sidecar/music/pulse-audio.ts | 48 ++++ 11 files changed, 491 insertions(+), 352 deletions(-) delete mode 100644 src/servers/api/cliamp/audio-ws.ts create mode 100644 src/servers/api/cliamp/relay.ts delete mode 100644 src/servers/api/cliamp/websocket.ts rename src/servers/{api/cliamp => sidecar/music}/asoundrc (100%) create mode 100644 src/servers/sidecar/music/cliamp-ws.test.ts create mode 100644 src/servers/sidecar/music/cliamp-ws.ts create mode 100644 src/servers/sidecar/music/pulse-audio.ts diff --git a/src/server.tsx b/src/server.tsx index 2fca4e2b..a6dacbe9 100644 --- a/src/server.tsx +++ b/src/server.tsx @@ -8,8 +8,7 @@ import { terminalWebsocket } from './servers/api/terminal/websocket'; import { chatWebsocket } from './servers/api/chat/websocket'; import { taskRunnerWebsocket } from './servers/api/tasks/task-executor'; import { pipelineWebsocket } from './servers/api/tasks/pipeline-executor'; -import { cliampWebsocket } from './servers/api/cliamp/websocket'; -import { cliampAudioWebsocket } from './servers/api/cliamp/audio-ws'; +import { cliampWebsocket, cliampAudioWebsocket } from './servers/api/cliamp/relay'; import { desktopWebsocket } from './servers/api/desktop/websocket'; import { vaultWebsocket, upgradeVaultWs } from './servers/api/vault/websocket'; import { findEntryByProxyId, touchEntry } from './servers/api/dev-server/router'; @@ -50,7 +49,7 @@ type WSData = { command?: string; cols?: number; rows?: number; - files?: string; + search?: string; // raw query string, for providers that relay it to a sidecar devServerPort?: number; devServerSlug?: string; wsProxyPath?: string; @@ -250,7 +249,6 @@ async function upgradeWs( const command = url.searchParams.get('command') ?? undefined; const cols = url.searchParams.get('cols') ? Number(url.searchParams.get('cols')) : undefined; const rows = url.searchParams.get('rows') ? Number(url.searchParams.get('rows')) : undefined; - const files = url.searchParams.get('files') ?? undefined; const ok = server.upgrade(req, { data: { userId: user.id, @@ -262,7 +260,7 @@ async function upgradeWs( command, cols, rows, - files, + search: url.search, }, }); if (!ok) return new Response('Upgrade failed', { status: 500 }); @@ -388,51 +386,8 @@ initQueue().catch((err) => console.error('[queue] failed to initialize:', err)); import { cleanupOnStartup } from './servers/api/tasks/pipeline-job-manager'; cleanupOnStartup().catch((err) => console.error('[pipeline-jobs] startup cleanup failed:', err)); -// Ensure PulseAudio is running with virtual sink for cliamp audio streaming -(async () => { - const pulseaudio = Bun.which('pulseaudio'); - const pactl = Bun.which('pactl'); - if (!pulseaudio || !pactl) { - console.log('[cliamp] pulseaudio not installed, skipping audio setup'); - return; - } - - // Start PulseAudio daemon if not running - const check = Bun.spawnSync({ cmd: [pulseaudio, '--check'], stdout: 'ignore', stderr: 'ignore' }); - if (check.exitCode !== 0) { - const start = Bun.spawnSync({ cmd: [pulseaudio, '--start', '-D'], stdout: 'ignore', stderr: 'ignore' }); - if (start.exitCode !== 0) { - console.error('[cliamp] failed to start pulseaudio'); - return; - } - console.log('[cliamp] pulseaudio started'); - } else { - console.log('[cliamp] pulseaudio already running'); - } - - // Load null sink if not already loaded - const sinks = Bun.spawnSync({ cmd: [pactl, 'list', 'short', 'sinks'], stdout: 'pipe', stderr: 'ignore' }); - const sinkList = sinks.stdout.toString(); - if (!sinkList.includes('virtual_out')) { - const load = Bun.spawnSync({ - cmd: [ - pactl, - 'load-module', - 'module-null-sink', - 'sink_name=virtual_out', - 'sink_properties=device.description=Virtual_Output', - ], - stdout: 'pipe', - stderr: 'pipe', - }); - if (load.exitCode !== 0) { - console.error('[cliamp] failed to load null sink:', load.stderr.toString().trim()); - } else { - console.log('[cliamp] virtual_out null sink loaded'); - } - } else { - console.log('[cliamp] virtual_out sink already exists'); - } -})(); +// PulseAudio and the `virtual_out` sink used to be set up here, at every boot of a process that has no +// audio responsibilities. They belong to the music sidecar, which owns both cliamp halves now +// (sidecar/music/pulse-audio.ts). // Pi check/install is handled by bootstrap.ts (imported above) diff --git a/src/servers/api/cliamp/audio-ws.ts b/src/servers/api/cliamp/audio-ws.ts deleted file mode 100644 index 5fc6bd40..00000000 --- a/src/servers/api/cliamp/audio-ws.ts +++ /dev/null @@ -1,91 +0,0 @@ -import type { ServerWebSocket } from 'bun'; -import { spawn, type Subprocess } from 'bun'; - -type WSData = { - userId: number; - email: string; - username: string; - role: string; -}; - -type AudioSession = { - proc: Subprocess; - closed: boolean; -}; - -const sessions = new Map, AudioSession>(); - -export const cliampAudioWebsocket = { - async open(ws: ServerWebSocket) { - const parecPath = Bun.which('parec'); - if (!parecPath) { - ws.close(4000, 'parec not found on host'); - return; - } - - let proc: Subprocess<'ignore', 'pipe', 'ignore'>; - try { - proc = spawn({ - cmd: [parecPath, '--format=s16le', '--rate=44100', '--channels=2', '-d', 'virtual_out.monitor'], - stdout: 'pipe', - stderr: 'ignore', - stdin: 'ignore', - }); - } catch { - ws.close(4000, 'Failed to start audio capture'); - return; - } - - const session: AudioSession = { proc, closed: false }; - sessions.set(ws, session); - console.log('[cliamp-audio] parec started, streaming to WS'); - - // Stream stdout chunks as binary WS frames - const reader = proc.stdout.getReader(); - let totalBytes = 0; - const pump = async () => { - try { - while (!session.closed) { - const { done, value } = await reader.read(); - if (done) break; - if (value && !session.closed) { - totalBytes += value.byteLength; - if (totalBytes <= value.byteLength) { - console.log(`[cliamp-audio] first chunk: ${value.byteLength} bytes`); - } - try { - ws.sendBinary(value); - } catch { - break; - } - } - } - } catch { - // stream ended or error - } finally { - if (!session.closed) { - session.closed = true; - try { ws.close(); } catch { /* ignore */ } - } - } - }; - - pump(); - }, - - message() { - // No client-to-server messages expected - }, - - close(ws: ServerWebSocket) { - console.log('[cliamp-audio] WS closed'); - const session = sessions.get(ws); - if (session) { - session.closed = true; - try { session.proc.kill(); } catch { /* ignore */ } - sessions.delete(ws); - } - }, - - drain() {}, -}; diff --git a/src/servers/api/cliamp/relay.ts b/src/servers/api/cliamp/relay.ts new file mode 100644 index 00000000..7a4fb1fd --- /dev/null +++ b/src/servers/api/cliamp/relay.ts @@ -0,0 +1,110 @@ +import type { ServerWebSocket } from 'bun'; +import { getMusicServerWsUrl } from '../music/sidecar-server'; + +// Platform side of the two cliamp sockets. Both used to spawn processes here — the `cliamp` player and a +// `parec` capture — which put the whole local-audio pipeline inside the thin proxy. They now live in the +// music sidecar (`sidecar/music/cliamp-ws.ts`), and this is what is left of them: authenticate the browser +// (done before the upgrade, in server.tsx), then pass frames through in both directions without reading +// them. Text or binary, no inspection — same dumb-pipe shape as the vault notifications relay. + +export type CliampWSData = { + provider: 'cliamp' | 'cliamp-audio'; + search?: string; // the browser's query string, forwarded minus the platform token +}; + +type UpstreamState = { + ws: WebSocket | null; + queue: (string | Uint8Array)[]; + ready: boolean; +}; + +// Bun hands frames over as `string | Buffer`; a Buffer is a Uint8Array at runtime, so forward as-is +// rather than copying every PCM chunk. +const asPayload = (raw: string | Buffer): string | Uint8Array => + typeof raw === 'string' ? raw : (raw as Uint8Array); + +// The sidecar has no use for the platform JWT and should not see it. +const forwardedQuery = (search: string | undefined): string => { + const params = new URLSearchParams(search ?? ''); + params.delete('token'); + const qs = params.toString(); + return qs ? `?${qs}` : ''; +}; + +function createCliampRelay(path: string) { + const upstreams = new Map, UpstreamState>(); + + return { + open(ws: ServerWebSocket) { + const base = getMusicServerWsUrl(); + if (!base) { + try { + ws.close(1011, 'Music sidecar not available'); + } catch { + /* already closed */ + } + return; + } + + const state: UpstreamState = { ws: null, queue: [], ready: false }; + upstreams.set(ws, state); + + const upstream = new WebSocket(`${base}${path}${forwardedQuery(ws.data.search)}`); + upstream.binaryType = 'arraybuffer'; + state.ws = upstream; + + upstream.addEventListener('open', () => { + state.ready = true; + for (const m of state.queue) upstream.send(m); + state.queue.length = 0; + }); + upstream.addEventListener('message', (ev) => { + try { + ws.send(ev.data as string | ArrayBuffer); + } catch { + /* client gone */ + } + }); + upstream.addEventListener('close', (ev) => { + upstreams.delete(ws); + try { + ws.close(ev.code || 1000, ev.reason || ''); + } catch { + /* already closed */ + } + }); + upstream.addEventListener('error', () => { + upstreams.delete(ws); + try { + ws.close(1011, 'upstream error'); + } catch { + /* already closed */ + } + }); + }, + + message(ws: ServerWebSocket, raw: string | Buffer) { + const state = upstreams.get(ws); + if (!state) return; + const payload = asPayload(raw); + if (state.ready && state.ws) state.ws.send(payload); + else state.queue.push(payload); // buffer until the upstream socket opens + }, + + close(ws: ServerWebSocket) { + const state = upstreams.get(ws); + if (!state) return; + try { + state.ws?.close(); + } catch { + /* already closed */ + } + upstreams.delete(ws); + }, + + drain() {}, + }; +} + +export const cliampWebsocket = createCliampRelay('/cliamp/ws'); +export const cliampAudioWebsocket = createCliampRelay('/cliamp/audio/ws'); diff --git a/src/servers/api/cliamp/websocket.ts b/src/servers/api/cliamp/websocket.ts deleted file mode 100644 index aabdc7f0..00000000 --- a/src/servers/api/cliamp/websocket.ts +++ /dev/null @@ -1,201 +0,0 @@ -import type { ServerWebSocket } from 'bun'; -import { spawn, type Subprocess } from 'bun'; -import { resolve, normalize, dirname, join } from 'node:path'; -import { fileURLToPath } from 'node:url'; -import { getOwnerHomeDir } from '@@/data-path'; - -const __dirname = dirname(fileURLToPath(import.meta.url)); -const ASOUNDRC_PATH = join(__dirname, 'asoundrc'); - -type WSData = { - userId: number; - email: string; - username: string; - files: string; -}; - -type CliampSession = { - proc: Subprocess<'pipe', 'pipe', 'pipe'>; - closed: boolean; -}; - -const sessions = new Map, CliampSession>(); - -const sendOutput = (ws: ServerWebSocket, data: string) => { - try { - ws.send(JSON.stringify({ type: 'output', data })); - } catch { - // ws already closed - } -}; - -const sendExit = (ws: ServerWebSocket) => { - try { - ws.send(JSON.stringify({ type: 'exit' })); - } catch { - // ws already closed - } -}; - -const validateFilePaths = (files: string[], homeDir: string): string[] | null => { - const resolved = files.map((f) => normalize(resolve(homeDir, f))); - for (const p of resolved) { - if (!p.startsWith(homeDir)) return null; - } - return resolved; -}; - -const findCliamp = (): string | null => { - const which = Bun.which('cliamp'); - if (which) return which; - const candidates = [ - process.env.GOPATH ? `${process.env.GOPATH}/bin/cliamp` : null, - `${process.env.HOME}/.local/go-path/bin/cliamp`, - `${process.env.HOME}/go/bin/cliamp`, - ]; - for (const bin of candidates) { - if (!bin) continue; - try { - const stat = Bun.spawnSync({ cmd: ['test', '-x', bin], stdout: 'ignore', stderr: 'ignore' }); - if (stat.exitCode === 0) return bin; - } catch { - /* ignore */ - } - } - return null; -}; - -const shellEscape = (s: string) => `'${s.replace(/'/g, "'\\''")}'`; - -export const cliampWebsocket = { - async open(ws: ServerWebSocket) { - const { email, files: filesParam } = ws.data; - - if (!filesParam) { - sendOutput(ws, '\r\n[Error] No files specified.\r\n'); - return; - } - - const cliampPath = findCliamp(); - if (!cliampPath) { - sendOutput(ws, '\r\n[Error] cliamp not found on host.\r\n'); - return; - } - - const homeDir = getOwnerHomeDir(email); - const rawFiles = [filesParam]; - - // Resolve paths relative to user home dir - const absoluteFiles = rawFiles.map((f) => { - if (f.startsWith('/')) return `${homeDir}${f}`; - return `${homeDir}/${f}`; - }); - - const validated = validateFilePaths(absoluteFiles, homeDir); - if (!validated) { - sendOutput(ws, '\r\n[Error] Invalid file path.\r\n'); - return; - } - - // Use `script` to allocate a PTY for cliamp (avoids node-pty dependency) - // script -qfc ' ' /dev/null - const cliampCmd = [shellEscape(cliampPath), ...validated.map(shellEscape)].join(' '); - console.log(`[cliamp] spawning: ${cliampCmd}`); - let proc: Subprocess<'pipe', 'pipe', 'pipe'>; - try { - proc = spawn({ - cmd: ['script', '-qfc', cliampCmd, '/dev/null'], - stdin: 'pipe', - stdout: 'pipe', - stderr: 'pipe', - cwd: homeDir, - env: { ...process.env, TERM: 'xterm-256color', PULSE_SINK: 'virtual_out', ALSA_CONFIG_PATH: ASOUNDRC_PATH }, - }); - } catch (err) { - const message = err instanceof Error ? err.message : 'Failed to start cliamp'; - sendOutput(ws, `\r\n[Error] ${message}\r\n`); - return; - } - - const session: CliampSession = { proc, closed: false }; - sessions.set(ws, session); - - // Pump stdout → WS - const reader = proc.stdout.getReader(); - const pump = async () => { - try { - while (!session.closed) { - const { done, value } = await reader.read(); - if (done) break; - if (value && !session.closed) { - sendOutput(ws, new TextDecoder().decode(value)); - } - } - } catch { - // stream ended - } finally { - if (!session.closed) { - session.closed = true; - sendExit(ws); - } - } - }; - pump(); - - // Also read stderr (cliamp may write there) - const stderrReader = proc.stderr.getReader(); - const pumpStderr = async () => { - try { - while (!session.closed) { - const { done, value } = await stderrReader.read(); - if (done) break; - if (value && !session.closed) { - sendOutput(ws, new TextDecoder().decode(value)); - } - } - } catch { - // stream ended - } - }; - pumpStderr(); - - // On process exit → notify client - proc.exited.then((code) => { - console.log(`[cliamp] process exited code=${code}`); - if (!session.closed) { - session.closed = true; - sendExit(ws); - } - sessions.delete(ws); - }); - }, - - message(ws: ServerWebSocket, raw: string | Buffer) { - const session = sessions.get(ws); - if (!session || session.closed) return; - - try { - const msg = JSON.parse(typeof raw === 'string' ? raw : raw.toString()); - if (msg.type === 'input' && msg.data) { - session.proc.stdin.write(msg.data); - } - } catch { - // ignore - } - }, - - close(ws: ServerWebSocket) { - const session = sessions.get(ws); - if (session) { - session.closed = true; - try { - session.proc.kill(); - } catch { - /* ignore */ - } - sessions.delete(ws); - } - }, - - drain() {}, -}; diff --git a/src/servers/api/music/router.ts b/src/servers/api/music/router.ts index c3cd6420..fac0dba8 100644 --- a/src/servers/api/music/router.ts +++ b/src/servers/api/music/router.ts @@ -24,15 +24,14 @@ musicRouter.all('/*', async (ctx) => { const target = `${baseUrl}${subpath}${url.search}`; // A from-scratch reindex holds this proxied connection open for minutes with no bytes flowing, which - // the main server's 60s idle timeout would drop. Extend it to 30 min for the build/progress endpoints - // (Bun passes the server as Hono's env). Matches the sidecar's own per-request extension. - if (subpath === '/reindex' || subpath === '/reindex/stream') { - const server = ctx.env as { timeout?: (req: Request, seconds: number) => void } | undefined; - try { - server?.timeout?.(ctx.req.raw, 1800); - } catch { - /* older Bun / no per-request timeout — the build still completes in the background */ - } + // the main server's 60s idle timeout would drop. Extend every request under this prefix to 30 min (Bun + // passes the server as Hono's env) — the proxy must not know which of the sidecar's routes are slow, and + // the sidecar applies its own per-request extension anyway. + const server = ctx.env as { timeout?: (req: Request, seconds: number) => void } | undefined; + try { + server?.timeout?.(ctx.req.raw, 1800); + } catch { + /* older Bun / no per-request timeout — the build still completes in the background */ } const method = ctx.req.method; diff --git a/src/servers/api/music/sidecar-server.ts b/src/servers/api/music/sidecar-server.ts index 2b92611a..22cd23f6 100644 --- a/src/servers/api/music/sidecar-server.ts +++ b/src/servers/api/music/sidecar-server.ts @@ -16,3 +16,8 @@ sidecar.on('music:server', (msg) => { export function getMusicServerUrl(): string | null { return serverPort ? `http://127.0.0.1:${serverPort}` : null; } + +/** Same server, ws:// scheme — for the cliamp sockets the platform relays (see api/cliamp/relay.ts). */ +export function getMusicServerWsUrl(): string | null { + return serverPort ? `ws://127.0.0.1:${serverPort}` : null; +} diff --git a/src/servers/api/cliamp/asoundrc b/src/servers/sidecar/music/asoundrc similarity index 100% rename from src/servers/api/cliamp/asoundrc rename to src/servers/sidecar/music/asoundrc diff --git a/src/servers/sidecar/music/cliamp-ws.test.ts b/src/servers/sidecar/music/cliamp-ws.test.ts new file mode 100644 index 00000000..2521c2f8 --- /dev/null +++ b/src/servers/sidecar/music/cliamp-ws.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it } from 'bun:test'; +import { cliampUpgradeData, musicWebsocket } from './cliamp-ws'; + +// The player socket refuses a path before it spawns anything, so these two cases exercise the whole +// server → handler → frame path without starting cliamp. Anything that would actually play needs a real +// file and a real audio sink, so it is not tested here. + +function serveOnce() { + const server = Bun.serve({ + port: 0, + hostname: '127.0.0.1', + fetch(req, srv) { + const url = new URL(req.url); + const data = cliampUpgradeData(url.pathname, url.searchParams); + if (data && srv.upgrade(req, { data })) return undefined as unknown as Response; + return new Response('nope', { status: 400 }); + }, + websocket: musicWebsocket, + }); + return server; +} + +function firstFrame(url: string): Promise { + return new Promise((resolve, reject) => { + const ws = new WebSocket(url); + const timer = setTimeout(() => reject(new Error('no frame')), 3000); + ws.addEventListener('message', (ev) => { + clearTimeout(timer); + ws.close(); + resolve(String(ev.data)); + }); + ws.addEventListener('error', () => { + clearTimeout(timer); + reject(new Error('socket error')); + }); + }); +} + +describe('cliamp player socket', () => { + it('rejects a path that escapes the owner home', async () => { + const server = serveOnce(); + try { + const frame = await firstFrame(`ws://127.0.0.1:${server.port}/cliamp/ws?files=../../etc/passwd`); + expect(JSON.parse(frame)).toEqual({ type: 'output', data: '\r\n[Error] Invalid file path.\r\n' }); + } finally { + server.stop(true); + } + }); + + it('reports a missing files param instead of spawning', async () => { + const server = serveOnce(); + try { + const frame = await firstFrame(`ws://127.0.0.1:${server.port}/cliamp/ws`); + expect(JSON.parse(frame)).toEqual({ type: 'output', data: '\r\n[Error] No files specified.\r\n' }); + } finally { + server.stop(true); + } + }); + + it('routes only the two cliamp paths', () => { + const q = new URLSearchParams(); + expect(cliampUpgradeData('/cliamp/ws', q)).toEqual({ kind: 'player', files: '' }); + expect(cliampUpgradeData('/cliamp/audio/ws', q)).toEqual({ kind: 'capture' }); + expect(cliampUpgradeData('/stream', q)).toBeNull(); + }); +}); diff --git a/src/servers/sidecar/music/cliamp-ws.ts b/src/servers/sidecar/music/cliamp-ws.ts new file mode 100644 index 00000000..7249d166 --- /dev/null +++ b/src/servers/sidecar/music/cliamp-ws.ts @@ -0,0 +1,232 @@ +import type { ServerWebSocket } from 'bun'; +import { spawn, type Subprocess } from 'bun'; +import { homedir } from 'node:os'; +import { join, normalize, resolve, sep } from 'node:path'; +import { VIRTUAL_SINK } from './pulse-audio'; + +// Local playback, both halves of it, owned by the process that owns the audio pipeline: +// +// /cliamp/ws — runs the `cliamp` TUI player against a file and pipes its terminal both ways +// /cliamp/audio/ws — captures what the sink hears and streams it to the browser as raw PCM +// +// Officer relays these two sockets and nothing else: it authenticates the browser and forwards frames. +// Every fact below — where the binary is, what a legal path is, which sink to play into, the ALSA config, +// the capture format — is pipeline knowledge and stays here. The frame shapes are the browser's contract +// ({type:'output'|'exit'} / {type:'input'} as JSON text, PCM as binary), so they are unchanged by the move. +// +// Both sockets are loopback-only, like the rest of this server: officer is the only client. + +const ASOUNDRC_PATH = join(import.meta.dir, 'asoundrc'); + +// Single super user, so the owner's home is the root every path is resolved against — same convention as +// stream-audio.ts and the indexer. +const ROOT_DIR = process.env.HOME_DIR ?? homedir(); + +// parec's output format IS the contract with the browser's AudioWorklet: signed 16-bit LE, 44.1kHz, stereo. +const CAPTURE_ARGS = ['--format=s16le', '--rate=44100', '--channels=2', '-d', `${VIRTUAL_SINK}.monitor`]; + +export type MusicWSData = { kind: 'player'; files: string } | { kind: 'capture' }; + +type Session = { + proc: Subprocess; + closed: boolean; +}; + +const sessions = new Map, Session>(); + +const sendOutput = (ws: ServerWebSocket, data: string) => { + try { + ws.send(JSON.stringify({ type: 'output', data })); + } catch { + /* ws already closed */ + } +}; + +const sendExit = (ws: ServerWebSocket) => { + try { + ws.send(JSON.stringify({ type: 'exit' })); + } catch { + /* ws already closed */ + } +}; + +// Home-relative or leading-slash paths both mean "under the owner's home"; anything that escapes it after +// normalisation is rejected. The trailing separator matters: without it a sibling directory whose name +// merely starts with the home path would pass. +const resolveInHome = (file: string): string | null => { + const abs = normalize(resolve(ROOT_DIR, file.startsWith('/') ? `.${file}` : file)); + return abs === ROOT_DIR || abs.startsWith(ROOT_DIR + sep) ? abs : null; +}; + +const findCliamp = (): string | null => { + const which = Bun.which('cliamp'); + if (which) return which; + const candidates = [ + process.env.GOPATH ? `${process.env.GOPATH}/bin/cliamp` : null, + `${ROOT_DIR}/.local/go-path/bin/cliamp`, + `${ROOT_DIR}/go/bin/cliamp`, + ]; + for (const bin of candidates) { + if (!bin) continue; + try { + const stat = Bun.spawnSync({ cmd: ['test', '-x', bin], stdout: 'ignore', stderr: 'ignore' }); + if (stat.exitCode === 0) return bin; + } catch { + /* ignore */ + } + } + return null; +}; + +const shellEscape = (s: string) => `'${s.replace(/'/g, "'\\''")}'`; + +// Pump a byte stream into the socket until it ends; `frame` decides how it lands on the wire. +function pump( + ws: ServerWebSocket, + session: Session, + stream: ReadableStream, + frame: (ws: ServerWebSocket, chunk: Uint8Array) => void, + onEnd?: () => void, +): void { + const reader = stream.getReader(); + void (async () => { + try { + while (!session.closed) { + const { done, value } = await reader.read(); + if (done) break; + if (value && !session.closed) frame(ws, value); + } + } catch { + /* stream ended */ + } finally { + onEnd?.(); + } + })(); +} + +function openPlayer(ws: ServerWebSocket, files: string): void { + if (!files) return sendOutput(ws, '\r\n[Error] No files specified.\r\n'); + + const cliampPath = findCliamp(); + if (!cliampPath) return sendOutput(ws, '\r\n[Error] cliamp not found on host.\r\n'); + + const target = resolveInHome(files); + if (!target) return sendOutput(ws, '\r\n[Error] Invalid file path.\r\n'); + + // `script` fakes a PTY for cliamp, which avoids a node-pty native dependency here. + const cliampCmd = `${shellEscape(cliampPath)} ${shellEscape(target)}`; + console.log(`[music] cliamp spawning: ${cliampCmd}`); + let proc: Subprocess<'pipe', 'pipe', 'pipe'>; + try { + proc = spawn({ + cmd: ['script', '-qfc', cliampCmd, '/dev/null'], + stdin: 'pipe', + stdout: 'pipe', + stderr: 'pipe', + cwd: ROOT_DIR, + env: { ...process.env, TERM: 'xterm-256color', PULSE_SINK: VIRTUAL_SINK, ALSA_CONFIG_PATH: ASOUNDRC_PATH }, + }); + } catch (err) { + return sendOutput(ws, `\r\n[Error] ${err instanceof Error ? err.message : 'Failed to start cliamp'}\r\n`); + } + + const session: Session = { proc, closed: false }; + sessions.set(ws, session); + + const decoder = new TextDecoder(); + const asText = (sock: ServerWebSocket, chunk: Uint8Array) => sendOutput(sock, decoder.decode(chunk)); + const end = () => { + if (session.closed) return; + session.closed = true; + sendExit(ws); + }; + pump(ws, session, proc.stdout, asText, end); + pump(ws, session, proc.stderr, asText); // cliamp writes some output there + + void proc.exited.then((code) => { + console.log(`[music] cliamp exited code=${code}`); + end(); + sessions.delete(ws); + }); +} + +function openCapture(ws: ServerWebSocket): void { + const parecPath = Bun.which('parec'); + if (!parecPath) { + ws.close(4000, 'parec not found on host'); + return; + } + + let proc: Subprocess<'ignore', 'pipe', 'ignore'>; + try { + proc = spawn({ cmd: [parecPath, ...CAPTURE_ARGS], stdin: 'ignore', stdout: 'pipe', stderr: 'ignore' }); + } catch { + ws.close(4000, 'Failed to start audio capture'); + return; + } + + const session: Session = { proc, closed: false }; + sessions.set(ws, session); + console.log('[music] parec started, streaming PCM to the relay'); + + pump( + ws, + session, + proc.stdout, + (sock, chunk) => { + try { + sock.sendBinary(chunk); + } catch { + session.closed = true; + } + }, + () => { + if (session.closed) return; + session.closed = true; + try { + ws.close(); + } catch { + /* already closed */ + } + }, + ); +} + +export const musicWebsocket = { + open(ws: ServerWebSocket) { + if (ws.data.kind === 'player') openPlayer(ws, ws.data.files); + else openCapture(ws); + }, + + message(ws: ServerWebSocket, raw: string | Buffer) { + const session = sessions.get(ws); + if (!session || session.closed || ws.data.kind !== 'player') return; // capture is one-way + try { + const msg = JSON.parse(typeof raw === 'string' ? raw : raw.toString()); + if (msg.type === 'input' && msg.data) (session.proc as Subprocess<'pipe'>).stdin.write(msg.data); + } catch { + /* not a frame we understand */ + } + }, + + close(ws: ServerWebSocket) { + const session = sessions.get(ws); + if (!session) return; + session.closed = true; + try { + session.proc.kill(); + } catch { + /* already gone */ + } + sessions.delete(ws); + }, + + drain() {}, +}; + +/** Upgrade one of the two cliamp sockets, or return null if this request is not for them. */ +export function cliampUpgradeData(pathname: string, search: URLSearchParams): MusicWSData | null { + if (pathname === '/cliamp/ws') return { kind: 'player', files: search.get('files') ?? '' }; + if (pathname === '/cliamp/audio/ws') return { kind: 'capture' }; + return null; +} diff --git a/src/servers/sidecar/music/index.ts b/src/servers/sidecar/music/index.ts index 0c60c4b5..083d9270 100644 --- a/src/servers/sidecar/music/index.ts +++ b/src/servers/sidecar/music/index.ts @@ -3,6 +3,8 @@ import { join, basename } from 'node:path'; import type { SidecarCommand, SidecarEvent } from '../protocol'; import { createSidecarConnector } from '../connect'; import { streamAudioFile } from './stream-audio'; +import { cliampUpgradeData, musicWebsocket } from './cliamp-ws'; +import { ensurePulseAudio } from './pulse-audio'; import { startNightlyReindex, stopNightlyReindex } from './nightly-reindex'; import { startMusicWatcher, stopMusicWatcher } from './watcher'; import { @@ -137,6 +139,10 @@ startNightlyReindex(); // Recursive watcher on ~/Music → localized reindex on any change. startMusicWatcher(); +// PulseAudio daemon + the `virtual_out` null sink both cliamp halves depend on. Officer used to do this at +// its own boot, which meant every restart of a process with no audio responsibilities re-checked the sink. +ensurePulseAudio(); + const server = Bun.serve({ port, hostname: '127.0.0.1', @@ -145,6 +151,15 @@ const server = Bun.serve({ idleTimeout: 255, async fetch(req, server) { const url = new URL(req.url); + + // The two cliamp sockets. Officer has already authenticated the browser and is relaying frames; the + // player and the capture themselves live here (cliamp-ws.ts). + const wsData = cliampUpgradeData(url.pathname, url.searchParams); + if (wsData) { + if (server.upgrade(req, { data: wsData })) return undefined as unknown as Response; + return new Response('Expected a WebSocket upgrade', { status: 400 }); + } + // A from-scratch reindex can take many minutes with no bytes flowing on the triggering request. // Give the build endpoints a 30-min idle timeout so they aren't dropped (/manifest is a pure read now). if (url.pathname === '/reindex' || url.pathname === '/reindex/stream') { @@ -419,6 +434,7 @@ const server = Bun.serve({ return new Response('Not found', { status: 404 }); }, + websocket: musicWebsocket, }); console.log(`[music] audio server listening on http://127.0.0.1:${port}`); diff --git a/src/servers/sidecar/music/pulse-audio.ts b/src/servers/sidecar/music/pulse-audio.ts new file mode 100644 index 00000000..3c4f207c --- /dev/null +++ b/src/servers/sidecar/music/pulse-audio.ts @@ -0,0 +1,48 @@ +// Host audio plumbing for local playback: a PulseAudio daemon and a null sink named `virtual_out`. +// cliamp plays *into* that sink (PULSE_SINK) and the capture side reads `virtual_out.monitor`, so the +// sink has to exist before either of them starts — which is why this runs at sidecar startup rather +// than on first play. Both steps are idempotent and both failures are non-fatal: a host without +// pulseaudio simply has no browser playback, and everything else the music sidecar does still works. + +export const VIRTUAL_SINK = 'virtual_out'; + +export function ensurePulseAudio(): void { + const pulseaudio = Bun.which('pulseaudio'); + const pactl = Bun.which('pactl'); + if (!pulseaudio || !pactl) { + console.log('[music] pulseaudio not installed, skipping audio setup'); + return; + } + + const check = Bun.spawnSync({ cmd: [pulseaudio, '--check'], stdout: 'ignore', stderr: 'ignore' }); + if (check.exitCode !== 0) { + const start = Bun.spawnSync({ cmd: [pulseaudio, '--start', '-D'], stdout: 'ignore', stderr: 'ignore' }); + if (start.exitCode !== 0) { + console.error('[music] failed to start pulseaudio'); + return; + } + console.log('[music] pulseaudio started'); + } else { + console.log('[music] pulseaudio already running'); + } + + const sinks = Bun.spawnSync({ cmd: [pactl, 'list', 'short', 'sinks'], stdout: 'pipe', stderr: 'ignore' }); + if (sinks.stdout.toString().includes(VIRTUAL_SINK)) { + console.log(`[music] ${VIRTUAL_SINK} sink already exists`); + return; + } + + const load = Bun.spawnSync({ + cmd: [ + pactl, + 'load-module', + 'module-null-sink', + `sink_name=${VIRTUAL_SINK}`, + 'sink_properties=device.description=Virtual_Output', + ], + stdout: 'pipe', + stderr: 'pipe', + }); + if (load.exitCode !== 0) console.error('[music] failed to load null sink:', load.stderr.toString().trim()); + else console.log(`[music] ${VIRTUAL_SINK} null sink loaded`); +}