moq-hls 0.1.0

//! HLS import: pull an HLS master/media playlist and publish it into MoQ.
//!
//! Watches an HLS master or media playlist, downloads each fMP4 segment as it
//! appears, and feeds it through moq-mux's fMP4 importer (which publishes a
//! `hang` broadcast + catalog). Classic HLS only for now (no LL-HLS partial
//! segments on the import side).

use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::path::PathBuf;
use std::time::Duration;

use bytes::Bytes;
use m3u8_rs::{
	AlternativeMedia, AlternativeMediaType, Map, MasterPlaylist, MediaPlaylist, MediaSegment, Resolution, VariantStream,
};
use moq_mux::catalog::Producer as CatalogProducer;
use moq_mux::container::fmp4::Import as Fmp4;
use moq_mux::select;
use reqwest::Client;
use tracing::{debug, info, warn};
use url::Url;

use crate::{Error, Result};

/// Per-request timeout for the default HTTP client (playlist + segment fetches).
const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);

/// Backoff before retrying after a failed import step, so a transient upstream
/// error (a 5xx, a truncated segment) doesn't tear down the whole import.
const ERROR_BACKOFF: Duration = Duration::from_secs(1);

/// Configuration for the single-rendition HLS import loop.
#[derive(Clone)]
pub struct Config {
	/// The master or media playlist URL or file path to import.
	pub playlist: String,

	/// An optional HTTP client to use for fetching the playlist and segments.
	/// If not provided, a default client will be created.
	pub client: Option<Client>,
}

impl Config {
	/// Create an import configuration for `playlist` using the default HTTP client.
	pub fn new(playlist: String) -> Self {
		Self { playlist, client: None }
	}

	/// Parse the playlist string into a URL.
	/// If it starts with http:// or https://, parse as URL.
	/// Otherwise, treat as a file path and convert to file:// URL.
	fn parse_playlist(&self) -> Result<Url> {
		if self.playlist.starts_with("http://") || self.playlist.starts_with("https://") {
			Url::parse(&self.playlist).map_err(|_| Error::InvalidPlaylistUrl)
		} else {
			let path = PathBuf::from(&self.playlist);
			let absolute = if path.is_absolute() {
				path
			} else {
				std::env::current_dir()?.join(path)
			};
			Url::from_file_path(&absolute).map_err(|_| Error::InvalidFilePath)
		}
	}
}

/// Result of a single import step.
struct StepOutcome {
	/// Number of media segments written during this step.
	pub wrote_segments: usize,
	/// Target segment duration (in seconds) from the playlist, if known.
	pub target_duration: Option<u64>,
}

/// HLS import that pulls an HLS media playlist and feeds the bytes into the fMP4 importer.
///
/// Provides `init()` to prime the importer with initial segments, and `run()`
/// to run the continuous import loop.
pub struct Import {
	/// Broadcast that all CMAF importers write into.
	broadcast: moq_net::BroadcastProducer,

	/// The catalog being produced.
	catalog: CatalogProducer,

	/// fMP4 importers for each discovered video rendition.
	/// Each importer feeds a separate MoQ track but shares the same catalog.
	video_importers: Vec<Fmp4>,

	/// fMP4 importer for the selected audio rendition, if any.
	audio_importer: Option<Fmp4>,

	client: Client,
	/// Parsed base URL for the playlist (file:// or http(s)://).
	base_url: Url,
	/// All discovered video variants (one per HLS rendition).
	video: Vec<TrackState>,
	/// Optional audio track shared across variants.
	audio: Option<TrackState>,
}

#[derive(Debug, Clone, Copy)]
enum TrackKind {
	Video(usize),
	Audio,
}

struct TrackState {
	playlist: Url,
	// Which roles this playlist's importer publishes (a muxed variant alongside a
	// separate audio rendition publishes video only).
	select: select::Broadcast,
	next_sequence: Option<u64>,
	next_discontinuity: Option<u64>,
	init_ready: bool,
}

impl TrackState {
	fn new(playlist: Url, select: select::Broadcast) -> Self {
		Self {
			playlist,
			select,
			next_sequence: None,
			next_discontinuity: None,
			init_ready: false,
		}
	}
}

