every.channel/crates/ec-node/tests/e2e_remote_website_watch_existing.rs

use std::ffi::OsStr;
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};

fn which(cmd: &str) -> Option<std::path::PathBuf> {
    which::which(cmd).ok()
}

fn chrome_path() -> Option<std::path::PathBuf> {
    let mac =
        std::path::PathBuf::from("/Applications/Google Chrome.app/Contents/MacOS/Google Chrome");
    if mac.exists() {
        return Some(mac);
    }
    which("google-chrome")
        .or_else(|| which("google-chrome-stable"))
        .or_else(|| which("chromium"))
}

fn ec_node_path() -> std::path::PathBuf {
    if let Ok(value) = std::env::var("EC_NODE_BIN") {
        return value.into();
    }
    if let Ok(value) = std::env::var("CARGO_BIN_EXE_ec_node") {
        return value.into();
    }
    if let Ok(value) = std::env::var("CARGO_BIN_EXE_ec-node") {
        return value.into();
    }
    let exe = std::env::current_exe().expect("current_exe");
    let debug_dir = exe
        .parent()
        .and_then(|p| p.parent())
        .expect("expected target/debug/deps");
    debug_dir.join("ec-node")
}

fn wait_for_canvas_element(tab: &headless_chrome::Tab, timeout: Duration) -> anyhow::Result<()> {
    let deadline = Instant::now() + timeout;
    while Instant::now() < deadline {
        let js = r#"(function() {
  return !!document.querySelector('moq-watch canvas');
})();"#;
        let v = tab.evaluate(js, false)?;
        if v.value.and_then(|v| v.as_bool()).unwrap_or(false) {
            return Ok(());
        }
        std::thread::sleep(Duration::from_millis(200));
    }
    anyhow::bail!("timed out waiting for canvas player");
}

fn wait_for_moq_watch_element(tab: &headless_chrome::Tab, timeout: Duration) -> anyhow::Result<()> {
    let deadline = Instant::now() + timeout;
    while Instant::now() < deadline {
        let js = r#"(function() {
  return !!document.querySelector('moq-watch');
})();"#;
        let v = tab.evaluate(js, false)?;
        if v.value.and_then(|v| v.as_bool()).unwrap_or(false) {
            return Ok(());
        }
        std::thread::sleep(Duration::from_millis(200));
    }
    anyhow::bail!("timed out waiting for <moq-watch> element");
}

fn wait_for_live_or_archive_player(
    tab: &headless_chrome::Tab,
    timeout: Duration,
) -> anyhow::Result<()> {
    let deadline = Instant::now() + timeout;
    while Instant::now() < deadline {
        let js = r#"(function() {
  return !!document.querySelector('moq-watch, video.archiveVideo');
})();"#;
        let v = tab.evaluate(js, false)?;
        if v.value.and_then(|v| v.as_bool()).unwrap_or(false) {
            return Ok(());
        }
        std::thread::sleep(Duration::from_millis(200));
    }
    anyhow::bail!("timed out waiting for live or archive player");
}

fn debug_player_state(tab: &headless_chrome::Tab) -> anyhow::Result<String> {
    let js = r#"(function() {
  let watch = document.querySelector('moq-watch');
  let canvas = document.querySelector('moq-watch canvas');
  let video = document.querySelector('video.archiveVideo');
  let placeholder = document.querySelector('.placeholder');
  let placeholderText = placeholder ? (placeholder.innerText || '') : null;
  let status = document.querySelector('.source-status');
  let statusText = status ? (status.innerText || '') : null;
  let statusLine = document.querySelector('#statusLine');
  let statusLineText = statusLine ? (statusLine.innerText || '') : null;
  let catalog = watch && watch.broadcast && watch.broadcast.catalog && watch.broadcast.catalog.peek
    ? watch.broadcast.catalog.peek()
    : null;
  let established = watch && watch.connection && watch.connection.established && watch.connection.established.peek
    ? watch.connection.established.peek()
    : null;
  let sources = Array.from(document.querySelectorAll('button[data-testid="global-watch"]')).length;
  let hint = document.querySelector('#hint');
  let hintText = hint ? (hint.innerText || '') : null;
  return JSON.stringify({
    hasWatch: !!watch,
    hasCanvas: !!canvas,
    canvasWidth: canvas ? canvas.width : null,
    canvasHeight: canvas ? canvas.height : null,
    hasArchiveVideo: !!video,
    videoCurrentTime: video ? video.currentTime : null,
    videoDuration: video ? video.duration : null,
    videoPaused: video ? video.paused : null,
    videoReadyState: video ? video.readyState : null,
    videoMuted: video ? video.muted : null,
    videoVolume: video ? video.volume : null,
    videoSrc: video ? (video.currentSrc || video.src || '') : null,
    muted: watch ? watch.muted : null,
    volume: watch ? watch.volume : null,
    connectionStatus: watch?.connection?.status?.peek ? watch.connection.status.peek() : null,
    connectionKind: established ? established.constructor?.name || null : null,
    broadcastStatus: watch?.broadcast?.status?.peek ? watch.broadcast.status.peek() : null,
    paused: watch?.backend?.paused?.peek ? watch.backend.paused.peek() : null,
    audioMuted: watch?.backend?.audio?.muted?.peek ? watch.backend.audio.muted.peek() : null,
    audioVolume: watch?.backend?.audio?.volume?.peek ? watch.backend.audio.volume.peek() : null,
    catalogSeen: !!catalog,
    catalogHasVideo: !!(catalog?.video?.renditions),
    catalogHasAudio: !!(catalog?.audio?.renditions),
    metrics: window.__ecPlaybackMetrics || null,
    statusLineText,
    hintText,
    placeholderText,
    statusText,
    sources
  });
})();"#;
    let v = tab.evaluate(js, false)?;
    Ok(v.value
        .and_then(|v| v.as_str().map(|s| s.to_string()))
        .unwrap_or_default())
}

