feuerwerk/src/viewport/streaming.ts

/**
 * Streaming a sound (#349) on the page's side: the copy is fetched a piece at a time, read page by
 * page, and decoded by WebCodecs a few seconds ahead of what is heard; the samples go to a
 * `stream-player` on the worklet and are let go there once played. Only the compressed bytes and the
 * first seconds of each copy are kept, so a long sound weighs a few megabytes rather than a hundred.
 */
import { OggOpusReader, type PlacedPacket } from '../core/music/ogg';
import { STREAM_RATE } from '../core/music/streamQueue';
import { StreamCursor, type StreamPlan } from '../core/music/streams';

/** Bytes fetched at a time, some 13 s at 160 kbit/s. */
const CHUNK_BYTES = 256 * 1024;
/** Samples of each copy decoded once it is named, so it starts at once (`StreamFile.ready`). */
const HEAD = 2 * STREAM_RATE;
/**
 * Samples decoded before a place jumped to and thrown away: measured in Chrome, 32 packets of 20 ms
 * make what follows the same to the bit as decoding from the start, 4 come out 0.06 off.
 */
const PREROLL = 32 * 960;
/** Samples kept sent ahead of what the worklet has read, and the most decoded at once. */
const LEAD = 4 * STREAM_RATE;
const BATCH = STREAM_RATE;

/** One copy, shared by every voice streaming it: its bytes as they come, and its first seconds decoded. */
export class StreamFile {
  readonly reader = new OggOpusReader();
  /** Samples per channel without the pre-skip, as the asset says. */
  head: Float32Array[] | null = null;
  failed = false;
  private fetched = 0;
  private complete = false;
  private fetching: Promise<void> | null = null;
  private heading: Promise<boolean> | null = null;

  constructor(
    readonly url: string,
    readonly frames: number,
  ) {}

  get channels(): number {
    return this.reader.channels ?? 2;
  }

  get preSkip(): number {
    return this.reader.preSkip ?? 0;
  }

  /** Bytes fetched so far (test hook). */
  get bytes(): number {
    return this.fetched;
  }

  /** Bytes of its head decoded, 0 before (test hook). */
  get headBytes(): number {
    return (this.head ?? []).reduce((sum, side) => sum + side.byteLength, 0);
  }

  /** Its head fetched and decoded; false where it cannot be had. */
  ready(): Promise<boolean> {
    return (this.heading ??= (async () => {
      try {
        this.head = await decodeSpan(this, 0, Math.min(HEAD, this.frames));
        return true;
      } catch (error: unknown) {
        this.failed = true;
        console.error(`[sounds] cannot stream ${this.url}`, error);
        return false;
      }
    })());
  }

  /** Fetches until the packet holding stream sample `sample` is there, or the copy has ended. */
  async reach(sample: number): Promise<void> {
    while (!this.complete && this.lastEnd() <= sample) await this.more();
  }

  /** Every byte of the copy, fetched if it is not yet: what the picture of it is decoded from. */
  async whole(): Promise<ArrayBuffer> {
    const response = await fetch(this.url);
    if (!response.ok) throw new Error(`HTTP ${response.status}`);
    return response.arrayBuffer();
  }

  private lastEnd(): number {
    const last = this.reader.packets.at(-1);
    return last ? last.at + last.samples : 0;
  }

  private more(): Promise<void> {
    return (this.fetching ??= this.fetchMore().finally(() => {
      this.fetching = null;
    }));
  }

  /** The next piece by `Range`; a server that sends the whole file instead is taken at its word. */
  private async fetchMore(): Promise<void> {
    const response = await fetch(this.url, {
      headers: { Range: `bytes=${this.fetched}-${this.fetched + CHUNK_BYTES - 1}` },
    });
    if (response.status === 416) {
      this.complete = true;
      return;
    }
    if (!response.ok) throw new Error(`HTTP ${response.status}`);
    const bytes = new Uint8Array(await response.arrayBuffer());
    if (response.status === 200 && this.fetched > 0) throw new Error('the server sent all of it again');
    this.reader.push(bytes);
    this.fetched += bytes.length;
    const total = Number(response.headers.get('content-range')?.split('/')[1]);
    if (response.status === 200 || bytes.length === 0 || (Number.isFinite(total) && this.fetched >= total))
      this.complete = true;
  }
}

/**
 * Decodes stream samples on one decoder, reading on from where it stopped while it is asked for what
 * follows, and settling first wherever it is asked to jump.
 */