/// Selection for a muxed rendition (the only source): publish every track.
fn select_muxed() -> select::Broadcast {
	select::Broadcast::default()
		.video(select::Video::default())
		.audio(select::Audio::default())
}

/// Selection for a video variant that has a separate audio rendition: video only.
fn select_video_only() -> select::Broadcast {
	select::Broadcast::default().video(select::Video::default())
}

/// Selection for a separate audio rendition: audio only.
fn select_audio_only() -> select::Broadcast {
	select::Broadcast::default().audio(select::Audio::default())
}

impl Import {
	/// Create a new HLS import that will write into the given broadcast.
	pub fn new(broadcast: moq_net::BroadcastProducer, catalog: CatalogProducer, cfg: Config) -> Result<Self> {
		let base_url = cfg.parse_playlist()?;
		let client = match cfg.client {
			Some(client) => client,
			None => Client::builder()
				.user_agent(concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION")))
				// Bound playlist/segment fetches so a stuck request can't wedge `run()`.
				.timeout(REQUEST_TIMEOUT)
				.build()?,
		};
		Ok(Self {
			broadcast,
			catalog,
			video_importers: Vec::new(),
			audio_importer: None,
			client,
			base_url,
			video: Vec::new(),
			audio: None,
		})
	}

	/// Fetch the latest playlist, download the init segment, and prime the importer with a buffer of segments.
	///
	/// Returns the number of segments buffered during initialization.
	pub async fn init(&mut self) -> Result<()> {
		let buffered = self.prime().await?;
		if buffered == 0 {
			warn!("HLS playlist had no new segments during init step");
		} else {
			info!(count = buffered, "buffered initial HLS segments");
		}
		Ok(())
	}

	/// Run the import loop until cancelled.
	///
	/// A failed step (e.g. a transient playlist fetch error) is logged and
	/// retried after a short backoff rather than ending the import.
	pub async fn run(&mut self) -> Result<()> {
		loop {
			let outcome = match self.step().await {
				Ok(outcome) => outcome,
				Err(err) => {
					warn!(%err, "HLS import step failed, retrying");
					tokio::time::sleep(ERROR_BACKOFF).await;
					continue;
				}
			};
			let delay = self.refresh_delay(outcome.target_duration, outcome.wrote_segments);

			info!(
				wrote_segments = outcome.wrote_segments,
				target_duration = ?outcome.target_duration,
				delay_secs = delay.as_secs_f32(),
				"HLS import step complete"
			);

			tokio::time::sleep(delay).await;
		}
	}

	/// Internal: fetch the latest playlist, download the init segment, and buffer segments.
	async fn prime(&mut self) -> Result<usize> {
		self.ensure_tracks().await?;

		let mut buffered = 0usize;
		const MAX_INIT_SEGMENTS: usize = 3; // Only process a few segments during init to avoid getting ahead of live stream

		// Prime all discovered video variants.
		//
		// Move the video track states out of `self` so we can safely mutate both
		// the importer and the tracks without running into borrow checker issues.
		let video_tracks = std::mem::take(&mut self.video);
		for (index, mut track) in video_tracks.into_iter().enumerate() {
			let playlist = self.fetch_media_playlist(track.playlist.clone()).await?;
			let count = self
				.consume_segments(TrackKind::Video(index), &mut track, &playlist, Some(MAX_INIT_SEGMENTS))
				.await?;
			buffered += count;
			self.video.push(track);
		}

		// Prime the shared audio track, if any.
		if let Some(mut track) = self.audio.take() {
			let playlist = self.fetch_media_playlist(track.playlist.clone()).await?;
			let count = self
				.consume_segments(TrackKind::Audio, &mut track, &playlist, Some(MAX_INIT_SEGMENTS))
				.await?;
			buffered += count;
			self.audio = Some(track);
		}

		Ok(buffered)
	}

	/// Perform a single import step for all active tracks.
	///
	/// This fetches the current media playlists, consumes any fresh segments,
	/// and returns how many segments were written along with the target
	/// duration to guide scheduling of the next step.
	async fn step(&mut self) -> Result<StepOutcome> {
		self.ensure_tracks().await?;

		let mut wrote = 0usize;
		let mut target_duration = None;

		// Ingest a step from all active video variants. A single variant failing is
		// logged and skipped (the track is always restored) so one bad rendition or
		// segment doesn't drop the others or abort the whole step.
		let video_tracks = std::mem::take(&mut self.video);
		for (index, mut track) in video_tracks.into_iter().enumerate() {
			match self
				.ingest(TrackKind::Video(index), &mut track, &mut target_duration)
				.await
			{
				Ok(count) => wrote += count,
				Err(err) => warn!(index, %err, "video rendition import step failed, will retry"),
			}
			self.video.push(track);
		}

		// Ingest from the shared audio track, if present.
		if let Some(mut track) = self.audio.take() {
			match self.ingest(TrackKind::Audio, &mut track, &mut target_duration).await {
				Ok(count) => wrote += count,
				Err(err) => warn!(%err, "audio rendition import step failed, will retry"),
			}
			self.audio = Some(track);
		}

		Ok(StepOutcome {
			wrote_segments: wrote,
			target_duration,
		})
	}

	/// Fetch one track's current media playlist and consume any fresh segments,
	/// recording the playlist's target duration if not already known.
	async fn ingest(
		&mut self,
		kind: TrackKind,
		track: &mut TrackState,
		target_duration: &mut Option<u64>,
	) -> Result<usize> {
		let playlist = self.fetch_media_playlist(track.playlist.clone()).await?;
		if target_duration.is_none() {
			*target_duration = Some(playlist.target_duration);
		}
		self.consume_segments(kind, track, &playlist, None).await
	}

	/// Compute the delay before the next import step should run.
	fn refresh_delay(&self, target_duration: Option<u64>, wrote_segments: usize) -> Duration {
		let base = target_duration
			.map(|dur| Duration::from_secs(dur.max(1)))
			.unwrap_or_else(|| Duration::from_millis(500));
		if wrote_segments == 0 {
			return base / 2;
		}

		base
	}

	async fn fetch_media_playlist(&self, url: Url) -> Result<MediaPlaylist> {
		let body = self.fetch_bytes(url).await?;

		// Nom errors take ownership of the input, so we need to stringify any error messages.
		let playlist = m3u8_rs::parse_media_playlist_res(&body).map_err(|e| Error::ParsePlaylist(e.to_string()))?;

		Ok(playlist)
	}

	async fn ensure_tracks(&mut self) -> Result<()> {
		// Tracks already discovered.
		if !self.video.is_empty() {
			return Ok(());
		}

		let body = self.fetch_bytes(self.base_url.clone()).await?;
		if let Ok((_, master)) = m3u8_rs::parse_master_playlist(&body) {
			let variants = select_variants(&master);
			if variants.is_empty() {
				return Err(Error::NoVariants);
			}

			// Choose an audio rendition first, so the video variants below know whether
			// they need to drop their muxed audio.
			if let Some(group_id) = variants.iter().find_map(|v| v.audio.as_deref()) {
				if let Some(audio_tag) = select_audio(&master, group_id) {
					if let Some(uri) = &audio_tag.uri {
						let audio_url = resolve_uri(&self.base_url, uri)?;
						self.audio = Some(TrackState::new(audio_url, select_audio_only()));
					} else {
						warn!(%group_id, "audio rendition missing URI");
					}
				} else {
					warn!(%group_id, "audio group not found in master playlist");
				}
			}

			// With a separate audio rendition, the variants are muxed but should publish
			// video only so the audio isn't duplicated; otherwise import every track.
			let variant_select = if self.audio.is_some() {
				select_video_only()
			} else {
				select_muxed()
			};
			for variant in &variants {
				let video_url = resolve_uri(&self.base_url, &variant.uri)?;
				self.video.push(TrackState::new(video_url, variant_select.clone()));
			}

			let audio_url = self.audio.as_ref().map(|a| a.playlist.to_string());
			info!(
				video_variants = variants.len(),
				audio = audio_url.as_deref().unwrap_or("none"),
				"selected master playlist renditions"
			);

			return Ok(());
		}

		// Fallback: treat the provided URL as a single (muxed) media playlist.
		self.video.push(TrackState::new(self.base_url.clone(), select_muxed()));
		Ok(())
	}

	async fn consume_segments(
		&mut self,
		kind: TrackKind,
		track: &mut TrackState,
		playlist: &MediaPlaylist,
		limit: Option<usize>,
	) -> Result<usize> {
		self.ensure_init_segment(kind, track, playlist).await?;

		let next_seq = track.next_sequence.unwrap_or(0);
		let playlist_seq = playlist.media_sequence;
		let total_segments = playlist.segments.len();
		let last_playlist_seq = playlist_seq + total_segments as u64;

		// Both out-of-window cases re-anchor to the current playlist (skip 0) and clear
		// `next_sequence` so the next push re-bases. The warning is suppressed on the
		// first step (`next_sequence` still None), where starting mid-window is normal.
		let skip = if next_seq > last_playlist_seq {
			if track.next_sequence.is_some() {
				warn!(
					?kind,
					next_sequence = next_seq,
					playlist_sequence = playlist_seq,
					last_playlist_sequence = last_playlist_seq,
					"imported ahead of playlist (upstream sequence reset?), re-anchoring to current window"
				);
			}
			track.next_sequence = None;
			0
		} else if next_seq < playlist_seq {
			if track.next_sequence.is_some() {
				warn!(
					?kind,
					next_sequence = next_seq,
					playlist_sequence = playlist_seq,
					"next_sequence behind playlist, resetting to start of playlist"
				);
			}
			track.next_sequence = None;
			0
		} else {
			(next_seq - playlist_seq) as usize
		};

		let available = total_segments.saturating_sub(skip);
		let to_process = if let Some(max) = limit {
			available.min(max)
		} else {
			available
		};

		info!(
			?kind,
			playlist_sequence = playlist_seq,
			next_sequence = next_seq,
			skip = skip,
			total_segments = total_segments,
			to_process = to_process,
			"consuming HLS segments"
		);

		if to_process > 0 {
			let base_seq = playlist_seq + skip as u64;

			// Discontinuity sequence names a fresh media timeline; the tag count for the
			// skipped prefix gets us to the discontinuity sequence of the first segment
			// we actually push.
			let mut discontinuity_seq = playlist.discontinuity_sequence;
			for segment in &playlist.segments[..skip] {
				if segment.discontinuity {
					discontinuity_seq = bump_discontinuity(discontinuity_seq)?;
				}
			}

			for (i, segment) in playlist.segments[skip..skip + to_process].iter().enumerate() {
				if segment.discontinuity {
					discontinuity_seq = bump_discontinuity(discontinuity_seq)?;
				}
				self.push_segment(kind, track, segment, base_seq + i as u64, discontinuity_seq)
					.await?;
			}
			info!(?kind, consumed = to_process, "consumed HLS segments");
		} else {
			debug!(?kind, "no fresh HLS segments available");
		}

		Ok(to_process)
	}

	async fn ensure_init_segment(
		&mut self,
		kind: TrackKind,
		track: &mut TrackState,
		playlist: &MediaPlaylist,
	) -> Result<()> {
		if track.init_ready {
			return Ok(());
		}

		let map = self.find_map(playlist).ok_or(Error::MissingMap)?;

		let url = resolve_uri(&track.playlist, &map.uri)?;
		let bytes = self.fetch_bytes(url).await?;
		let importer = match kind {
			TrackKind::Video(index) => self.ensure_video_importer_for(index, &track.select),
			TrackKind::Audio => self.ensure_audio_importer(&track.select),
		};

		// The importer buffers internally, so a fully-parsed init segment leaves it
		// initialized; any trailing partial atom just waits for the next segment. A
		// segment that never yields a moov surfaces later as a decode error.
		importer.decode(&bytes)?;

		track.init_ready = true;
		info!(?kind, "loaded HLS init segment");
		Ok(())
	}

	async fn push_segment(
		&mut self,
		kind: TrackKind,
		track: &mut TrackState,
		segment: &MediaSegment,
		sequence: u64,
		discontinuity_sequence: u64,
	) -> Result<()> {
		if segment.uri.is_empty() {
			return Err(Error::EmptySegmentUri);
		}

		let url = resolve_uri(&track.playlist, &segment.uri)?;
		let bytes = self.fetch_bytes(url).await?;

		// `consume_segments` always runs `ensure_init_segment` before reaching here, so
		// the importer is already initialized.
		let importer = match kind {
			TrackKind::Video(index) => self.ensure_video_importer_for(index, &track.select),
			TrackKind::Audio => self.ensure_audio_importer(&track.select),
		};

		// HLS media sequence names the live window, while discontinuity sequence names a
		// new media timeline. Whenever we join, skip ahead, or cross a discontinuity, anchor
		// the MoQ group sequence to both so consumers do not wait on groups HLS has moved
		// past. Contiguous segments let the importer auto-increment instead; we still pack
		// (and so validate) the sequence on that path so media sequence can't silently
		// auto-increment into the discontinuity bits.
		let group_sequence = moq_sequence(discontinuity_sequence, sequence)?;
		if track.next_sequence != Some(sequence) || track.next_discontinuity != Some(discontinuity_sequence) {
			importer.seek(group_sequence)?;
		}

		importer.decode(&bytes)?;
		track.next_sequence = Some(sequence + 1);
		track.next_discontinuity = Some(discontinuity_sequence);

		Ok(())
	}

	fn find_map<'a>(&self, playlist: &'a MediaPlaylist) -> Option<&'a Map> {
		playlist.segments.iter().find_map(|segment| segment.map.as_ref())
	}

	async fn fetch_bytes(&self, url: Url) -> Result<Bytes> {
		if url.scheme() == "file" {
			let path = url.to_file_path().map_err(|_| Error::InvalidFileUrl)?;
			let bytes = tokio::fs::read(&path).await.map_err(Error::from)?;
			Ok(Bytes::from(bytes))
		} else {
			let response = self.client.get(url).send().await.map_err(Error::from)?;
			let response = response.error_for_status().map_err(Error::from)?;
			let bytes = response.bytes().await.map_err(Error::from)?;
			Ok(bytes)
		}
	}

	/// Create or retrieve the fMP4 importer for a specific video rendition.
	///
	/// Each video variant gets its own importer so that their tracks remain
	/// independent while still contributing to the same shared catalog.
	fn ensure_video_importer_for(&mut self, index: usize, select: &select::Broadcast) -> &mut Fmp4 {
		while self.video_importers.len() <= index {
			let importer = Fmp4::new(self.broadcast.clone(), self.catalog.clone()).with_select(select.clone());
			self.video_importers.push(importer);
		}

		self.video_importers.get_mut(index).unwrap()
	}

	/// Create or retrieve the fMP4 importer for the audio rendition.
	fn ensure_audio_importer(&mut self, select: &select::Broadcast) -> &mut Fmp4 {
		let broadcast = self.broadcast.clone();
		let catalog = self.catalog.clone();
		let select = select.clone();
		self.audio_importer
			.get_or_insert_with(|| Fmp4::new(broadcast, catalog).with_select(select))
	}

	#[cfg(test)]
	fn has_video_importer(&self) -> bool {
		!self.video_importers.is_empty()
	}

	#[cfg(test)]
	fn has_audio_importer(&self) -> bool {
		self.audio_importer.is_some()
	}
}