fn canvas_motion_sample(tab: &headless_chrome::Tab) -> anyhow::Result<Option<(f64, u32)>> {
    let js = r#"(function() {
  let source = document.querySelector('moq-watch canvas');
  if (!source || !source.width || !source.height) return null;
  let canvas = window.__ec_motion_canvas || (window.__ec_motion_canvas = document.createElement('canvas'));
  canvas.width = 160;
  canvas.height = 90;
  let ctx = canvas.getContext('2d', { willReadFrequently: true });
  ctx.drawImage(source, 0, 0, canvas.width, canvas.height);
  let data = ctx.getImageData(0, 0, canvas.width, canvas.height).data;
  let hash = 2166136261 >>> 0;
  for (let i = 0; i < data.length; i += 16) {
    hash ^= data[i]; hash = Math.imul(hash, 16777619);
    hash ^= data[i + 1]; hash = Math.imul(hash, 16777619);
    hash ^= data[i + 2]; hash = Math.imul(hash, 16777619);
  }
  return JSON.stringify({ currentTime: performance.now() / 1000, hash: hash >>> 0 });
})();"#;
    let v = tab.evaluate(&js, false)?;
    let Some(s) = v.value.and_then(|v| v.as_str().map(|s| s.to_string())) else {
        return Ok(None);
    };
    let value: serde_json::Value = serde_json::from_str(&s)?;
    let current_time = value
        .get("currentTime")
        .and_then(|v| v.as_f64())
        .unwrap_or_default();
    let hash = value
        .get("hash")
        .and_then(|v| v.as_u64())
        .unwrap_or_default() as u32;
    Ok(Some((current_time, hash)))
}

fn archive_video_motion_sample(
    tab: &headless_chrome::Tab,
) -> anyhow::Result<Option<serde_json::Value>> {
    let js = r#"(function() {
  let video = document.querySelector('video.archiveVideo');
  if (!video) return null;
  if (video.paused) video.play().catch(() => {});
  return JSON.stringify({
    wallTime: performance.now() / 1000,
    currentTime: video.currentTime || 0,
    readyState: video.readyState || 0,
    paused: !!video.paused,
    ended: !!video.ended,
    muted: !!video.muted,
    volume: video.volume || 0,
    src: video.currentSrc || video.src || ''
  });
})();"#;
    let v = tab.evaluate(js, false)?;
    let Some(s) = v.value.and_then(|v| v.as_str().map(|s| s.to_string())) else {
        return Ok(None);
    };
    Ok(Some(serde_json::from_str(&s)?))
}