class StreamDecoder {
  private decoder: AudioDecoder | null = null;
  /** The next packet to hand it, and the stream sample the next sample it gives back is at. */
  private next = 0;
  private outAt = 0;
  /** What it has given back and not yet been asked for, from `outAt` back. */
  private spare: Float32Array[] = [];
  private outputs: Float32Array[][] = [];
  private fed = 0;
  private given = 0;
  private wake: (() => void) | null = null;
  private failure: Error | null = null;

  constructor(private readonly file: StreamFile) {}

  /** The samples of the copy from `from` to `to`, counted without its pre-skip. */
  async read(from: number, to: number): Promise<Float32Array[]> {
    const { file } = this;
    const start = from + file.preSkip;
    const end = to + file.preSkip;
    const held = () => this.outAt - (this.spare[0]?.length ?? 0);
    if (!this.decoder || start !== held()) await this.jump(start);
    const spareFrom = held();
    const parts: Float32Array[][] = [this.spare];
    let have = this.outAt;
    this.spare = [];
    while (have < end) {
      await file.reach(have);
      const packets = file.reader.packets;
      if (this.next >= packets.length) break;
      // All of it that is fetched goes in at once: waiting on the page for each packet of 20 ms, a
      // busy page decoded slower than it played and the player went short (CI tasks 638, 640, 651).
      let upTo = have;
      while (this.next < packets.length && upTo < end) {
        const packet = packets[this.next]!;
        this.feed(packet);
        upTo = packet.at + packet.samples;
      }
      const got = await this.take();
      parts.push(got);
      have += got[0]?.length ?? 0;
    }
    this.outAt = have;
    const joined = join(parts, file.channels, have - spareFrom);
    const offset = start - spareFrom;
    const kept = joined.map((side) => side.slice(offset, offset + (end - start)));
    this.spare = joined.map((side) => side.slice(offset + (end - start)));
    return kept;
  }

  close(): void {
    if (this.decoder && this.decoder.state !== 'closed') this.decoder.close();
    this.decoder = null;
  }

  /** A new decoder, from far enough before `sample` that it has settled there; what came before goes. */
  private async jump(sample: number): Promise<void> {
    this.close();
    const { file } = this;
    await file.reach(sample);
    const packets = file.reader.packets;
    let index = 0;
    while (index + 1 < packets.length && packets[index + 1]!.at <= sample - PREROLL) index += 1;
    this.next = index;
    this.outAt = packets[index]?.at ?? 0;
    this.spare = [];
    this.outputs = [];
    this.fed = this.given = 0;
    this.failure = null;
    const decoder = new AudioDecoder({
      output: (data) => {
        const sides = Array.from({ length: file.channels }, (_, c) => {
          const side = new Float32Array(data.numberOfFrames);
          data.copyTo(side, { planeIndex: Math.min(c, data.numberOfChannels - 1), format: 'f32-planar' });
          return side;
        });
        data.close();
        this.outputs.push(sides);
        this.given += 1;
        this.wake?.();
      },
      error: (error) => {
        this.failure = error;
        this.wake?.();
      },
    });
    // No description: Chrome would apply the pre-skip by itself, and only to the first packet it gets.
    decoder.configure({ codec: 'opus', sampleRate: STREAM_RATE, numberOfChannels: file.channels });
    this.decoder = decoder;
    // Every packet that ends before where it is asked for settles it and is thrown away; `read` cuts
    // the rest off the one it starts in.
    while (this.next < packets.length && packets[this.next]!.at + packets[this.next]!.samples <= sample)
      this.feed(packets[this.next]!);
    await this.take();
    this.outAt = packets[this.next]?.at ?? this.outAt;
  }

  private feed(packet: PlacedPacket): void {
    const decoder = this.decoder!;
    decoder.decode(
      new EncodedAudioChunk({
        type: 'key',
        timestamp: Math.round((packet.at * 1e6) / STREAM_RATE),
        data: packet.data,
      }),
    );
    this.next += 1;
    this.fed += 1;
  }

  /** The samples of the packets handed over since the last call, once the decoder has given them back. */
  private async take(): Promise<Float32Array[]> {
    while (this.given < this.fed && !this.failure)
      await new Promise<void>((resolve) => {
        this.wake = resolve;
      });
    this.wake = null;
    if (this.failure) throw this.failure;
    const out = this.outputs.splice(0);
    return join(
      out,
      this.file.channels,
      out.reduce((sum, part) => sum + (part[0]?.length ?? 0), 0),
    );
  }
}