fn select_audio<'a>(master: &'a MasterPlaylist, group_id: &str) -> Option<&'a AlternativeMedia> {
	let mut first = None;
	let mut default = None;

	for alternative in master
		.alternatives
		.iter()
		.filter(|alt| alt.media_type == AlternativeMediaType::Audio && alt.group_id == group_id)
	{
		if first.is_none() {
			first = Some(alternative);
		}
		if alternative.default {
			default = Some(alternative);
			break;
		}
	}

	default.or(first)
}

fn select_variants(master: &MasterPlaylist) -> Vec<&VariantStream> {
	// Map codec strings into a coarse "family" so we can prefer H.264 over others.
	fn codec_family(codec: &str) -> Option<&'static str> {
		if codec.starts_with("avc1.") || codec.starts_with("avc3.") {
			Some("h264")
		} else {
			None
		}
	}

	// Extract the first *video* codec token from the CODECS attribute. A list like
	// `mp4a.40.2,avc1.4d401f` (audio first) must still surface the video codec.
	fn first_video_codec(variant: &VariantStream) -> Option<&str> {
		let codecs = variant.codecs.as_deref()?;
		codecs
			.split(',')
			.map(|s| s.trim())
			.find(|codec| codec_family(codec).is_some())
	}

	// Consider only non-i-frame variants with a URI and a known codec family.
	let candidates: Vec<(&VariantStream, &str, &str)> = master
		.variants
		.iter()
		.filter(|variant| !variant.is_i_frame && !variant.uri.is_empty())
		.filter_map(|variant| {
			let codec = first_video_codec(variant)?;
			let family = codec_family(codec)?;
			Some((variant, codec, family))
		})
		.collect();

	if candidates.is_empty() {
		return Vec::new();
	}

	// Prefer families in this order, falling back to the first available.
	const FAMILY_PREFERENCE: &[&str] = &["h264"];

	let families_present: Vec<&str> = candidates.iter().map(|(_, _, fam)| *fam).collect();

	let target_family = FAMILY_PREFERENCE
		.iter()
		.find(|fav| families_present.iter().any(|fam| fam == *fav))
		.copied()
		.unwrap_or(families_present[0]);

	// Keep only variants in the chosen family.
	let family_variants: Vec<&VariantStream> = candidates
		.into_iter()
		.filter(|(_, _, fam)| *fam == target_family)
		.map(|(variant, _, _)| variant)
		.collect();

	// Deduplicate by resolution, keeping the lowest-bandwidth variant for each size.
	let mut by_resolution: HashMap<Option<Resolution>, &VariantStream> = HashMap::new();

	for variant in family_variants {
		let key = variant.resolution;
		let bandwidth = variant.average_bandwidth.unwrap_or(variant.bandwidth);

		match by_resolution.entry(key) {
			Entry::Vacant(entry) => {
				entry.insert(variant);
			}
			Entry::Occupied(mut entry) => {
				let existing = entry.get();
				let existing_bw = existing.average_bandwidth.unwrap_or(existing.bandwidth);
				if bandwidth < existing_bw {
					entry.insert(variant);
				}
			}
		}
	}

	by_resolution.values().cloned().collect()
}

