feat(plugin): audio stream

src/i18n/resources/en.json

@@ -351,6 +351,40 @@
       "description": "Apply compression to audio (lowers the volume of the loudest parts of the signal and raises the volume of the softest parts)",
       "name": "Audio Compressor"
     },
+    "audio-stream": {
+      "description": "Stream audio as Opus/Ogg over HTTP (Icecast-compatible) for external applications",
+      "menu": {
+        "port": {
+          "label": "Port"
+        },
+        "quality-latency": {
+          "label": "Quality & Latency",
+          "submenu": {
+            "bitrate": {
+              "label": "Bitrate"
+            },
+            "buffer-size": {
+              "label": "Buffer Size"
+            },
+            "channels": {
+              "label": "Channels",
+              "mono": "Mono",
+              "stereo": "Stereo"
+            },
+            "sample-rate": {
+              "label": "Sample Rate"
+            }
+          }
+        }
+      },
+      "name": "Audio Stream",
+      "prompt": {
+        "port": {
+          "label": "Enter the port for the audio stream server:\nStream URL: {{streamUrl}}",
+          "title": "Audio Stream Port"
+        }
+      }
+    },
     "auth-proxy-adapter": {
       "description": "Support for the use of authentication proxy services",
       "menu": {

src/pear-desktop.ts

@@ -39,3 +39,9 @@ declare module '*.css?inline' {
 
   export default css;
 }
+
+declare module '*.js?raw' {
+  const javascript: string;
+
+  export default javascript;
+}

src/plugins/audio-stream/BroadcastStream.ts

@@ -0,0 +1,48 @@
+export class BroadcastStream {
+  private subscribers: Set<ReadableStreamDefaultController<Uint8Array>> =
+    new Set();
+
+  // Get a new stream. Any priming pages (e.g. cached Ogg header pages) are
+  // enqueued first so the subscriber can initialise its decoder before the
+  // live audio pages arrive.
+  subscribe(primingPages: Uint8Array[] = []) {
+    let controller!: ReadableStreamDefaultController<Uint8Array>;
+    const subscribers = this.subscribers;
+    const stream = new ReadableStream<Uint8Array>({
+      start(c) {
+        controller = c;
+        for (const page of primingPages) c.enqueue(page);
+        subscribers.add(c);
+      },
+      cancel() {
+        subscribers.delete(controller);
+      },
+    });
+
+    return stream;
+  }
+
+  // Write data to all readers.
+  write(chunk: Uint8Array) {
+    for (const controller of this.subscribers) {
+      // Drop slow clients whose queue has backed up rather than buffering
+      // chunks for them unboundedly.
+      if ((controller.desiredSize ?? 0) <= 0) {
+        this.subscribers.delete(controller);
+        continue;
+      }
+      try {
+        controller.enqueue(chunk);
+      } catch {
+        this.subscribers.delete(controller);
+      }
+    }
+  }
+
+  close() {
+    for (const controller of this.subscribers) {
+      controller.close();
+    }
+    this.subscribers.clear();
+  }
+}

src/plugins/audio-stream/StreamProcessor.js

@@ -0,0 +1,36 @@
+/* eslint-disable no-undef, typescript/no-unsafe-assignment, typescript/no-unsafe-member-access, typescript/no-unsafe-call */
+class RecorderProcessor extends AudioWorkletProcessor {
+  constructor(options) {
+    super();
+    const bufferSize = options?.processorOptions?.bufferSize || 4096;
+    // Prepare an interleaved stereo buffer [L,R,L,R,...]
+    this.buffer = new Float32Array(bufferSize * 2);
+    this.bufferIndex = 0;
+  }
+
+  process(inputs, outputs) {
+    const input = inputs[0];
+    if (input && input[0]) {
+      const left = input[0];
+      const right = input[1] || left; // if mono input, duplicate for right
+      for (let i = 0; i < left.length; i++) {
+        this.buffer[this.bufferIndex++] = left[i];
+        this.buffer[this.bufferIndex++] = right[i];
+        if (this.bufferIndex >= this.buffer.length) {
+          // Buffer full: send a copy to the main thread
+          this.port.postMessage(new Float32Array(this.buffer));
+          this.bufferIndex = 0;
+        }
+      }
+    }
+    // Optionally pass the audio through unchanged. outputs[0] can be an empty
+    // array (no channels connected), so guard each channel before .set().
+    if (outputs[0] && inputs[0]) {
+      if (outputs[0][0] && inputs[0][0]) outputs[0][0].set(inputs[0][0]);
+      if (outputs[0][1] && inputs[0][1]) outputs[0][1].set(inputs[0][1]);
+    }
+    return true;
+  }
+}
+
+registerProcessor('recorder-processor', RecorderProcessor);

src/plugins/audio-stream/backend.ts

@@ -0,0 +1,135 @@
+import { serve, type ServerType } from '@hono/node-server';
+import { Hono } from 'hono';
+import { stream } from 'hono/streaming';
+
+import { registerCallback, type SongInfo } from '@/providers/song-info';
+import { createBackend } from '@/utils';
+
+import { BroadcastStream } from './BroadcastStream';
+import { type AudioStreamConfig } from './config';
+import { OggOpusMuxer, OggDechainer } from './ogg-opus';
+
+const VENDOR = 'Pear Desktop';
+
+let config: AudioStreamConfig;
+const broadcast = new BroadcastStream();
+
+// Current track metadata (no ffprobe - comes straight from the player).
+let currentSong: SongInfo | null = null;
+
+// Chained Ogg/Opus stream: one logical stream per song. Pages go straight to
+// every subscriber; the muxer caches the current stream's header pages for late
+// joiners.
+const muxer = new OggOpusMuxer((page) => broadcast.write(page));
+
+// OpusTags comments for the current track (text only).
+function currentComments(): string[] {
+  const comments: string[] = [];
+  if (currentSong?.title) comments.push(`TITLE=${currentSong.title}`);
+  if (currentSong?.artist) comments.push(`ARTIST=${currentSong.artist}`);
+  if (currentSong?.album) comments.push(`ALBUM=${currentSong.album}`);
+  return comments.length ? comments : ['TITLE=Pear Desktop'];
+}
+
+export const backend = createBackend<
+  {
+    app: Hono;
+    server?: ServerType;
+  },
+  AudioStreamConfig
+>({
+  app: new Hono().get('/stream', (ctx) => {
+    // Per-song TEXT metadata is carried in-band via chained Ogg logical streams
+    // (OpusTags per track). Some clients can't follow chains - browsers
+    // (<audio>/MSE) reload on a new BOS, and VLC's clock chokes on it - so they
+    // get a de-chained single logical stream instead.
+    const ua = ctx.req.header('User-Agent') ?? '';
+    const needsDechain = /Mozilla|VLC/i.test(ua);
+
+    ctx.header('Content-Type', 'audio/ogg');
+    ctx.header('Transfer-Encoding', 'chunked');
+    ctx.header('Access-Control-Allow-Origin', '*');
+
+    if (!needsDechain) {
+      ctx.header('icy-name', 'Pear Desktop');
+      ctx.header('icy-url', 'https://github.com/pear-devs/pear-desktop');
+      ctx.header(
+        'icy-audio-info',
+        `ice-channels=${config.channels};ice-samplerate=48000;ice-bitrate=${Math.round(
+          config.bitrate / 1000,
+        )}`,
+      );
+      ctx.header('icy-pub', '1');
+    }
+
+    ctx.header('Server', 'Pear Desktop');
+
+    return stream(ctx, async (stream) => {
+      // New subscriber gets the cached OpusHead + OpusTags pages first, so the
+      // decoder can initialise before any audio page arrives.
+      let readable = broadcast.subscribe(muxer.headerPages);
+
+      if (needsDechain) {
+        const dechainer = new OggDechainer();
+        readable = readable.pipeThrough(
+          new TransformStream<Uint8Array, Uint8Array>({
+            transform(page, controller) {
+              for (const out of dechainer.push(page)) controller.enqueue(out);
+            },
+          }),
+        );
+      }
+
+      return await stream.pipe(readable);
+    });
+  }),
+
+  async start({ getConfig, ipc }) {
+    config = await getConfig();
+
+    this.server = serve(
+      {
+        fetch: this.app.fetch.bind(this.app),
+        hostname: config.hostname,
+        port: config.port,
+      },
+      ({ address, port }) => console.log('Listening on', { address, port }),
+    );
+
+    // Track metadata (no ffprobe needed). On an actual song change, start a new
+    // logical stream so the new title/artist/album are embedded in-band.
+    // SongInfo also fires for play/pause and time updates, so gate on videoId.
+    let lastVideoId = '';
+    registerCallback((songInfo: SongInfo) => {
+      currentSong = songInfo;
+      if (songInfo.videoId && songInfo.videoId !== lastVideoId) {
+        lastVideoId = songInfo.videoId;
+        if (muxer.ready) muxer.chain(VENDOR, currentComments());
+      }
+    });
+
+    // OpusHead (from WebCodecs decoderConfig.description) opens the first stream.
+    ipc.on('audio-stream:opus-head', (head: Uint8Array) => {
+      muxer.setHead(head);
+      muxer.start(VENDOR, currentComments());
+    });
+
+    // Each Opus packet → one Ogg audio page. durationUs is the packet length;
+    // Opus granule positions are counted in 48 kHz samples.
+    ipc.on(
+      'audio-stream:opus',
+      (packet: { bytes: Uint8Array; durationUs: number }) => {
+        const samples = (packet.durationUs * 48000) / 1_000_000;
+        muxer.writePacket(packet.bytes, samples);
+      },
+    );
+  },
+  async stop() {
+    if (!this.server) return;
+
+    await new Promise<void>((resolve) => this.server!.close(() => resolve()));
+  },
+  onConfigChange(newConfig) {
+    config = newConfig;
+  },
+});

src/plugins/audio-stream/config.ts

@@ -0,0 +1,20 @@
+export interface AudioStreamConfig {
+  enabled: boolean;
+  port: number;
+  hostname: string;
+  // Audio settings for the Opus/Ogg stream.
+  sampleRate: number; // Capture sample rate hint (Opus always encodes at 48 kHz)
+  bitrate: number; // Opus target bitrate in bits/sec (e.g. 128000)
+  channels: number; // Number of channels (2 = stereo)
+  bufferSize: number; // AudioWorklet batch size in frames - affects latency
+}
+
+export const defaultAudioStreamConfig: AudioStreamConfig = {
+  enabled: false,
+  port: 8765,
+  hostname: '0.0.0.0',
+  sampleRate: 48000, // 48kHz - Opus native rate
+  bitrate: 128000, // 128 kbps Opus
+  channels: 2, // Stereo
+  bufferSize: 4096, // AudioWorklet batch (frames)
+};

src/plugins/audio-stream/index.ts

@@ -0,0 +1,17 @@
+import { t } from '@/i18n';
+import { createPlugin } from '@/utils';
+
+import { backend } from './backend';
+import { defaultAudioStreamConfig } from './config';
+import { onMenu } from './menu';
+import { renderer } from './renderer';
+
+export default createPlugin({
+  name: () => t('plugins.audio-stream.name'),
+  description: () => t('plugins.audio-stream.description'),
+  restartNeeded: false,
+  config: defaultAudioStreamConfig,
+  backend,
+  renderer,
+  menu: onMenu,
+});