carplay/test/playback.test.js

import assert from 'node:assert/strict';
import { execFileSync, spawn } from 'node:child_process';
import { once } from 'node:events';
import fs from 'node:fs/promises';
import { createServer } from 'node:http';
import os from 'node:os';
import path from 'node:path';
import { setTimeout as delay } from 'node:timers/promises';
import { after, before, test } from 'node:test';
import { fileURLToPath } from 'node:url';

import WebSocket from 'ws';

const root = fileURLToPath(new URL('..', import.meta.url));
let directory;
let movie;
let live;
let videoOnly;
let audioOnly;
let combined;

before(async () => {
  directory = await fs.mkdtemp(path.join(os.tmpdir(), 'carplay-playback-test-'));
  const inputs = ['-y', '-hide_banner', '-loglevel', 'error',
    '-f', 'lavfi', '-i', 'testsrc2=size=160x90:rate=12',
    '-f', 'lavfi', '-i', 'sine=frequency=440:sample_rate=48000', '-t', '8'];
  execFileSync('ffmpeg', [...inputs, '-c:v', 'mpeg4', '-c:a', 'aac', `${directory}/movie.mp4`]);
  execFileSync('ffmpeg', [...inputs, '-c:v', 'mpeg2video', '-c:a', 'mp2', '-f', 'mpegts', `${directory}/live.ts`]);
  const file = await fs.readFile(`${directory}/movie.mp4`);
  combined = file;
  execFileSync('ffmpeg', ['-y', '-hide_banner', '-loglevel', 'error', '-i', `${directory}/movie.mp4`,
    '-map', '0:v', '-c', 'copy', '-an', `${directory}/video.mp4`,
    '-map', '0:a', '-c', 'copy', '-vn', `${directory}/audio.m4a`]);
  videoOnly = await fs.readFile(`${directory}/video.mp4`);
  audioOnly = await fs.readFile(`${directory}/audio.m4a`);
  let offset = 0;
  while (file.toString('ascii', offset + 4, offset + 8) !== 'moov') {
    const size = file.readUInt32BE(offset);
    assert.ok(size >= 8 && offset + size < file.length);
    offset += size;
  }
  // Leave media chunk offsets intact, but put metadata beyond a large padding
  // box. HTTP seeking should skip it instead of downloading it through a pipe.
  const padding = Buffer.alloc(16 * 1024 * 1024);
  padding.writeUInt32BE(padding.length);
  padding.write('free', 4);
  movie = Buffer.concat([file.subarray(0, offset), padding, file.subarray(offset)]);
  live = await fs.readFile(`${directory}/live.ts`);
});

after(async () => { await fs.rm(directory, { recursive: true, force: true }); });

async function until(check, timeoutMs = 10000) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    const result = await check();
    if (result) return result;
    await delay(50);
  }
  throw new Error('Timed out waiting for playback condition');
}

