officerdev/platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'sidecars-music' into sidecars

Browse Source
This commit is contained in:
pastilhas
2026-07-30 05:43:01 +00:00
parent 5efffa9dcb b83a6c7e77
commit f724d4b9d4
11 changed files with 491 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server.tsx
+6 -51
View File
@@ -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)
src/servers/api/cliamp/audio-ws.ts
-91
View File
@@ -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<ServerWebSocket<WSData>, AudioSession>();
export const cliampAudioWebsocket = {
async open(ws: ServerWebSocket<WSData>) {
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<WSData>) {
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() {},
};
src/servers/api/cliamp/relay.ts
+110
View File
@@ -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<ArrayBuffer>)[];
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<ArrayBuffer> =>
typeof raw === 'string' ? raw : (raw as Uint8Array<ArrayBuffer>);
// 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<ServerWebSocket<CliampWSData>, UpstreamState>();
return {
open(ws: ServerWebSocket<CliampWSData>) {
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<CliampWSData>, 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<CliampWSData>) {
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');
src/servers/api/cliamp/websocket.ts
-201
View File
@@ -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<ServerWebSocket<WSData>, CliampSession>();
const sendOutput = (ws: ServerWebSocket<WSData>, data: string) => {
try {
ws.send(JSON.stringify({ type: 'output', data }));
} catch {
// ws already closed
}
};
const sendExit = (ws: ServerWebSocket<WSData>) => {
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<WSData>) {
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 '<command> <args...>' /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<WSData>, 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<WSData>) {
const session = sessions.get(ws);
if (session) {
session.closed = true;
try {
session.proc.kill();
} catch {
/* ignore */
}
sessions.delete(ws);
}
},
drain() {},
};
src/servers/api/music/router.ts
+8 -9
View File
@@ -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;
src/servers/api/music/sidecar-server.ts
+5
View File
@@ -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;
}
src/servers/api/cliamp/asoundrc → src/servers/sidecar/music/asoundrc
View File
src/servers/sidecar/music/cliamp-ws.test.ts
+66
View File
@@ -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<string> {
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();
});
});
src/servers/sidecar/music/cliamp-ws.ts
+232
View File
@@ -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<ServerWebSocket<MusicWSData>, Session>();
const sendOutput = (ws: ServerWebSocket<MusicWSData>, data: string) => {
try {
ws.send(JSON.stringify({ type: 'output', data }));
} catch {
/* ws already closed */
}
};
const sendExit = (ws: ServerWebSocket<MusicWSData>) => {
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<MusicWSData>,
session: Session,
stream: ReadableStream<Uint8Array>,
frame: (ws: ServerWebSocket<MusicWSData>, 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<MusicWSData>, 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<MusicWSData>, 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<MusicWSData>): 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<MusicWSData>) {
if (ws.data.kind === 'player') openPlayer(ws, ws.data.files);
else openCapture(ws);
},
message(ws: ServerWebSocket<MusicWSData>, 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<MusicWSData>) {
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;
}
src/servers/sidecar/music/index.ts
+16
View File
@@ -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}`);
src/servers/sidecar/music/pulse-audio.ts
+48
View File
@@ -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`);
}