/** The samples of a copy from `from` to `to` on a decoder of their own: what a head is. */
export async function decodeSpan(file: StreamFile, from: number, to: number): Promise<Float32Array[]> {
  await file.reach(0);
  if (file.reader.channels === null) throw new Error('no Opus in it');
  const decoder = new StreamDecoder(file);
  try {
    return await decoder.read(from, to);
  } finally {
    decoder.close();
  }
}

function join(parts: readonly Float32Array[][], channels: number, length: number): Float32Array[] {
  return Array.from({ length: channels }, (_, c) => {
    const side = new Float32Array(Math.max(0, length));
    let at = 0;
    for (const part of parts) {
      const data = part[c] ?? part[0];
      if (!data) continue;
      side.set(data.subarray(0, Math.max(0, length - at)), at);
      at += data.length;
    }
    return side;
  });
}

/**
 * Keeps one `stream-player` fed along its plan: the head where the plan is in it, decoded samples
 * after that, never more than `LEAD` ahead of what it has read. Tells `ended` once it has played out.
 */
export class StreamFeed {
  private readonly cursor: StreamCursor;
  private readonly decoder: StreamDecoder;
  private node: AudioWorkletNode | null = null;
  private sent = 0;
  private used = 0;
  private pumping = false;
  private stopped = false;
  private finished = false;
  /** Samples that came too late to be heard, as the worklet last said (test hook). */
  skipped = 0;

  constructor(
    private readonly file: StreamFile,
    plan: StreamPlan,
    private readonly ended: () => void,
  ) {
    this.cursor = new StreamCursor(plan);
    this.decoder = new StreamDecoder(file);
  }

  /**
   * What the player starts with where its plan starts in the head: handed over as it is made, so it
   * is there for the first sample heard. Sent by message, it came a few milliseconds late, and a
   * sound started at once lost its attack. Empty where the plan starts past the head.
   */
  opening(): Float32Array[] {
    const at = this.cursor.peek();
    const head = this.file.head;
    if (at === null || !head || at >= (head[0]?.length ?? 0)) return [];
    const span = this.cursor.next(head[0]!.length - at)!;
    this.sent += span.to - span.from;
    return head.map((side) => side.slice(span.from, span.to));
  }

  /** Feeds that player from now on. */
  attach(node: AudioWorkletNode): void {
    this.node = node;
    node.port.onmessage = (event: MessageEvent) => {
      const data: unknown = event.data;
      if (data === 'ended') {
        this.close();
        this.ended();
      } else if (typeof data === 'object' && data !== null && 'used' in data) {
        const report = data as { used: number; skipped: number };
        this.used = report.used;
        this.skipped = report.skipped;
        void this.pump();
      }
    };
    void this.pump();
  }

  stop(): void {
    if (this.stopped) return;
    this.close();
    if (!this.node) return;
    this.node.port.postMessage('stop');
    this.node.port.close();
    this.node.disconnect();
  }

  /** Bytes of samples sent and not yet read by the worklet, at most `LEAD` of them (test hook). */
  get ahead(): number {
    return Math.max(0, this.sent - this.used) * this.file.channels * Float32Array.BYTES_PER_ELEMENT;
  }

  /** Whether it is decoding or sending right now: what an offline clock waits for (test hook). */
  get busy(): boolean {
    return this.pumping;
  }

  /** Read anew each time: a stop comes in while a stretch is being decoded. */
  private halted(): boolean {
    return this.stopped;
  }

  private close(): void {
    this.stopped = true;
    if (this.node) this.node.port.onmessage = null;
    this.decoder.close();
  }

  private async pump(): Promise<void> {
    const node = this.node;
    if (!node || this.pumping || this.stopped || this.finished) return;
    this.pumping = true;
    try {
      while (!this.halted() && this.sent - this.used < LEAD) {
        const at = this.cursor.peek();
        if (at === null) {
          this.finished = true;
          node.port.postMessage('end');
          break;
        }
        const head = this.file.head;
        const inHead = head !== null && at < (head[0]?.length ?? 0);
        const span = this.cursor.next(inHead ? head[0]!.length - at : BATCH)!;
        const samples = inHead
          ? head.map((side) => side.slice(span.from, span.to))
          : await this.decoder.read(span.from, span.to);
        if (this.halted()) break;
        node.port.postMessage(
          samples,
          samples.map((side) => side.buffer),
        );
        this.sent += span.to - span.from;
      }
    } catch (error: unknown) {
      console.error(`[sounds] streaming ${this.file.url} broke off`, error);
      if (!this.halted()) node.port.postMessage('end');
      this.finished = true;
    } finally {
      this.pumping = false;
    }
  }
}