fn resolve_uri(base: &Url, value: &str) -> std::result::Result<Url, url::ParseError> {
	if let Ok(url) = Url::parse(value) {
		return Ok(url);
	}

	base.join(value)
}

/// Advance the running discontinuity sequence, rejecting a u64 wrap on absurd input.
fn bump_discontinuity(sequence: u64) -> Result<u64> {
	sequence.checked_add(1).ok_or(Error::SequenceOverflow {
		kind: "discontinuity",
		value: sequence,
	})
}

/// Pack HLS discontinuity + media sequence into a single MoQ group sequence.
///
/// HLS media sequence alone can rewind after an upstream reset, and discontinuity
/// sequence alone cannot order segments inside the same epoch. The lower 48 bits hold
/// the media sequence (ample for realistic playlists) while the upper 16 bits hold the
/// discontinuity sequence, so a new epoch always sorts after every segment of the last.
fn moq_sequence(discontinuity_sequence: u64, media_sequence: u64) -> Result<u64> {
	const MEDIA_BITS: u32 = 48;
	const MEDIA_MASK: u64 = (1u64 << MEDIA_BITS) - 1;
	const DISCONTINUITY_MASK: u64 = u64::MAX >> MEDIA_BITS;

	if media_sequence > MEDIA_MASK {
		return Err(Error::SequenceOverflow {
			kind: "media",
			value: media_sequence,
		});
	}
	if discontinuity_sequence > DISCONTINUITY_MASK {
		return Err(Error::SequenceOverflow {
			kind: "discontinuity",
			value: discontinuity_sequence,
		});
	}

	Ok((discontinuity_sequence << MEDIA_BITS) | media_sequence)
}