fn wait_for_canvas_or_archive_motion(
    tab: &headless_chrome::Tab,
    timeout: Duration,
) -> anyhow::Result<String> {
    let deadline = Instant::now() + timeout;
    let mut first_canvas: Option<(f64, u32)> = None;
    let mut first_video_time: Option<f64> = None;
    while Instant::now() < deadline {
        if let Some(sample) = canvas_motion_sample(tab)? {
            if let Some((first_time, first_hash)) = first_canvas {
                if sample.0 > first_time + 0.5 && sample.1 != first_hash {
                    return Ok("moq-canvas".to_string());
                }
            } else {
                first_canvas = Some(sample);
            }
        }

        if let Some(sample) = archive_video_motion_sample(tab)? {
            let current_time = sample
                .get("currentTime")
                .and_then(|v| v.as_f64())
                .unwrap_or_default();
            let ready_state = sample
                .get("readyState")
                .and_then(|v| v.as_u64())
                .unwrap_or_default();
            let ended = sample
                .get("ended")
                .and_then(|v| v.as_bool())
                .unwrap_or(false);
            if ready_state >= 2 && !ended {
                if let Some(first) = first_video_time {
                    if current_time > first + 0.5 {
                        return Ok("archive-video".to_string());
                    }
                } else {
                    first_video_time = Some(current_time);
                }
            }
        }

        std::thread::sleep(Duration::from_millis(500));
    }
    let st = debug_player_state(tab).unwrap_or_default();
    anyhow::bail!("timed out waiting for live or archive motion\nplayer_state={st}");
}

fn wait_for_playback_probe_ok(
    tab: &headless_chrome::Tab,
    timeout: Duration,
) -> anyhow::Result<String> {
    let deadline = Instant::now() + timeout;
    let mut last_metrics = String::new();
    while Instant::now() < deadline {
        let js = r#"(function() {
  const metrics = window.__ecPlaybackMetrics || null;
  return metrics ? JSON.stringify(metrics) : "";
})();"#;
        let v = tab.evaluate(js, false)?;
        last_metrics = v
            .value
            .and_then(|v| v.as_str().map(|s| s.to_string()))
            .unwrap_or_default();
        if !last_metrics.is_empty() {
            let metrics: serde_json::Value = serde_json::from_str(&last_metrics)?;
            let ok = metrics.get("ok").and_then(|v| v.as_bool()).unwrap_or(false);
            let samples = metrics
                .get("samples")
                .and_then(|v| v.as_u64())
                .unwrap_or_default();
            let changed = metrics
                .get("changed_samples")
                .and_then(|v| v.as_u64())
                .unwrap_or_default();
            let longest_static = metrics
                .get("longest_same_hash_ms")
                .and_then(|v| v.as_u64())
                .unwrap_or_default();
            if ok && samples >= 8 && changed >= 2 && longest_static < 5_000 {
                return Ok(last_metrics);
            }
        }
        std::thread::sleep(Duration::from_millis(250));
    }
    let st = debug_player_state(tab).unwrap_or_default();
    anyhow::bail!(
        "timed out waiting for playback probe ok\nplayer_state={st}\nmetrics={last_metrics}"
    );
}

fn wait_for_unmuted_player(tab: &headless_chrome::Tab, timeout: Duration) -> anyhow::Result<()> {
    let deadline = Instant::now() + timeout;
    while Instant::now() < deadline {
        let js = r#"(function() {
  let watch = document.querySelector('moq-watch');
  let video = document.querySelector('video.archiveVideo');
  return (!!watch && watch.muted === false && watch.volume > 0 && !watch.hasAttribute('muted')) ||
    (!!video && video.muted === false && video.volume > 0);
})();"#;
        let v = tab.evaluate(js, false)?;
        if v.value.and_then(|v| v.as_bool()).unwrap_or(false) {
            return Ok(());
        }
        std::thread::sleep(Duration::from_millis(200));
    }
    let st = debug_player_state(tab).unwrap_or_default();
    anyhow::bail!("timed out waiting for unmuted player\nplayer_state={st}");
}

fn watch_url(
    site_url: &str,
    relay_url: &str,
    stream_id: &str,
    verify: bool,
) -> anyhow::Result<String> {
    let mut url = url::Url::parse(site_url)?;
    url.set_path("/watch");
    url.query_pairs_mut()
        .clear()
        .append_pair("url", relay_url)
        .append_pair("name", stream_id);
    if verify {
        url.query_pairs_mut().append_pair("verify", "1");
    }
    Ok(url.to_string())
}