async function startApp(t, { mode = 'relay', probe = 'real', resolver, resolverMetadata = {}, resolverFormat, singleConnectionHosts = '', metadataEnabled = '0' } = {}) {
  const temp = await fs.mkdtemp(path.join(directory, 'app-'));
  const requests = [];
  const upstream = createServer((request, response) => {
    const pathname = new URL(request.url, 'http://localhost').pathname;
    const isLive = pathname === '/live' || pathname === '/iptv.mp4';
    const data = isLive ? live : pathname === '/video.mp4' ? videoOnly
      : pathname === '/audio.m4a' ? audioOnly : pathname === '/combined.mp4' ? combined : movie;
    const branch = pathname === '/video.mp4' ? 'video' : pathname === '/audio.m4a' ? 'audio' : null;
    if (branch && (request.headers['x-branch'] !== branch || request.headers['x-common'] !== 'shared')) {
      response.writeHead(403).end();
      return;
    }
    const match = /^bytes=(\d+)-(\d*)$/.exec(request.headers.range ?? '');
    const start = match ? Number(match[1]) : 0;
    const end = match?.[2] ? Math.min(Number(match[2]), data.length - 1) : data.length - 1;
    requests.push({ url: pathname, start, range: Boolean(match), branch: request.headers['x-branch'] });
    response.writeHead(match ? 206 : 200, {
      'content-type': isLive ? 'video/mp2t' : 'video/mp4',
      'content-length': end - start + 1,
      'accept-ranges': 'bytes',
      ...(match ? { 'content-range': `bytes ${start}-${end}/${data.length}` } : {}),
    });
    let position = start;
    const timer = setInterval(() => {
      if (response.writableNeedDrain) return;
      const next = Math.min(position + 64 * 1024, end + 1);
      response.write(data.subarray(position, next));
      position = next;
      if (position > end) response.end();
    }, isLive ? 10 : 50);
    response.on('close', () => clearInterval(timer));
  });
  await new Promise(resolve => upstream.listen(0, '127.0.0.1', resolve));
  const sourceBase = `http://127.0.0.1:${upstream.address().port}`;
  let ytDlp = 'yt-dlp';
  if (resolver) {
    ytDlp = `${temp}/yt-dlp`;
    let payload = {
      duration: 8, http_headers: { 'X-Common': 'shared' }, ...resolverMetadata,
      ...(['combined', 'unknown-codecs', 'protected-combined'].includes(resolver) ? {
        url: `${sourceBase}/${resolver === 'protected-combined' ? 'iptv.mp4' : 'combined.mp4'}?token=combined-secret`,
        ...(resolver === 'unknown-codecs' ? {} : { vcodec: 'mpeg4', acodec: 'aac' }),
      } : { requested_formats: [
        ...(resolver === 'missing-video' ? [] : [
          { url: `${sourceBase}/video.mp4?token=video-secret`, vcodec: 'mpeg4', acodec: 'none', http_headers: { 'X-Branch': 'video' } },
        ]),
        ...(resolver === 'missing-audio' ? [] : [
          { url: `${sourceBase}/audio.m4a?token=audio-secret`, vcodec: 'none', acodec: 'aac', http_headers: { 'X-Branch': 'audio' } },
        ]),
      ] }),
    };
    if (resolver === 'playlist' || resolver === 'empty-playlist') {
      payload = { _type: 'playlist', duration: 999, entries: resolver === 'playlist' ? [payload] : [] };
    }
    await fs.writeFile(ytDlp, `#!/usr/bin/env node
require('node:fs').writeFileSync(${JSON.stringify(`${temp}/yt-dlp-args.json`)}, JSON.stringify(process.argv.slice(2)));
if (${JSON.stringify(resolver)} === 'fail') {
  process.stderr.write('Extraction failed for https://example.test/video?token=resolver-secret');
  process.exit(1);
}
process.stdout.write(${JSON.stringify(JSON.stringify(payload))});
`, { mode: 0o755 });
  }
  const reservation = createServer();
  await new Promise(resolve => reservation.listen(0, '127.0.0.1', resolve));
  const port = reservation.address().port;
  await new Promise(resolve => reservation.close(resolve));
  let ffprobe = 'ffprobe';
  if (probe !== 'real') {
    ffprobe = `${temp}/ffprobe`;
    await fs.writeFile(ffprobe, `#!/usr/bin/env node
const fs = require('node:fs');
const counter = ${JSON.stringify(`${temp}/attempts`)};
const attempt = Number(fs.existsSync(counter) ? fs.readFileSync(counter, 'utf8') : 0) + 1;
fs.writeFileSync(counter, String(attempt));
if (${JSON.stringify(probe)} === 'fail') {
  process.stderr.write('Temporary upstream failure'); process.exit(1);
} else if (attempt === 1) {
  setInterval(() => {}, 1000);
} else {
  const child = require('node:child_process').spawn('ffprobe', process.argv.slice(2), {stdio: 'inherit'});
  process.on('SIGTERM', () => child.kill('SIGTERM'));
  child.on('exit', code => process.exit(code ?? 1));
}
`, { mode: 0o755 });
  }
  const child = spawn(process.execPath, ['server/index.js'], {
    cwd: root, detached: true,
    env: { ...process.env, PORT: String(port), PLAYBACK_CONNECTION_MODE: mode,
      METADATA_PROBE_ENABLED: metadataEnabled, METADATA_PROBE_TIMEOUT_MS: '1000', FFPROBE_PATH: ffprobe,
      SINGLE_CONNECTION_HOSTS: singleConnectionHosts, YT_DLP_PATH: ytDlp, YT_DLP_FORMAT: resolverFormat,
      FFMPEG_INPUT_SEEKABLE: '0', FFMPEG_HTTP_RECONNECT: '0', LOCAL_VIDEOS: '',
      RECENT_URLS_PATH: `${temp}/recent.json`, FAVORITES_PATH: `${temp}/favorites.json` },
    stdio: ['ignore', 'pipe', 'pipe'],
  });
  let logs = '';
  child.stdout.on('data', chunk => logs += chunk);
  child.stderr.on('data', chunk => logs += chunk);
  t.after(async () => {
    try { process.kill(-child.pid, 'SIGTERM'); } catch {}
    if (child.exitCode === null && child.signalCode === null) await once(child, 'close');
    // Include any worker still shutting down in this isolated process group.
    try { process.kill(-child.pid, 'SIGKILL'); } catch {}
    upstream.closeAllConnections();
    await new Promise(resolve => upstream.close(resolve));
  });
  const base = `http://127.0.0.1:${port}`;
  await until(async () => {
    try { return (await fetch(`${base}/api/health`)).ok; } catch { return false; }
  });
  return {
    base, requests, logs: () => logs,
    attempts: async () => Number(await fs.readFile(`${temp}/attempts`, 'utf8')),
    ytDlpArgs: async () => JSON.parse(await fs.readFile(`${temp}/yt-dlp-args.json`, 'utf8')),
    create: async (pathname = '/movie.mp4') => fetch(`${base}/api/session`, {
      method: 'POST', headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ url: pathname.startsWith('http') ? pathname : `${sourceBase}${pathname}`, fps: 12, width: 160 }),
    }).then(response => response.json()),
    metadata: id => fetch(`${base}/api/session/${id}`).then(response => response.json()),
  };
}