#[cfg(test)]
mod tests {
	use super::*;

	#[test]
	fn hls_config_new_sets_fields() {
		let url = "https://example.com/stream.m3u8".to_string();
		let cfg = Config::new(url.clone());
		assert_eq!(cfg.playlist, url);
	}

	#[test]
	fn select_variants_handles_audio_first_codecs() {
		// CODECS lists the audio codec first; the video codec must still be found.
		let master = b"#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1000000,CODECS=\"mp4a.40.2,avc1.4d401f\"\nvideo.m3u8\n";
		let (_, master) = m3u8_rs::parse_master_playlist(master).unwrap();
		let variants = select_variants(&master);
		assert_eq!(variants.len(), 1);
	}

	#[test]
	fn hls_import_starts_without_importers() {
		let mut broadcast = moq_net::Broadcast::new().produce();
		let catalog = CatalogProducer::new(&mut broadcast).unwrap();
		let url = "https://example.com/master.m3u8".to_string();
		let cfg = Config::new(url);
		let hls = Import::new(broadcast, catalog, cfg).unwrap();

		assert!(!hls.has_video_importer());
		assert!(!hls.has_audio_importer());
	}

	/// Resolve `ensure_tracks` against a master playlist written to a temp file.
	async fn discover(master_body: &str) -> Import {
		use std::sync::atomic::{AtomicUsize, Ordering};
		static COUNTER: AtomicUsize = AtomicUsize::new(0);

		let n = COUNTER.fetch_add(1, Ordering::Relaxed);
		let dir = std::env::temp_dir().join(format!("moq-hls-test-{}-{n}", std::process::id()));
		std::fs::create_dir_all(&dir).unwrap();
		let path = dir.join("master.m3u8");
		std::fs::write(&path, master_body).unwrap();

		let mut broadcast = moq_net::Broadcast::new().produce();
		let catalog = CatalogProducer::new(&mut broadcast).unwrap();
		// `Config` takes a filesystem path for non-http inputs.
		let cfg = Config::new(path.to_str().unwrap().to_string());
		let mut hls = Import::new(broadcast, catalog, cfg).unwrap();
		hls.ensure_tracks().await.unwrap();
		hls
	}