#[test]
#[ignore]
fn e2e_remote_website_watch_existing_stream_id() -> anyhow::Result<()> {
    let chrome = match chrome_path() {
        Some(p) => p,
        None => return Ok(()), // skip
    };
    // We still want ffmpeg around for parity with other E2Es (and to discourage "works only without media tools").
    if which("ffmpeg").is_none() {
        return Ok(()); // skip
    }

    let site_url = std::env::var("EVERY_CHANNEL_SITE_URL")
        .unwrap_or_else(|_| "https://every.channel/".to_string());
    let relay_url = std::env::var("EVERY_CHANNEL_RELAY_URL")
        .unwrap_or_else(|_| "https://relay.every.channel/anon".to_string());
    let stream_id = match std::env::var("EVERY_CHANNEL_STREAM_ID") {
        Ok(v) if !v.trim().is_empty() => v,
        _ => return Ok(()), // skip
    };

    let launch_options = headless_chrome::LaunchOptionsBuilder::default()
        .path(Some(chrome))
        .headless(true)
        .args(vec![
            OsStr::new("--autoplay-policy=no-user-gesture-required"),
            OsStr::new("--disable-application-cache"),
            OsStr::new("--disable-service-worker"),
            OsStr::new("--disk-cache-size=0"),
        ])
        .build()
        .unwrap();
    let browser = headless_chrome::Browser::new(launch_options)?;
    let tab = browser.new_tab()?;
    tab.navigate_to(&watch_url(&site_url, &relay_url, &stream_id, false)?)?;
    tab.wait_until_navigated()?;

    // Ensure either the native MoQ player or the archive live-edge fallback is instantiated.
    if let Err(err) = wait_for_live_or_archive_player(&tab, Duration::from_secs(90)) {
        let st = debug_player_state(&tab).unwrap_or_default();
        anyhow::bail!("{err}\nplayer_state={st}");
    }

    tab.evaluate(
        r#"(function() {
  const canvas = document.querySelector('moq-watch canvas');
  if (canvas) canvas.click();
  const audioButton = document.querySelector('#audioBtn');
  if (audioButton && audioButton.getAttribute('aria-pressed') !== 'true') {
    audioButton.click();
  }
})();"#,
        false,
    )?;
    wait_for_unmuted_player(&tab, Duration::from_secs(10))?;
    let playback_path = wait_for_canvas_or_archive_motion(&tab, Duration::from_secs(60))?;
    eprintln!("playback path: {playback_path}");

    Ok(())
}

#[test]
#[ignore]
fn e2e_remote_website_watch_synthetic_relay_stream() -> anyhow::Result<()> {
    if which("ffmpeg").is_none() {
        return Ok(()); // skip
    }
    let chrome = match chrome_path() {
        Some(p) => p,
        None => return Ok(()), // skip
    };

    let site_url = std::env::var("EVERY_CHANNEL_SITE_URL")
        .unwrap_or_else(|_| "https://every.channel/".to_string());
    let relay_url = std::env::var("EVERY_CHANNEL_RELAY_URL")
        .unwrap_or_else(|_| "https://relay.every.channel/anon".to_string());
    let tls_disable_verify = std::env::var("EVERY_CHANNEL_RELAY_TLS_DISABLE_VERIFY")
        .map(|v| v != "0" && v.to_lowercase() != "false")
        .unwrap_or(true);

    let ts = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis();
    let stream_id = format!("e2e-synthetic-{ts}");
    let ec_node = ec_node_path();

    let mut publisher = Command::new(&ec_node);
    publisher
        .arg("wt-publish")
        .arg("--url")
        .arg(&relay_url)
        .arg("--name")
        .arg(&stream_id)
        .arg("--realtime-input")
        .arg("--input-format")
        .arg("lavfi")
        .arg("--input")
        .arg("testsrc2=size=1280x720:rate=30")
        .stdout(Stdio::null())
        .stderr(Stdio::inherit());
    if tls_disable_verify {
        publisher.arg("--tls-disable-verify");
    }
    let mut publisher = publisher.spawn()?;

    let test_result = (|| -> anyhow::Result<()> {
        let launch_options = headless_chrome::LaunchOptionsBuilder::default()
            .path(Some(chrome))
            .headless(true)
            .args(vec![
                OsStr::new("--autoplay-policy=no-user-gesture-required"),
                OsStr::new("--disable-application-cache"),
                OsStr::new("--disable-service-worker"),
                OsStr::new("--disk-cache-size=0"),
                OsStr::new("--mute-audio"),
            ])
            .build()
            .unwrap();
        let browser = headless_chrome::Browser::new(launch_options)?;
        let tab = browser.new_tab()?;
        tab.navigate_to(&watch_url(&site_url, &relay_url, &stream_id, true)?)?;
        tab.wait_until_navigated()?;

        wait_for_moq_watch_element(&tab, Duration::from_secs(90))?;
        wait_for_canvas_element(&tab, Duration::from_secs(90))?;
        let metrics = wait_for_playback_probe_ok(&tab, Duration::from_secs(60))?;
        eprintln!("playback metrics: {metrics}");
        Ok(())
    })();

    let _ = publisher.kill();
    let _ = publisher.wait();
    test_result
}