function connect(t, base, id) {
  const abort = new AbortController();
  const socket = new WebSocket(`${base.replace('http:', 'ws:')}/frames/${id}`);
  const close = () => { abort.abort(); socket.terminate(); };
  t.after(close);
  const timer = setTimeout(close, 10000);
  const frame = new Promise((resolve, reject) => {
    socket.on('message', (data, binary) => { if (binary) resolve(data); });
    socket.on('error', reject);
    socket.on('close', () => reject(new Error('Frames closed before first image')));
  });
  const audio = new Promise((resolve, reject) => {
    void (async () => {
      const response = await fetch(`${base}/audio/${id}`, { signal: abort.signal });
      assert.equal(response.status, 200);
      let bytes = 0;
      for await (const chunk of response.body) {
        bytes += chunk.length;
        if (bytes >= 8192) resolve(bytes);
      }
      reject(new Error('Audio ended before enough data arrived'));
    })().catch(reject);
  });
  return {
    ready: Promise.all([frame, audio]).finally(() => clearTimeout(timer)),
    close,
  };
}

test('recorded MP4 starts with random access and recovers seeking after a probe timeout', { timeout: 20000 }, async t => {
  const app = await startApp(t, { probe: 'timeout-once' });
  const session = await app.create();
  assert.equal(session.metadataStatus, 'pending');
  const streams = connect(t, app.base, session.id);
  const [firstFrame] = await streams.ready;
  assert.equal(firstFrame.readDoubleLE(0), 0);
  assert.equal(firstFrame.readUInt16BE(8), 0xffd8);
  assert.match(app.logs(), /source=direct mode=split metadata=pending/);
  assert.ok(app.requests.some(request => request.start > 16 * 1024 * 1024), 'ffmpeg must skip to metadata at EOF');
  const metadata = await until(async () => {
    const value = await app.metadata(session.id);
    return value.metadataStatus === 'ready' && value;
  });
  assert.equal(metadata.seekable, true);
  assert.ok(metadata.duration >= 8 && metadata.duration < 9);
  assert.equal(await app.attempts(), 2);
  assert.match(app.logs(), /duration probe retry .*ffprobe timed out/);
  streams.close();
  await until(() => (app.logs().match(/ffmpeg exited/g) ?? []).length === 2);
  const response = await fetch(`${app.base}/api/session/${session.id}/seek`, {
    method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ time: 4 }),
  });
  assert.equal(response.status, 200);
  const sought = await response.json();
  assert.equal(sought.seekSeconds, 4);
  assert.equal(sought.seekGeneration, 1);
  const resumed = connect(t, app.base, session.id);
  const [nextFrame] = await resumed.ready;
  assert.notDeepEqual(nextFrame.subarray(8), firstFrame.subarray(8), 'seek must change the video content');
  resumed.close();
  await until(() => (app.logs().match(/ffmpeg exited/g) ?? []).length === 4);
});

