music/cliamp/cliamp-ws.ts

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 = 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;
}