	/// A master with a separate audio rendition: variants publish video only, the
	/// alternate rendition publishes audio only.
	#[tokio::test]
	async fn discover_splits_separate_audio() {
		let master = "#EXTM3U\n\
			#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID=\"aud\",NAME=\"en\",URI=\"audio.m3u8\"\n\
			#EXT-X-STREAM-INF:BANDWIDTH=1000000,CODECS=\"avc1.4d401f,mp4a.40.2\",AUDIO=\"aud\"\n\
			video.m3u8\n";
		let hls = discover(master).await;

		assert_eq!(hls.video.len(), 1);
		assert!(hls.video[0].select.has_video() && !hls.video[0].select.has_audio());

		let audio = hls.audio.as_ref().expect("separate audio rendition");
		assert!(audio.select.has_audio() && !audio.select.has_video());
	}

	/// A master whose variant carries muxed A/V (no separate audio group) publishes
	/// every track.
	#[tokio::test]
	async fn discover_muxed_variant_keeps_both() {
		let master = "#EXTM3U\n\
			#EXT-X-STREAM-INF:BANDWIDTH=1000000,CODECS=\"avc1.4d401f\"\n\
			video.m3u8\n";
		let hls = discover(master).await;

		assert_eq!(hls.video.len(), 1);
		assert!(hls.video[0].select.has_video() && hls.video[0].select.has_audio());
		assert!(hls.audio.is_none());
	}

	#[test]
	fn moq_sequence_orders_discontinuities_after_media_sequence() {
		// A new epoch outranks every segment of the previous one, even a higher media seq.
		let last_of_epoch_0 = moq_sequence(0, u64::from(u32::MAX)).unwrap();
		let first_of_epoch_1 = moq_sequence(1, 0).unwrap();
		assert!(first_of_epoch_1 > last_of_epoch_0);
	}

	#[test]
	fn moq_sequence_preserves_media_order_within_epoch() {
		assert!(moq_sequence(3, 10).unwrap() > moq_sequence(3, 9).unwrap());
	}

	#[test]
	fn moq_sequence_rejects_unrepresentable_media_sequence() {
		let err = moq_sequence(0, 1u64 << 48).unwrap_err();
		assert!(matches!(err, Error::SequenceOverflow { kind: "media", .. }));
	}

	#[test]
	fn moq_sequence_rejects_unrepresentable_discontinuity_sequence() {
		let err = moq_sequence(1u64 << 16, 0).unwrap_err();
		assert!(matches!(
			err,
			Error::SequenceOverflow {
				kind: "discontinuity",
				..
			}
		));
	}
}