test('failed metadata retries are bounded and finish as unavailable', { timeout: 12000 }, async t => {
  const app = await startApp(t, { probe: 'fail' });
  const session = await app.create();
  await until(async () => (await app.metadata(session.id)).metadataStatus === 'unavailable');
  assert.equal(await app.attempts(), 3);
  for (let i = 0; i < 3; i++) assert.equal((await app.metadata(session.id)).seekable, false);
  assert.equal(await app.attempts(), 3);
});

test('live relay still uses exactly one upstream connection and no metadata probe', { timeout: 15000 }, async t => {
  const app = await startApp(t, { probe: 'fail' });
  const session = await app.create('/live');
  assert.equal(session.metadataStatus, 'disabled');
  const streams = connect(t, app.base, session.id);
  await streams.ready;
  assert.equal(app.requests.length, 1);
  assert.equal(app.requests[0].range, false);
  const recovery = await fetch(`${app.base}/api/session/${session.id}/seek`, {
    method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ time: 5, allowBestEffort: true }),
  });
  assert.equal(recovery.status, 409);
  assert.equal(app.requests.length, 1);
  await assert.rejects(app.attempts(), { code: 'ENOENT' });
  streams.close();
  await until(() => app.logs().includes('relay closed'));
  assert.equal((app.logs().match(/ffmpeg exited/g) ?? []).length, 2);
});

test('subscription hosts stay on one connection even for MP4 URLs with global split and probing enabled', { timeout: 15000 }, async t => {
  const app = await startApp(t, { mode: 'split', probe: 'fail', singleConnectionHosts: '127.0.0.1', metadataEnabled: '1' });
  const session = await app.create('/iptv.mp4');
  assert.equal(session.metadataStatus, 'disabled');
  assert.equal(session.seekable, false);
  const streams = connect(t, app.base, session.id);
  await streams.ready;
  assert.match(app.logs(), /source=direct mode=relay metadata=disabled/);
  for (const allowBestEffort of [false, true]) {
    const result = await fetch(`${app.base}/api/session/${session.id}/seek`, {
      method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ time: 5, allowBestEffort }),
    });
    assert.equal(result.status, 409);
  }
  assert.equal(app.requests.length, 1);
  assert.equal(app.requests[0].range, false);
  await assert.rejects(app.attempts(), { code: 'ENOENT' });
  streams.close();
  await until(() => app.logs().includes('relay closed'));
});

for (const { label, url, kind, resolver, mode = 'single' } of [
  ...['separate', 'combined'].map(resolver => ({ label: 'YouTube', url: 'https://www.youtube.com/watch?v=test', kind: 'youtube', resolver })),
  ...['separate', 'combined', 'unknown-codecs', 'playlist'].map(resolver => ({ label: 'X', url: 'https://x.com/example/status/123?s=20', kind: 'twitter', resolver, mode: 'relay' })),
  { label: 'Twitter', url: 'https://twitter.com/example/status/123/video/2', kind: 'twitter', resolver: 'combined' },
]) {
  test(`${label} ${resolver} sources play and seek with Node enabled and no extra probe`, { timeout: 15000 }, async t => {
    const app = await startApp(t, { mode, probe: 'fail', resolver });
    const session = await app.create(url);
    assert.equal(session.duration, 8);
    assert.equal(session.metadataStatus, 'ready');
    assert.equal(session.seekable, true);
    const args = await app.ytDlpArgs();
    assert.equal(args[args.indexOf('--js-runtimes') + 1], `node:${process.execPath}`);
    assert.match(args[args.indexOf('--format') + 1], /bestvideo.*\+bestaudio/);
    if (kind === 'twitter') {
      assert.equal(args[args.indexOf('--format') + 1], 'best[protocol=https]/bestvideo[protocol=https]+bestaudio[protocol=https]');
    }
    assert.equal(args.at(-1), url, 'preserve the selected video and query parameters');
    assert.ok(args.includes('--no-playlist'));
    if (kind === 'twitter') assert.equal(args[args.indexOf('--playlist-items') + 1], '1');
    const recents = await fetch(`${app.base}/api/recent-urls`).then(response => response.json());
    assert.equal(recents.urls[0].url, url);
    const streams = connect(t, app.base, session.id);
    const [frame] = await streams.ready;
    assert.ok(Number.isFinite(frame.readDoubleLE(0)));
    assert.equal(frame.readUInt16BE(8), 0xffd8, 'frames must remain JPEGs');
    assert.equal(frame.readUInt16BE(frame.length - 2), 0xffd9);
    streams.close();
    await until(() => (app.logs().match(/ffmpeg exited/g) ?? []).length === 2);
    const result = await fetch(`${app.base}/api/session/${session.id}/seek`, {
      method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ time: 4 }),
    });
    assert.equal(result.status, 200);
    assert.equal((await result.json()).seekSeconds, 4);
    const resumed = connect(t, app.base, session.id);
    await resumed.ready;
    resumed.close();
    await until(() => (app.logs().match(/ffmpeg exited/g) ?? []).length === 4);
    if (resolver === 'separate' || resolver === 'playlist') {
      assert.ok(app.requests.some(request => request.url === '/video.mp4' && request.branch === 'video'));
      assert.ok(app.requests.some(request => request.url === '/audio.m4a' && request.branch === 'audio'));
    } else {
      assert.ok(app.requests.every(request => request.url === '/combined.mp4'));
    }
    assert.ok(app.logs().includes(`source=${kind} mode=split metadata=ready`));
    assert.doesNotMatch(app.logs(), /(?:video|audio|combined)-secret/);
    await assert.rejects(app.attempts(), { code: 'ENOENT' });
  });
}

for (const host of ['youtu.be', 'x.com', 'twitter.com']) {
  test(`${host} resolution failures are logged without exposing source URLs`, async t => {
    const app = await startApp(t, { resolver: 'fail' });
    const session = await app.create(`https://${host}/example/status/123`);
    assert.match(session.error, /yt-dlp failed to resolve/);
    assert.ok(app.logs().includes(`yt-dlp resolve failed host=${host}`));
    assert.doesNotMatch(app.logs(), /resolver-secret|example\.test/);
    if (host !== 'youtu.be') assert.match(session.error, /X\/Twitter URL/);
  });
}

test('audio-less YouTube results are rejected before any playback worker starts', async t => {
  const app = await startApp(t, { resolver: 'missing-audio' });
  const session = await app.create('https://youtu.be/test');
  assert.match(session.error, /did not return playable video and audio streams/);
  assert.equal(app.requests.length, 0);
});

test('X/Twitter desktop and mobile URLs use the resolver, with hostname boundaries enforced', async t => {
  const app = await startApp(t, { resolver: 'combined', probe: 'fail' });
  for (const host of ['x.com', 'www.x.com', 'mobile.x.com', 'm.x.com', 'twitter.com', 'www.twitter.com', 'mobile.twitter.com', 'm.twitter.com']) {
    const url = `https://${host}/i/web/status/123`;
    assert.equal((await app.create(url)).metadataStatus, 'ready');
    assert.equal((await app.ytDlpArgs()).at(-1), url);
  }
  const lastArgs = await app.ytDlpArgs();
  for (const host of ['notx.com', 'nottwitter.com', 'x.com.example.test', 'twitter.com.example.test']) {
    assert.equal((await app.create(`https://${host}/example/status/123`)).metadataStatus, 'disabled');
    assert.deepEqual(await app.ytDlpArgs(), lastArgs, 'unrelated hosts must not invoke yt-dlp');
  }
  await assert.rejects(app.attempts(), { code: 'ENOENT' });
});

test('X preserves an explicitly configured yt-dlp format selector', async t => {
  const resolverFormat = 'best[protocol=https][height<=720]';
  const app = await startApp(t, { resolver: 'combined', resolverFormat });
  assert.equal((await app.create('https://x.com/example/status/123')).metadataStatus, 'ready');
  const args = await app.ytDlpArgs();
  assert.equal(args[args.indexOf('--format') + 1], resolverFormat);
});

for (const resolver of ['missing-audio', 'missing-video', 'empty-playlist']) {
  test(`X ${resolver} results are rejected before playback starts`, async t => {
    const app = await startApp(t, { resolver });
    const session = await app.create('https://x.com/example/status/123');
    assert.match(session.error, /did not return playable video and audio streams/);
    assert.equal(app.requests.length, 0);
  });
}

for (const resolverMetadata of [{ duration: null }, { duration: 0 }, { is_live: true }, { live_status: 'is_live' }]) {
  test(`X metadata ${JSON.stringify(resolverMetadata)} does not enable seeking`, async t => {
    const app = await startApp(t, { resolver: 'combined', resolverMetadata, probe: 'fail', mode: 'single' });
    const session = await app.create('https://x.com/example/status/123');
    assert.equal(session.duration, null);
    assert.equal(session.seekable, false);
    assert.equal(session.metadataStatus, 'pending');
    assert.match(app.logs(), /source=twitter mode=split metadata=pending/);
  });
}

test('protected page hosts bypass yt-dlp, probing, and recovery seeking', async t => {
  const app = await startApp(t, { resolver: 'fail', probe: 'fail', mode: 'split', metadataEnabled: '1', singleConnectionHosts: 'X.COM, twitter.com, youtube.com' });
  for (const host of ['x.com', 'mobile.x.com', 'twitter.com', 'www.youtube.com']) {
    const session = await app.create(`https://${host}/example/status/123`);
    assert.equal(session.metadataStatus, 'disabled');
    assert.equal(session.seekable, false);
    const recovery = await fetch(`${app.base}/api/session/${session.id}/seek`, {
      method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ time: 5, allowBestEffort: true }),
    });
    assert.equal(recovery.status, 409);
  }
  assert.match(app.logs(), /source=direct mode=relay metadata=disabled/);
  await assert.rejects(app.ytDlpArgs(), { code: 'ENOENT' });
  await assert.rejects(app.attempts(), { code: 'ENOENT' });
  assert.equal(app.requests.length, 0);
});

test('resolved combined sources on protected hosts use one relay connection', { timeout: 15000 }, async t => {
  const app = await startApp(t, { resolver: 'protected-combined', probe: 'fail', mode: 'split', metadataEnabled: '1', singleConnectionHosts: '127.0.0.1' });
  const session = await app.create('https://x.com/example/status/123');
  assert.equal(session.metadataStatus, 'disabled');
  assert.equal(session.duration, null);
  assert.equal(session.seekable, false);
  const streams = connect(t, app.base, session.id);
  await streams.ready;
  assert.equal(app.requests.length, 1);
  assert.equal(app.requests[0].range, false);
  assert.match(app.logs(), /source=twitter mode=relay metadata=disabled/);
  await assert.rejects(app.attempts(), { code: 'ENOENT' });
  streams.close();
  await until(() => app.logs().includes('relay closed'));
});

test('resolved separate sources cannot bypass single-connection hosts', async t => {
  const app = await startApp(t, { resolver: 'separate', singleConnectionHosts: '127.0.0.1' });
  const session = await app.create('https://x.com/example/status/123');
  assert.match(session.error, /separate connections to a single-connection host/);
  assert.equal(app.requests.length, 0);
});