use std::marker::PhantomData;
use std::time::Duration;

use super::Estimate;
use super::Producer;
use super::hang::{Catalog, CatalogExt};

/// A catalog config that can be published as a named rendition.
///
/// Implement it on your own config type to get the full catalog lifecycle: reservation gating,
/// removal on drop, and optional jitter/bitrate detection. [`VideoConfig`](hang::catalog::VideoConfig)
/// and [`AudioConfig`](hang::catalog::AudioConfig) implement it for every extension; a custom
/// config implements it for the one [`CatalogExt`] that holds it. Publish a media track under it
/// with [`Reserved::track`], or a data track with [`Producer::json_stream`] and the like when it
/// embeds a data config (see [`IntoRendition`](super::IntoRendition)):
///
/// ```
/// # use std::collections::BTreeMap;
/// # use hang::catalog::{JsonConfig, Mode};
/// # use moq_mux::catalog::{Estimate, RenditionConfig};
/// # use moq_mux::catalog::hang::{Catalog, CatalogExt};
/// # use serde::{Deserialize, Serialize};
/// #[derive(Serialize, Deserialize, Clone, Default)]
/// struct Ext {
///     #[serde(rename = "com.example.mavlink", default)]
///     mavlink: BTreeMap<String, Mavlink>,
/// }
/// impl CatalogExt for Ext {}
///
/// #[derive(Serialize, Deserialize, Clone)]
/// struct Mavlink {
///     config: JsonConfig,
///     sysid: u8,
/// }
///
/// impl AsMut<JsonConfig> for Mavlink {
///     fn as_mut(&mut self) -> &mut JsonConfig {
///         &mut self.config
///     }
/// }
///
/// impl RenditionConfig<Ext> for Mavlink {
///     fn insert(self, catalog: &mut Catalog<Ext>, name: &str) {
///         catalog.ext.mavlink.insert(name.to_string(), self);
///     }
///     fn get_mut<'a>(catalog: &'a mut Catalog<Ext>, name: &str) -> Option<&'a mut Self> {
///         catalog.ext.mavlink.get_mut(name)
///     }
///     fn remove(catalog: &mut Catalog<Ext>, name: &str) {
///         catalog.ext.mavlink.remove(name);
///     }
///
///     // Opt into detection through the embedded config.
///     fn detects() -> bool {
///         true
///     }
///     fn estimate(&self) -> Estimate {
///         Estimate::default()
///             .with_bitrate(self.config.bitrate)
///             .with_jitter(self.config.jitter)
///             .with_delay(self.config.delay)
///     }
///     fn set_estimate(&mut self, estimate: Estimate) {
///         self.config.bitrate = estimate.bitrate;
///         self.config.jitter = estimate.jitter;
///         self.config.delay = estimate.delay;
///     }
/// }
///
/// # fn example(
/// #     broadcast: &mut moq_net::broadcast::Producer,
/// #     catalog: &moq_mux::catalog::Producer<Ext>,
/// # ) -> moq_mux::Result<()> {
/// let info = moq_net::track::Info::default().with_compress(true);
/// let track = broadcast.create_track("telemetry", info)?;
/// // The producer fixes the mode, so the one passed here is only a placeholder.
/// let entry = Mavlink { config: JsonConfig::new(Mode::Stream), sysid: 1 };
/// let mut telemetry = catalog.json_stream::<serde_json::Value>(track, entry)?;
/// telemetry.append(&serde_json::json!({ "heartbeat": 1 }))?;
/// # Ok(())
/// # }
/// ```
///
/// Note that `insert` takes the whole [`Catalog`], not just the extension, so the built-in media
/// configs use the same trait. Writing to `catalog.video` / `catalog.audio` / `catalog.text` from a
/// custom config fights the media pipeline for those sections; stay in your own.
///
/// [`Reserved::track`] and [`Producer::track`](super::Producer::track) enroll the track in the
/// broadcast timeline, measure it, and keep its estimate current automatically.
pub trait RenditionConfig<E: CatalogExt>: Clone + Send + 'static {
	/// Whether container or data-track writes should update this config's estimate fields.
	fn detects() -> bool {
		false
	}

	/// Insert or replace this config under `name`.
	fn insert(self, catalog: &mut Catalog<E>, name: &str);

	/// Borrow the config stored under `name`, if it's present.
	fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self>;

	/// Remove the config stored under `name`.
	fn remove(catalog: &mut Catalog<E>, name: &str);

	/// The config's current [`Estimate`] fields. Defaults to none, disabling detection.
	fn estimate(&self) -> Estimate {
		Estimate::default()
	}

	/// Replace the config's [`Estimate`] fields. Defaults to discarding them.
	fn set_estimate(&mut self, _estimate: Estimate) {}
}

impl<E: CatalogExt> RenditionConfig<E> for hang::catalog::JsonConfig {
	fn insert(self, catalog: &mut Catalog<E>, name: &str) {
		catalog.json.tracks.insert(name.to_string(), self);
	}
	fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self> {
		catalog.json.tracks.get_mut(name)
	}
	fn remove(catalog: &mut Catalog<E>, name: &str) {
		catalog.json.tracks.remove(name);
	}

	fn detects() -> bool {
		true
	}
	fn estimate(&self) -> Estimate {
		Estimate::default()
			.with_jitter(self.jitter)
			.with_bitrate(self.bitrate)
			.with_delay(self.delay)
	}
	fn set_estimate(&mut self, estimate: Estimate) {
		self.jitter = estimate.jitter;
		self.bitrate = estimate.bitrate;
		self.delay = estimate.delay;
	}
}

impl<E: CatalogExt> RenditionConfig<E> for hang::catalog::BinaryConfig {
	fn insert(self, catalog: &mut Catalog<E>, name: &str) {
		catalog.binary.tracks.insert(name.to_string(), self);
	}
	fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self> {
		catalog.binary.tracks.get_mut(name)
	}
	fn remove(catalog: &mut Catalog<E>, name: &str) {
		catalog.binary.tracks.remove(name);
	}

	fn detects() -> bool {
		true
	}
	fn estimate(&self) -> Estimate {
		Estimate::default()
			.with_jitter(self.jitter)
			.with_bitrate(self.bitrate)
			.with_delay(self.delay)
	}
	fn set_estimate(&mut self, estimate: Estimate) {
		self.jitter = estimate.jitter;
		self.bitrate = estimate.bitrate;
		self.delay = estimate.delay;
	}
}

/// Caller-provided catalog fields for a video track: a starting point for what the importer detects.
///
/// Every field is optional and fills only a gap the stream leaves; a value the stream reveals (the
/// dimensions from an SPS, ...) always wins, so a detected change just updates the catalog. A
/// [`codec`](Self::codec) alone is enough to publish the catalog before the first keyframe (the
/// decoder config then arrives in band). Use it for fields the stream can't reveal (bitrate) or to
/// skip that wait. The MSF catalog declares only hinted estimates (`bitrate`, `framerate`, `jitter`,
/// `delay`), never measured ones, so an MSF track carries only those set here.
///
/// Video importers take one because they build the [`VideoConfig`](hang::catalog::VideoConfig)
/// themselves, out of the bitstream, so the caller never holds the config to set these on. An audio
/// importer publishes a complete config from its init bytes, so the field those bytes can't carry
/// (bitrate) rides on [`AudioInit`](crate::import::AudioInit) instead.
#[derive(Clone, Default, Debug, PartialEq)]
#[non_exhaustive]
pub struct VideoHint {
	/// Human-readable rendition name, plumbed through from [`Init::label`](crate::import::Init::label).
	///
	/// Not a hint: every other field here seeds something the stream may also reveal, while a label
	/// can only ever come from the caller. It rides along so one `apply` writes the whole config.
	pub(crate) label: Option<String>,
	/// The video codec.
	pub codec: Option<hang::catalog::VideoCodec>,
	/// The encoded width in pixels.
	pub coded_width: Option<u32>,
	/// The encoded height in pixels.
	pub coded_height: Option<u32>,
	/// The width of the aspect ratio the whole picture is displayed at, not the sample aspect.
	pub display_aspect_width: Option<u32>,
	/// The height of the aspect ratio the whole picture is displayed at.
	pub display_aspect_height: Option<u32>,
	/// The colour the pixels carry. Each field the stream signals outranks the hint's.
	pub color: Option<hang::catalog::Color>,
	/// The maximum bitrate in bits per second.
	pub bitrate: Option<u64>,
	/// The frame rate in frames per second.
	pub framerate: Option<f64>,
	/// The rendition's [`reorder`](hang::catalog::VideoConfig::reorder) depth.
	pub reorder: Option<u8>,
	/// If true, the decoder optimizes for latency.
	pub optimize_for_latency: Option<bool>,
	/// The maximum jitter before the next frame is emitted.
	pub jitter: Option<Duration>,
	/// How far this rendition trails the broadcast's earliest rendition.
	pub delay: Option<Duration>,
	/// The container wrapping each frame on the wire.
	///
	/// Unlike the other fields this is a choice, not a hint: the bitstream never reveals a
	/// container, so it is applied to every config the importer publishes rather than filling a gap.
	pub container: hang::catalog::Container,
	/// The rendition's [`preference`](hang::catalog::VideoConfig::preference).
	///
	/// A choice like the container: the bitstream never reveals it, so every config gets it.
	pub preference: i32,
}

/// Fill `slot` from `value` only when the slot is still empty, so a value the stream detected always
/// wins over the caller's hint.
fn fill<T>(slot: &mut Option<T>, value: Option<T>) {
	if slot.is_none() {
		*slot = value;
	}
}

impl From<hang::catalog::VideoConfig> for VideoHint {
	/// Carry a whole rendition across as hints, for a caller that already knows what its encoder
	/// emits (moq-video probes its encoder for exactly this).
	///
	/// Total by construction: every field the hint can hold is taken from the config, so there is no
	/// per-field copy for a caller to forget. Fields with no hint slot (`broadcast`, `description`,
	/// `enabled`) are set through the catalog directly.
	fn from(config: hang::catalog::VideoConfig) -> Self {
		Self {
			label: config.label,
			codec: Some(config.codec),
			coded_width: config.coded_width,
			coded_height: config.coded_height,
			display_aspect_width: config.display_aspect_width,
			display_aspect_height: config.display_aspect_height,
			color: config.color,
			bitrate: config.bitrate,
			framerate: config.framerate,
			reorder: config.reorder,
			optimize_for_latency: config.optimize_for_latency,
			jitter: config.jitter,
			delay: config.delay,
			container: config.container,
			preference: config.preference,
		}
	}
}

impl VideoHint {
	/// Fill a detected video config's absent optional fields from these hints.
	///
	/// Only the gaps: a value the stream detects (e.g. the dimensions from an SPS) is left untouched,
	/// so a resolution change updates the catalog instead of conflicting with the hint. An importer
	/// calls this on every config it publishes, before handing it to
	/// [`container::Producer::set`](crate::container::Producer::set), so a hinted field counts as
	/// supplied and is never overwritten by detection.
	pub fn apply(&self, config: &mut hang::catalog::VideoConfig) {
		fill(&mut config.label, self.label.clone());
		fill(&mut config.coded_width, self.coded_width);
		fill(&mut config.coded_height, self.coded_height);
		fill(&mut config.display_aspect_width, self.display_aspect_width);
		fill(&mut config.display_aspect_height, self.display_aspect_height);
		crate::codec::color::fill(&mut config.color, self.color.as_ref(), "hint");
		fill(&mut config.bitrate, self.bitrate);
		fill(&mut config.framerate, self.framerate);
		fill(&mut config.reorder, self.reorder);
		fill(&mut config.optimize_for_latency, self.optimize_for_latency);
		fill(&mut config.jitter, self.jitter);
		fill(&mut config.delay, self.delay);
		config.container = self.container.clone();
		config.preference = self.preference;
	}

	/// Build a config from these fields alone, or `None` if the codec is missing. Used to publish
	/// the catalog before the stream is parsed.
	pub fn to_config(&self) -> Option<hang::catalog::VideoConfig> {
		let codec = self.codec.clone()?;
		let mut config = hang::catalog::VideoConfig::new(codec);
		self.apply(&mut config);
		Some(config)
	}
}

impl<E: CatalogExt> RenditionConfig<E> for hang::catalog::VideoConfig {
	fn detects() -> bool {
		true
	}

	fn insert(self, catalog: &mut Catalog<E>, name: &str) {
		catalog.video.renditions.insert(name.to_string(), self);
	}
	fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self> {
		catalog.video.renditions.get_mut(name)
	}
	fn remove(catalog: &mut Catalog<E>, name: &str) {
		catalog.video.renditions.remove(name);
	}

	fn estimate(&self) -> Estimate {
		Estimate::default()
			.with_jitter(self.jitter)
			.with_bitrate(self.bitrate)
			.with_delay(self.delay)
			.with_framerate(self.framerate)
	}
	fn set_estimate(&mut self, estimate: Estimate) {
		self.jitter = estimate.jitter;
		self.bitrate = estimate.bitrate;
		self.delay = estimate.delay;
		self.framerate = estimate.framerate;
	}
}

impl<E: CatalogExt> RenditionConfig<E> for hang::catalog::AudioConfig {
	fn detects() -> bool {
		true
	}

	fn insert(self, catalog: &mut Catalog<E>, name: &str) {
		catalog.audio.renditions.insert(name.to_string(), self);
	}
	fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self> {
		catalog.audio.renditions.get_mut(name)
	}
	fn remove(catalog: &mut Catalog<E>, name: &str) {
		catalog.audio.renditions.remove(name);
	}

	fn estimate(&self) -> Estimate {
		Estimate::default()
			.with_jitter(self.jitter)
			.with_bitrate(self.bitrate)
			.with_delay(self.delay)
	}
	fn set_estimate(&mut self, estimate: Estimate) {
		self.jitter = estimate.jitter;
		self.bitrate = estimate.bitrate;
		self.delay = estimate.delay;
	}
}

impl<E: CatalogExt> RenditionConfig<E> for hang::catalog::TextConfig {
	fn insert(self, catalog: &mut Catalog<E>, name: &str) {
		catalog.text.renditions.insert(name.to_string(), self);
	}
	fn get_mut<'a>(catalog: &'a mut Catalog<E>, name: &str) -> Option<&'a mut Self> {
		catalog.text.renditions.get_mut(name)
	}
	fn remove(catalog: &mut Catalog<E>, name: &str) {
		catalog.text.renditions.remove(name);
	}
	fn estimate(&self) -> Estimate {
		Estimate::default().with_jitter(self.jitter).with_delay(self.delay)
	}
	fn set_estimate(&mut self, estimate: Estimate) {
		self.jitter = estimate.jitter;
		self.delay = estimate.delay;
	}
}

/// How far short of a whole frame duration a value still counts as reaching it, as a fraction of the
/// period: a reorder of exactly one frame must not miss its step to rounding in the frame duration.
const STEP_SLACK: u32 = 100;

/// `measured` if it reaches a whole `period` beyond the one `published` sits in, else `published`.
fn quantize(published: Option<Duration>, measured: Option<Duration>, period: Duration) -> Option<Duration> {
	let step =
		|value: Option<Duration>| value.map_or(0, |value| (value + period / STEP_SLACK).as_nanos() / period.as_nanos());
	if step(measured) > step(published) {
		measured
	} else {
		published
	}
}

/// Refuse an edit that redefines rendition `name`'s track identity in place: a publisher whose
/// decoder identity changed publishes under a new name instead.
fn check_identity<E: CatalogExt>(previous: &Catalog<E>, next: &Catalog<E>, name: &str) -> crate::Result<()> {
	if let (Some(previous), Some(next)) = (previous.video.renditions.get(name), next.video.renditions.get(name))
		&& !previous.same_identity(next)
	{
		return Err(crate::Error::IdentityChanged(name.to_string()));
	}
	if let (Some(previous), Some(next)) = (previous.audio.renditions.get(name), next.audio.renditions.get(name))
		&& !previous.same_identity(next)
	{
		return Err(crate::Error::IdentityChanged(name.to_string()));
	}
	Ok(())
}

/// A clonable reservation context handed to importers so they declare their tracks up front.
///
/// Made via [`Producer::reserve`]. While any `Reserved` clone is alive the track set may still
/// grow, so the catalog is withheld from the broadcast. Each track constructor reserves a
/// rendition by name and counts as outstanding until its returned producer publishes a config or
/// is dropped. Once every clone is dropped *and* every reservation resolves, the first catalog
/// snapshot is published atomically with the complete track list, so a one-shot muxer (fMP4,
/// MPEG-TS) never sees a half-converged catalog.
///
/// It belongs to one [`Timebase`](super::Timebase), whose offset the importer handed it shifts its
/// timestamps by. Clones share that timebase.
pub struct Reserved<E: CatalogExt = ()> {
	timebase: super::Timebase<E>,
	hls_hint: Option<std::time::Duration>,
}

impl<E: CatalogExt> Reserved<E> {
	pub(super) fn new(timebase: super::Timebase<E>) -> Self {
		timebase.catalog.add_reserver();
		Self {
			timebase,
			hls_hint: None,
		}
	}

	/// Declare the track's regular group duration, which fixes its HLS segment duration up front
	/// unless [`Config::with_hls`](super::Config::with_hls) disabled HLS records.
	pub fn with_hls_hint(mut self, duration: impl Into<Option<std::time::Duration>>) -> Self {
		self.hls_hint = duration.into();
		self
	}

	/// Track properties for a media track under this catalog, carrying any retention it declares.
	/// See [`Producer::track_info`](super::Producer::track_info).
	pub fn track_info(&self, priority: i8) -> moq_net::track::Info {
		self.timebase.catalog.track_info(priority)
	}

	/// Reserve a rendition of config type `C` under `name`, returning its internal owner.
	///
	/// The owner holds its own `Reserved` clone, so the catalog stays withheld until its config is
	/// published (or it is dropped). Prefer [`video`](Self::video) / [`audio`](Self::audio) for the
	/// built-in media configs.
	///
	/// Errors if the name is already taken in this section, whether by a live rendition or by an
	/// entry in the catalog. Sections are independent, so the same name in `video` and `audio` is
	/// two unrelated renditions.
	pub(crate) fn init<C: RenditionConfig<E>>(&self, name: impl Into<String>) -> crate::Result<Rendition<E, C>> {
		Rendition::new(self.clone(), name.into())
	}

	/// Publish a reserved video track and own its catalog rendition.
	pub fn video<C: crate::container::Container>(
		&self,
		track: moq_net::track::Producer,
		container: C,
		config: impl Into<Option<hang::catalog::VideoConfig>>,
	) -> crate::Result<crate::container::Producer<C, hang::catalog::VideoConfig>>
	where
		crate::Error: From<C::Error>,
	{
		self.track(track, container, config)
	}

	/// Publish a reserved audio track and own its catalog rendition.
	pub fn audio<C: crate::container::Container>(
		&self,
		track: moq_net::track::Producer,
		container: C,
		config: impl Into<Option<hang::catalog::AudioConfig>>,
	) -> crate::Result<crate::container::Producer<C, hang::catalog::AudioConfig>>
	where
		crate::Error: From<C::Error>,
	{
		self.track(track, container, config)
	}

	/// Publish a reserved text track and own its catalog rendition.
	pub fn text<C: crate::container::Container>(
		&self,
		track: moq_net::track::Producer,
		container: C,
		config: impl Into<Option<hang::catalog::TextConfig>>,
	) -> crate::Result<crate::container::Producer<C, hang::catalog::TextConfig>>
	where
		crate::Error: From<C::Error>,
	{
		self.track(track, container, config)
	}

	/// Publish a reserved track using a custom catalog config.
	pub fn track<C, R>(
		&self,
		track: moq_net::track::Producer,
		container: C,
		config: impl Into<Option<R>>,
	) -> crate::Result<crate::container::Producer<C, R>>
	where
		C: crate::container::Container,
		R: RenditionConfig<E>,
		crate::Error: From<C::Error>,
	{
		let rendition = self.init(track.name())?;
		self.timebase
			.catalog
			.media(track, container, rendition, config.into(), self.hls_hint)
	}

	/// Resolve a timestamp on the broadcast's shared clock (see [`Producer::timestamp`]).
	pub fn timestamp(&self, hint: Option<moq_net::Timestamp>) -> crate::Result<moq_net::Timestamp> {
		self.timebase.catalog.timestamp(hint)
	}

	/// The underlying catalog [`Producer`], for edits that outlive this reservation.
	///
	/// A container importer holds one to edit the catalog directly (e.g. its per-frame reconcile, or
	/// track removals after its initial set is declared) while the reservation itself is dropped to
	/// open the gate. The returned handle does not gate: only live `Reserved`s do.
	pub(crate) fn producer(&self) -> Producer<E> {
		self.timebase.catalog.clone()
	}

	/// The timebase this reservation belongs to, which outlives it without gating the catalog.
	///
	/// An importer anchors through it on its first frame and shifts every frame by its offset.
	pub(crate) fn timebase(&self) -> super::Timebase<E> {
		self.timebase.clone()
	}
}

impl<E: CatalogExt> Clone for Reserved<E> {
	fn clone(&self) -> Self {
		Self::new(self.timebase.clone()).with_hls_hint(self.hls_hint)
	}
}

impl<E: CatalogExt> Drop for Reserved<E> {
	fn drop(&mut self) {
		self.timebase.catalog.release_reserver();
	}
}

/// A reserved rendition of config type `C`, retired from the catalog on drop.
///
/// Made via [`Reserved::init`] (or [`video`](Reserved::video) / [`audio`](Reserved::audio)). Fill
/// it in with [`set`](Self::set) and refine it in place with [`update`](Self::update). Until it's
/// set (or dropped) it holds a [`Reserved`] clone, so an unresolved rendition keeps the initial
/// catalog publish gated. On drop the rendition is removed from the shared catalog.
///
/// It owns its name for its whole lifetime, from reservation rather than from `set`, so nothing
/// else can reserve that name in the same section meanwhile. That's what a lazily-configured
/// importer needs: an H.264 track publishes no config until its first SPS, and the name has to be
/// unavailable through that window or an entry written into it is overwritten by `set` and then
/// deleted by this `Drop`. Author a rendition by holding one of these, not by writing to
/// `catalog.video` / `catalog.audio` through a [`Guard`](super::Guard), which is raw access to the
/// catalog document and enforces nothing.
pub(crate) struct Rendition<E: CatalogExt, C: RenditionConfig<E>> {
	catalog: Producer<E>,
	name: String,
	/// The reservation this rendition holds until its config is set (or it's dropped unfulfilled).
	/// `Some` gates the initial publish; cleared by [`set`](Self::set).
	gate: Option<Reserved<E>>,
	/// Whether a config has been published, so a lazily-configured importer (e.g. H.264 before its
	/// SPS) holds the handle without a catalog entry, and drops without a spurious removal.
	present: bool,
	/// The fields the config carried at [`set`](Self::set): authoritative, never overwritten by
	/// detection. Whatever is absent here is what detection fills, and stays absent from MSF.
	supplied: Estimate,
	/// The last estimate handed to [`estimate`](Self::estimate), kept so a config set afterwards
	/// starts from whatever was already measured.
	detected: Estimate,
	/// The last estimate written to the config, so an unchanged one doesn't republish the catalog.
	published: Option<Estimate>,
	_config: PhantomData<fn() -> C>,
}

/// A single video track's catalog rendition. See [`Rendition`].
pub(crate) type VideoTrack<E = ()> = Rendition<E, hang::catalog::VideoConfig>;
/// A single audio track's catalog rendition. See [`Rendition`].
pub(crate) type AudioTrack<E = ()> = Rendition<E, hang::catalog::AudioConfig>;
/// A single text (caption/subtitle) track's catalog rendition. See [`Rendition`].
impl<E: CatalogExt, C: RenditionConfig<E>> Rendition<E, C> {
	/// The successor of this rendition's track, under a newly minted name, for a decoder identity
	/// change. `hls_hint` carries the segment duration the old track declared.
	pub(crate) fn fresh(
		&self,
		container: super::hang::Container,
		info: moq_net::track::Info,
		hls_hint: Option<std::time::Duration>,
	) -> crate::Result<crate::container::Producer<super::hang::Container, C>> {
		let track = self.catalog.fresh_track(&self.name, info)?;
		let rendition = self.catalog.rendition(track.name())?;
		self.catalog.media(track, container, rendition, None, hls_hint)
	}

	/// Finish and forget the HLS records of `track`, a predecessor this rendition replaced.
	pub(crate) fn retire(&self, track: &str) -> crate::Result<()> {
		self.catalog.retire(track)
	}

	fn new(reserved: Reserved<E>, name: String) -> crate::Result<Self> {
		Self::owned(reserved.producer(), Some(reserved), name)
	}

	pub(super) fn live(catalog: Producer<E>, name: String) -> crate::Result<Self> {
		Self::owned(catalog, None, name)
	}

	fn owned(catalog: Producer<E>, gate: Option<Reserved<E>>, name: String) -> crate::Result<Self> {
		// Take the name now, not at `set`: a lazily-configured importer (H.264 before its first SPS)
		// has no catalog entry until much later, and the name has to be ours for that whole window.
		catalog.acquire::<C>(&name)?;

		Ok(Self {
			catalog,
			gate,
			name,
			present: false,
			supplied: Estimate::default(),
			detected: Estimate::default(),
			published: None,
			_config: PhantomData,
		})
	}

	/// The track name this rendition is keyed by.
	pub fn name(&self) -> &str {
		&self.name
	}

	/// A fresh estimator measuring `delay` against the catalog's other renditions.
	pub(crate) fn estimator(&self) -> super::Estimator {
		self.catalog.estimator()
	}

	/// The broadcast clock the catalog stamps its tracks on, read without fixing it.
	pub(crate) fn clock(&self) -> crate::Clock {
		self.catalog.current_clock()
	}

	/// Resolve a timestamp on the broadcast's shared clock (see [`Producer::timestamp`]).
	pub fn timestamp(&self, hint: Option<moq_net::Timestamp>) -> crate::Result<moq_net::Timestamp> {
		self.catalog.timestamp(hint)
	}

	/// Validate and publish the rendition, fulfilling its reservation only after the edit succeeds.
	///
	/// Whatever [`Estimate`] fields `config` already carries are authoritative and left alone; the
	/// rest are filled by [`estimate`](Self::estimate), seeded with anything already measured before
	/// the rendition existed (a dirty start or a B-frame reorder). Only the authoritative ones reach
	/// the MSF catalog. A caller who wants to pre-empt detection sets the field on the config, or (for
	/// a config an importer builds out of the bitstream) hands the importer a hint like [`VideoHint`].
	pub(crate) fn set(&mut self, mut config: C) -> crate::Result<()> {
		let supplied = config.estimate();
		let resolved = Self::resolved(&supplied, &self.detected);
		self.check_decrease(&resolved)?;
		config.set_estimate(resolved.clone());
		{
			let mut guard = self.catalog.modify()?;
			let mut next = (*guard).clone();
			config.insert(&mut next, &self.name);
			check_identity(&guard, &next, &self.name)?;
			// Serialization must succeed before the reserved snapshot retains the edit.
			serde_json::to_writer(std::io::sink(), &next).map_err(moq_json::Error::from)?;
			self.present = true;
			*guard = next;
			guard.list::<C>(&self.name);
			if C::detects() {
				guard.hint::<C>(&self.name, supplied.clone());
			}
			guard.commit()?;
		}
		self.supplied = supplied;
		self.published = Some(resolved);
		self.gate = None;
		Ok(())
	}

	/// The supplied fields, with anything absent filled from what was detected.
	fn resolved(supplied: &Estimate, detected: &Estimate) -> Estimate {
		let mut estimate = supplied.clone();
		if estimate.jitter.is_none() {
			estimate.jitter = detected.jitter;
		}
		if estimate.bitrate.is_none() {
			estimate.bitrate = detected.bitrate;
		}
		if estimate.delay.is_none() {
			estimate.delay = detected.delay;
		}
		if estimate.framerate.is_none() {
			estimate.framerate = detected.framerate;
		}
		estimate
	}

	/// Publish a measured [`Estimate`], filling only the fields the config didn't supply at
	/// [`set`](Self::set).
	///
	/// Mint the reading from an [`Estimator`](super::Estimator), usually the one owned by a
	/// [`container::Producer`](crate::container::Producer). Call it after every write: an estimate
	/// that resolves to what the catalog already carries doesn't republish it. A `jitter` or `delay`
	/// rise republishes only once it reaches the next whole frame duration above the published value,
	/// and then at most once a second (see [`Guard::commit_estimate`](super::Guard::commit_estimate)),
	/// going out on a later call.
	///
	/// Calling this before [`set`](Self::set) is not wasted: the measurement is remembered and seeds
	/// the config once it lands.
	pub(crate) fn estimate(&mut self, reading: impl Into<super::Reading>) -> crate::Result<()> {
		let super::Reading { estimate, period } = reading.into();
		if !C::detects() {
			return Ok(());
		}
		if !self.present {
			self.detected = estimate;
			return Ok(());
		}
		let mut resolved = Self::resolved(&self.supplied, &estimate);
		// A measurement never lowers the published jitter or delay, including one raised through
		// `modify`.
		let mut config = self.config()?;
		let published = config.estimate();
		resolved.jitter = resolved.jitter.max(published.jitter);
		resolved.delay = resolved.delay.max(published.delay);
		// Measurement creeps up as the encoder's latency does, a step per frame at worst; only a whole
		// frame's rise is worth a republish. Until a period is known the throttle alone paces it.
		let period = resolved
			.framerate
			.filter(|framerate| *framerate > 0.0)
			.and_then(|framerate| Duration::try_from_secs_f64(framerate.recip()).ok())
			.or(period)
			.filter(|period| !period.is_zero());
		if let Some(period) = period {
			resolved.jitter = quantize(published.jitter, resolved.jitter, period);
			resolved.delay = quantize(published.delay, resolved.delay, period);
		}
		self.detected = estimate;
		if self.published.as_ref() == Some(&resolved) {
			// Still an observation: it releases a rise held back until its window ended. A closed
			// catalog has nothing left to release.
			if let Ok(guard) = self.catalog.modify() {
				guard.commit_estimate(true)?;
			}
			return Ok(());
		}

		// Bitrate and framerate are already measured over a second of media, so they never churn;
		// only jitter and delay climb a step per frame.
		let throttled = self.published.as_ref().is_some_and(|published| {
			published.bitrate == resolved.bitrate && published.framerate == resolved.framerate
		});
		config.set_estimate(resolved.clone());
		self.stage(config)?.commit_estimate(throttled)?;
		self.published = Some(resolved);
		Ok(())
	}

	/// Refuse a config that lowers the jitter or delay already advertised to subscribers.
	fn check_decrease(&self, next: &Estimate) -> crate::Result<()> {
		if !self.present {
			return Ok(());
		}
		let previous = self.config()?.estimate();
		if next.jitter < previous.jitter {
			return Err(crate::Error::JitterDecreased);
		}
		if next.delay < previous.delay {
			return Err(crate::Error::DelayDecreased);
		}
		Ok(())
	}

	/// Refine the rendition, refusing an unserializable edit before retaining it.
	pub(crate) fn config(&self) -> crate::Result<C> {
		if !self.present {
			return Err(crate::Error::NotPublished);
		}
		self.catalog.config::<C>(&self.name).ok_or(crate::Error::NotPublished)
	}

	pub(crate) fn replace(&mut self, config: C) -> crate::Result<()> {
		self.stage(config)?.commit()
	}

	/// Advertise `bitrate` as a supplied value that measurement no longer replaces, leaving every
	/// other field, measured or not, as published.
	pub(crate) fn supply_bitrate(&mut self, bitrate: u64) -> crate::Result<()> {
		let mut config = self.config()?;
		let mut estimate = config.estimate();
		estimate.bitrate = Some(bitrate);
		config.set_estimate(estimate);
		let mut supplied = self.supplied.clone();
		supplied.bitrate = Some(bitrate);
		let name = self.name.clone();
		let mut guard = self.stage(config)?;
		if C::detects() {
			guard.hint::<C>(&name, supplied.clone());
		}
		guard.commit()?;
		self.supplied = supplied;
		Ok(())
	}

	/// Write `config` into the catalog, leaving the caller to choose how the guard publishes it.
	fn stage(&mut self, config: C) -> crate::Result<super::Guard<'_, E>> {
		if !self.present {
			return Err(crate::Error::NotPublished);
		}
		self.check_decrease(&config.estimate())?;
		let mut guard = self.catalog.modify()?;
		let mut next = (*guard).clone();
		config.insert(&mut next, &self.name);
		check_identity(&guard, &next, &self.name)?;
		serde_json::to_writer(std::io::sink(), &next).map_err(moq_json::Error::from)?;
		*guard = next;
		Ok(guard)
	}
}

// Manual so a config that isn't `Debug` doesn't cost the handle its own, and because the interesting
// state is the name and whether it has published yet, not the estimate bookkeeping.
impl<E: CatalogExt, C: RenditionConfig<E>> std::fmt::Debug for Rendition<E, C> {
	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
		f.debug_struct("Rendition")
			.field("name", &self.name)
			.field("published", &self.present)
			.finish_non_exhaustive()
	}
}

impl<E: CatalogExt, C: RenditionConfig<E>> Drop for Rendition<E, C> {
	fn drop(&mut self) {
		// The entry and the name it holds are released together under one lock. Removing mutates the
		// catalog, so the guard publishes it (immediately if live, else it accumulates until the gate
		// opens). A closed catalog has nothing left to release from.
		if let Ok(mut catalog) = self.catalog.modify() {
			catalog.release::<C>(&self.name, self.present);
		}
		// Our reservation (`gate`) drops here. If still held (never set), its release flushes any
		// staged change; if already released by `set`, this is a no-op.
	}
}

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

	/// A millisecond track: the frames written here carry timestamps.
	fn timed() -> moq_net::track::Info {
		moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI)
	}

	fn video_track() -> (moq_net::broadcast::Producer, super::super::Producer, VideoTrack) {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap();
		let reserved = catalog.reserve();
		let rendition = reserved.init::<hang::catalog::VideoConfig>("v").unwrap();
		// Drop the standalone reservation so only the rendition's own gate remains, which `set`
		// clears; the broadcast handle is returned so the produced tracks outlive the catalog.
		drop(reserved);
		(broadcast, catalog, rendition)
	}

	fn config(bitrate: Option<u64>, jitter: Option<Duration>) -> hang::catalog::VideoConfig {
		let mut config = hang::catalog::VideoConfig::new(hang::catalog::VideoCodec::VP8);
		config.bitrate = bitrate;
		config.jitter = jitter;
		config
	}

	#[test]
	fn video_identity_edits_are_refused_without_changing_the_catalog() {
		let original = config(Some(1_000_000), None);
		let mut changed = Vec::new();
		let mut codec = original.clone();
		codec.codec = "avc1.42c01e".parse().unwrap();
		changed.push(codec);
		let mut description = original.clone();
		description.description = Some(bytes::Bytes::from_static(&[1, 2, 3]));
		changed.push(description);
		let mut container = original.clone();
		container.container = hang::catalog::Container::Loc;
		changed.push(container);
		let mut width = original.clone();
		width.display_aspect_width = Some(4);
		changed.push(width);
		let mut height = original.clone();
		height.display_aspect_height = Some(3);
		changed.push(height);
		for next in changed {
			for operation in ["set", "replace", "stage"] {
				let (_broadcast, catalog, mut rendition) = video_track();
				rendition.set(original.clone()).unwrap();
				let refused = match operation {
					"set" => rendition.set(next.clone()).is_err(),
					"replace" => rendition.replace(next.clone()).is_err(),
					"stage" => rendition.stage(next.clone()).is_err(),
					_ => unreachable!(),
				};
				assert!(refused, "{operation} retained identity change {next:?}");
				assert_eq!(catalog.snapshot().video.renditions["v"], original);
			}
		}
	}

	#[test]
	fn video_size_and_estimates_can_change_in_place() {
		let (_broadcast, catalog, mut rendition) = video_track();
		let mut original = config(Some(1_000_000), None);
		original.coded_width = Some(1280);
		original.coded_height = Some(720);
		original.framerate = Some(60.0);
		rendition.set(original.clone()).unwrap();
		let mut next = original;
		next.coded_width = Some(1920);
		next.coded_height = Some(1080);
		next.bitrate = Some(500_000);
		next.framerate = Some(30.0);
		rendition.set(next.clone()).unwrap();
		assert_eq!(catalog.snapshot().video.renditions["v"], next);
	}

	#[test]
	fn audio_identity_edits_are_refused_without_changing_the_catalog() {
		let original = hang::catalog::AudioConfig::new(hang::catalog::AudioCodec::Opus, 48_000, 2);
		let mut changed = Vec::new();
		let mut codec = original.clone();
		codec.codec = hang::catalog::AudioCodec::Flac;
		changed.push(codec);
		let mut description = original.clone();
		description.description = Some(bytes::Bytes::from_static(&[1, 2, 3]));
		changed.push(description);
		let mut container = original.clone();
		container.container = hang::catalog::Container::Loc;
		changed.push(container);
		let mut sample_rate = original.clone();
		sample_rate.sample_rate = 44_100;
		changed.push(sample_rate);
		let mut channels = original.clone();
		channels.channel_count = 1;
		changed.push(channels);
		for next in changed {
			for operation in ["set", "replace", "stage"] {
				let mut broadcast = moq_net::broadcast::Info::new().produce();
				let catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap();
				let reserved = catalog.reserve();
				let mut rendition = reserved.init::<hang::catalog::AudioConfig>("a").unwrap();
				drop(reserved);
				rendition.set(original.clone()).unwrap();
				let refused = match operation {
					"set" => rendition.set(next.clone()).is_err(),
					"replace" => rendition.replace(next.clone()).is_err(),
					"stage" => rendition.stage(next.clone()).is_err(),
					_ => unreachable!(),
				};
				assert!(refused, "{operation} retained identity change {next:?}");
				assert_eq!(catalog.snapshot().audio.renditions["a"], original);
			}
		}
	}

	#[test]
	fn rejected_jitter_is_not_retained() {
		let (_broadcast, catalog, mut rendition) = video_track();
		for jitter in [Duration::ZERO, Duration::MAX] {
			assert!(rendition.set(config(None, Some(jitter))).is_err());
			assert!(catalog.snapshot().video.renditions.is_empty());
		}
		rendition.set(config(None, Some(Duration::from_millis(100)))).unwrap();
		let mut invalid = rendition.config().unwrap();
		invalid.jitter = Some(Duration::ZERO);
		assert!(rendition.replace(invalid).is_err());
		assert_eq!(
			catalog.snapshot().video.renditions.values().next().unwrap().jitter,
			Some(Duration::from_millis(100))
		);
	}

	#[test]
	fn published_jitter_never_decreases() {
		let (_broadcast, catalog, mut rendition) = video_track();
		rendition.set(config(None, Some(Duration::from_millis(100)))).unwrap();
		let smaller = config(None, Some(Duration::from_millis(50)));
		assert!(matches!(
			rendition.set(smaller.clone()),
			Err(crate::Error::JitterDecreased)
		));
		assert!(matches!(rendition.replace(smaller), Err(crate::Error::JitterDecreased)));
		assert_eq!(
			catalog.snapshot().video.renditions["v"].jitter,
			Some(Duration::from_millis(100))
		);

		let (_broadcast, catalog, mut detected) = video_track();
		detected.set(config(None, None)).unwrap();
		detected
			.estimate(Estimate::default().with_jitter(Duration::from_millis(100)))
			.unwrap();
		// A lower measurement holds the published value rather than failing the write path.
		detected
			.estimate(Estimate::default().with_jitter(Duration::from_millis(50)))
			.unwrap();
		assert_eq!(
			catalog.snapshot().video.renditions["v"].jitter,
			Some(Duration::from_millis(100))
		);
	}

	#[test]
	fn published_delay_never_decreases() {
		let (_broadcast, catalog, mut rendition) = video_track();
		let delayed = |delay| {
			let mut config = config(None, None);
			config.delay = delay;
			config
		};
		rendition.set(delayed(Some(Duration::from_millis(200)))).unwrap();
		for smaller in [Some(Duration::from_millis(100)), None] {
			assert!(matches!(
				rendition.set(delayed(smaller)),
				Err(crate::Error::DelayDecreased)
			));
			assert!(matches!(
				rendition.replace(delayed(smaller)),
				Err(crate::Error::DelayDecreased)
			));
		}
		assert_eq!(
			catalog.snapshot().video.renditions["v"].delay,
			Some(Duration::from_millis(200))
		);

		let (_broadcast, catalog, mut detected) = video_track();
		detected.set(config(None, None)).unwrap();
		detected
			.estimate(Estimate::default().with_delay(Duration::from_millis(200)))
			.unwrap();
		// A lower measurement holds the published value rather than failing the write path.
		detected
			.estimate(Estimate::default().with_delay(Duration::from_millis(100)))
			.unwrap();
		assert_eq!(
			catalog.snapshot().video.renditions["v"].delay,
			Some(Duration::from_millis(200))
		);
	}

	#[test]
	fn measurement_keeps_a_jitter_raised_by_modify() {
		let (_broadcast, catalog, mut rendition) = video_track();
		rendition.set(config(None, None)).unwrap();
		rendition
			.estimate(Estimate::default().with_jitter(Duration::from_millis(60)))
			.unwrap();
		let mut raised = rendition.config().unwrap();
		raised.jitter = Some(Duration::from_millis(200));
		rendition.replace(raised).unwrap();

		rendition
			.estimate(Estimate::default().with_jitter(Duration::from_millis(80)))
			.unwrap();
		assert_eq!(
			catalog.snapshot().video.renditions["v"].jitter,
			Some(Duration::from_millis(200))
		);
		rendition
			.estimate(Estimate::default().with_jitter(Duration::from_millis(300)))
			.unwrap();
		assert_eq!(
			catalog.snapshot().video.renditions["v"].jitter,
			Some(Duration::from_millis(300))
		);
	}

	#[test]
	fn published_text_jitter_never_decreases() {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap();
		let reserved = catalog.reserve();
		let mut rendition = reserved.init::<hang::catalog::TextConfig>("t").unwrap();
		drop(reserved);

		let text = |jitter| {
			let mut config = hang::catalog::TextConfig::new(hang::catalog::TextFormat::Utf8);
			config.jitter = jitter;
			config
		};
		rendition.set(text(Some(Duration::from_millis(100)))).unwrap();
		for smaller in [Some(Duration::from_millis(50)), None] {
			assert!(matches!(
				rendition.set(text(smaller)),
				Err(crate::Error::JitterDecreased)
			));
		}
		assert_eq!(
			catalog.snapshot().text.renditions["t"].jitter,
			Some(Duration::from_millis(100))
		);
	}

	#[test]
	fn importer_returns_rejected_jitter() {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap();
		let reserved = catalog.reserve();
		let track = broadcast
			.create_track("audio", hang::container::track_info(hang::catalog::PRIORITY.audio))
			.unwrap();
		let mut config: hang::catalog::AudioConfig = crate::codec::aac::Config {
			profile: 2,
			sample_rate: 48_000,
			channel_count: 2,
		}
		.try_into()
		.unwrap();
		config.jitter = Some(Duration::ZERO);
		assert!(crate::codec::aac::Import::new(track, reserved, config).is_err());
		assert!(catalog.snapshot().audio.renditions.is_empty());
	}

	fn ts(micros: u64) -> moq_net::Timestamp {
		moq_net::Timestamp::from_micros(micros).unwrap()
	}

	/// Feed ~40ms 100 kB frames over more than the bitrate window, with a measured
	/// 40ms container batch span. PTS spacing alone says nothing about jitter.
	fn feed<E: CatalogExt, C: RenditionConfig<E>>(rendition: &mut Rendition<E, C>) {
		let mut estimator = super::super::Estimator::new();
		estimator.burst(Duration::from_millis(40));
		for i in 0..60u64 {
			let t = ts(i * 40_000);
			estimator.cut(Some(t));
			estimator.write(t, 100_000);
			rendition.estimate(estimator.reading()).unwrap();
		}
	}

	#[test]
	fn detects_absent_jitter_and_bitrate() {
		let (_broadcast, catalog, mut rendition) = video_track();
		rendition.set(config(None, None)).unwrap();
		feed(&mut rendition);

		let snapshot = catalog.snapshot();
		let config = snapshot.video.renditions.get("v").unwrap();
		assert!(config.jitter.is_some(), "absent jitter should be auto-detected");
		assert!(config.bitrate.is_some(), "absent bitrate should be auto-detected");
		assert_eq!(config.framerate, Some(25.0), "absent framerate should be auto-detected");
	}

	#[test]
	fn keeps_provided_jitter_and_bitrate() {
		let (_broadcast, catalog, mut rendition) = video_track();
		rendition
			.set(config(Some(123), Some(Duration::from_millis(50))))
			.unwrap();
		feed(&mut rendition);

		let snapshot = catalog.snapshot();
		let config = snapshot.video.renditions.get("v").unwrap();
		assert_eq!(config.bitrate, Some(123), "a provided bitrate must not be overwritten");
		assert_eq!(
			config.jitter,
			Some(Duration::from_millis(50)),
			"a provided jitter must not be overwritten"
		);
	}

	/// A caller hands moq-video a whole `VideoConfig`, so the conversion into hints must be total.
	/// Hand-copying it field by field is what dropped the label on that path.
	#[test]
	fn a_config_converts_into_hints_without_losing_the_label() {
		let (_broadcast, catalog, mut rendition) = video_track();

		let mut source = config(Some(456), None);
		source.label = Some("Main camera".to_string());
		source.coded_width = Some(1920);
		source.coded_height = Some(1080);
		let hint = VideoHint::from(source);

		// The importer applies the hint to each config it publishes, so check the label survives
		// both the up-front publish and a later one the stream resolved.
		let mut first = config(None, None);
		hint.apply(&mut first);
		rendition.set(first).unwrap();
		feed(&mut rendition);
		assert_eq!(
			catalog.snapshot().video.renditions.get("v").unwrap().label.as_deref(),
			Some("Main camera"),
			"the label must survive the config the importer publishes up front"
		);

		let mut resolved = config(None, None);
		resolved.coded_width = Some(1280);
		resolved.coded_height = Some(720);
		hint.apply(&mut resolved);
		rendition.set(resolved).unwrap();
		feed(&mut rendition);

		let snapshot = catalog.snapshot();
		let published = snapshot.video.renditions.get("v").unwrap();
		assert_eq!(
			published.label.as_deref(),
			Some("Main camera"),
			"the label must survive a config the stream resolved later"
		);
		assert_eq!(published.bitrate, Some(456), "the rest of the config converts too");
	}

	/// A hint is applied by the importer before `set`, so a hinted field is indistinguishable from
	/// one the stream revealed: supplied, and never overwritten by detection.
	#[test]
	fn hinted_fields_count_as_supplied() {
		let (_broadcast, catalog, mut rendition) = video_track();

		let hint = VideoHint {
			bitrate: Some(456),
			framerate: Some(30.0),
			..Default::default()
		};
		let mut config = config(None, None);
		hint.apply(&mut config);

		rendition.set(config).unwrap();
		feed(&mut rendition);

		let snapshot = catalog.snapshot();
		let config = snapshot.video.renditions.get("v").unwrap();
		assert_eq!(config.bitrate, Some(456), "a hinted bitrate must not be overwritten");
		assert_eq!(
			config.framerate,
			Some(30.0),
			"a hinted framerate must not be overwritten"
		);
		assert!(config.jitter.is_some(), "the unhinted jitter should still be detected");
	}

	/// Frames flow before the config resolves (H.264 publishes until its first SPS), so a
	/// measurement taken while the rendition is still empty has to seed the config once it lands.
	#[test]
	fn measurements_before_set_seed_the_config() {
		let (_broadcast, catalog, mut rendition) = video_track();
		feed(&mut rendition);
		rendition.set(config(None, None)).unwrap();

		let snapshot = catalog.snapshot();
		let config = snapshot.video.renditions.get("v").unwrap();
		assert!(
			config.jitter.is_some(),
			"the jitter measured before `set` should carry over"
		);
		assert!(
			config.bitrate.is_some(),
			"the bitrate measured before `set` should carry over"
		);
	}

	/// A detected value the stream later reveals wins: `set` recaptures what the new config carries.
	#[test]
	fn resetting_recaptures_supplied() {
		let (_broadcast, catalog, mut rendition) = video_track();
		rendition.set(config(None, None)).unwrap();
		feed(&mut rendition);
		assert!(catalog.snapshot().video.renditions.get("v").unwrap().bitrate.is_some());

		rendition.set(config(Some(789), None)).unwrap();
		feed(&mut rendition);

		let snapshot = catalog.snapshot();
		let config = snapshot.video.renditions.get("v").unwrap();
		assert_eq!(config.bitrate, Some(789), "the re-set bitrate is now authoritative");
	}

	/// A rendition owns its name from the moment it's reserved, not from `set`. H.264 publishes
	/// frames before its first SPS resolves the config, and the name has to be unavailable through
	/// that window: an entry written into the gap used to be overwritten by `set` and then deleted
	/// by the importer's `Drop`, taking the caller's rendition with it.
	#[test]
	fn an_unresolved_rendition_still_owns_its_name() {
		let (_broadcast, catalog, mut rendition) = video_track();
		assert!(
			catalog.snapshot().video.renditions.is_empty(),
			"nothing is published until the config resolves"
		);

		let err = catalog
			.reserve()
			.init::<hang::catalog::VideoConfig>("v")
			.expect_err("an unresolved rendition still owns its name");
		assert!(matches!(err, crate::Error::Hang(hang::Error::Duplicate(name)) if name == "v"));

		// The refusal left nothing behind, so the importer's own config is what lands.
		rendition.set(config(Some(123), None)).unwrap();
		assert_eq!(
			catalog.snapshot().video.renditions.get("v").unwrap().bitrate,
			Some(123),
			"the importer's config owns the rendition"
		);
	}

	/// A resolved rendition still owns its name, now backed by a catalog entry.
	#[test]
	fn a_resolved_rendition_still_owns_its_name() {
		let (_broadcast, catalog, mut rendition) = video_track();
		rendition.set(config(Some(789), None)).unwrap();

		assert!(
			catalog.reserve().init::<hang::catalog::VideoConfig>("v").is_err(),
			"owned by the live rendition"
		);
		assert_eq!(
			catalog.snapshot().video.renditions.get("v").unwrap().bitrate,
			Some(789),
			"the refused reservation left the rendition intact"
		);
	}

	/// Ownership lasts exactly as long as the rendition, so the name is free again once it drops
	/// and the entry it published goes with it.
	#[test]
	fn dropping_a_rendition_frees_its_name() {
		let (_broadcast, catalog, mut rendition) = video_track();
		rendition.set(config(None, None)).unwrap();
		drop(rendition);
		assert!(
			catalog.snapshot().video.renditions.is_empty(),
			"the entry is retired with its owner"
		);

		let mut replacement = catalog
			.reserve()
			.init::<hang::catalog::VideoConfig>("v")
			.expect("the name is free once the rendition drops");
		replacement.set(config(Some(456), None)).unwrap();
		assert_eq!(catalog.snapshot().video.renditions.get("v").unwrap().bitrate, Some(456));
	}

	/// Sections are independent maps, so a video rendition says nothing about the same name in
	/// audio.
	#[test]
	fn a_name_is_owned_per_section() {
		let (_broadcast, catalog, _video) = video_track();
		catalog
			.reserve()
			.init::<hang::catalog::AudioConfig>("v")
			.expect("an audio rendition is a different section from a video one");
	}

	/// An entry already in the catalog is taken too, even with no rendition behind it: a caller
	/// that hand-wrote one through the guard still gets to keep it.
	#[test]
	fn an_existing_entry_owns_its_name() {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let mut catalog = super::super::Producer::new(&mut broadcast, super::super::Config::default()).unwrap();
		catalog.modify().unwrap().video.insert("v", config(None, None)).unwrap();

		assert!(
			catalog.reserve().init::<hang::catalog::VideoConfig>("v").is_err(),
			"the catalog already carries it"
		);
	}

	/// Each rendition gets its own HLS records, advertised in its own entry, and only when enabled.
	#[test]
	fn renditions_get_their_own_hls_records() {
		for enabled in [false, true] {
			let mut broadcast = moq_net::broadcast::Info::new().produce();
			let catalog_config = super::super::Config::default().with_hls(enabled.then(crate::hls::Config::default));
			let mut catalog = super::super::Producer::new(&mut broadcast, catalog_config).unwrap();

			let hint = Some(std::time::Duration::from_secs(4));
			let video0 = catalog.enroll("video0", hint).unwrap();
			let video1 = catalog.enroll("video1", hint).unwrap();
			assert_eq!(video0.is_some(), enabled);
			assert_eq!(video1.is_some(), enabled);
			{
				let mut guard = catalog.modify().unwrap();
				guard.video.insert("video0", config(None, None)).unwrap();
				guard.video.insert("video1", config(None, None)).unwrap();
			}

			let snapshot = catalog.snapshot();
			let hls = |name: &str| snapshot.video.renditions[name].hls.clone();
			match enabled {
				true => {
					let four = std::time::Duration::from_secs(4);
					assert_eq!(hls("video0"), Some(hang::catalog::Hls::new("video0.hls", four)));
					assert_eq!(hls("video1"), Some(hang::catalog::Hls::new("video1.hls", four)));
				}
				false => {
					assert_eq!(hls("video0"), None);
					assert_eq!(hls("video1"), None);
				}
			}
		}
	}

	mod custom {
		use std::collections::BTreeMap;

		use serde::{Deserialize, Serialize};

		use super::*;

		#[derive(Serialize, Deserialize, Clone, Default, Debug, PartialEq)]
		struct TelemetryExt {
			#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
			telemetry: BTreeMap<String, Telemetry>,
			#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
			exploding: BTreeMap<String, Exploding>,
			#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
			stubborn: BTreeMap<String, Stubborn>,
		}

		impl CatalogExt for TelemetryExt {}

		#[derive(Serialize, Deserialize, Clone, Default, Debug, PartialEq)]
		struct Telemetry {
			schema: String,
			#[serde(default, skip_serializing_if = "Option::is_none")]
			bitrate: Option<u64>,
		}

		impl RenditionConfig<TelemetryExt> for Telemetry {
			fn detects() -> bool {
				true
			}

			fn insert(self, catalog: &mut Catalog<TelemetryExt>, name: &str) {
				catalog.ext.telemetry.insert(name.to_string(), self);
			}
			fn get_mut<'a>(catalog: &'a mut Catalog<TelemetryExt>, name: &str) -> Option<&'a mut Self> {
				catalog.ext.telemetry.get_mut(name)
			}
			fn remove(catalog: &mut Catalog<TelemetryExt>, name: &str) {
				catalog.ext.telemetry.remove(name);
			}

			// Opts into bitrate detection only; jitter is left undetected.
			fn estimate(&self) -> Estimate {
				Estimate::default().with_bitrate(self.bitrate)
			}
			fn set_estimate(&mut self, estimate: Estimate) {
				self.bitrate = estimate.bitrate;
			}
		}

		fn telemetry(bitrate: Option<u64>) -> Telemetry {
			Telemetry {
				schema: "gps/v1".to_string(),
				bitrate,
			}
		}

		fn produce() -> (moq_net::broadcast::Producer, crate::catalog::Producer<TelemetryExt>) {
			let mut broadcast = moq_net::broadcast::Info::new().produce();
			let config = crate::catalog::Config::default().with_catalog(Catalog::<TelemetryExt>::default());
			let catalog = crate::catalog::Producer::new(&mut broadcast, config).unwrap();
			(broadcast, catalog)
		}

		/// A config whose `insert` panics, to poison the catalog lock the way any third-party
		/// [`RenditionConfig`] can: `set` runs it while holding that lock.
		#[derive(Serialize, Deserialize, Clone, Default, Debug, PartialEq)]
		struct Exploding {
			/// Write the entry before panicking, leaving a partially applied config behind.
			wrote: bool,
		}

		impl RenditionConfig<TelemetryExt> for Exploding {
			fn insert(self, catalog: &mut Catalog<TelemetryExt>, name: &str) {
				if self.wrote {
					catalog.ext.exploding.insert(name.to_string(), self);
				}
				panic!("insert exploded");
			}
			fn get_mut<'a>(catalog: &'a mut Catalog<TelemetryExt>, name: &str) -> Option<&'a mut Self> {
				catalog.ext.exploding.get_mut(name)
			}
			fn remove(catalog: &mut Catalog<TelemetryExt>, name: &str) {
				catalog.ext.exploding.remove(name);
			}
		}

		/// A panicking `insert` poisons the catalog lock while `set` holds it. The producer has to
		/// survive that: an unwinding `Rendition::drop` takes the same lock to release its name, and
		/// a panic there would be a panic during a panic, which aborts the process rather than
		/// failing a test.
		#[test]
		fn a_poisoned_lock_does_not_take_the_producer_down() {
			let (_broadcast, catalog) = produce();
			let reserved = catalog.reserve();

			let err = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
				let mut boom = reserved.init::<Exploding>("gps").unwrap();
				boom.set(Exploding { wrote: false }).unwrap();
			}))
			.expect_err("the config's insert panics");
			assert!(
				err.downcast_ref::<&str>().is_some_and(|msg| *msg == "insert exploded"),
				"the panic is the one the config raised, not a lock failure"
			);

			// The rendition unwound, so it released its name and the catalog is still usable.
			assert!(catalog.snapshot().ext.telemetry.is_empty());
			reserved
				.init::<Exploding>("gps")
				.expect("the unwound rendition released its name");
		}

		/// A config whose cleanup hook panics after mutating, the other half of untrusted caller code:
		/// `Drop` runs `remove` while holding the catalog lock.
		#[derive(Serialize, Deserialize, Clone, Default, Debug, PartialEq)]
		struct Stubborn;

		impl RenditionConfig<TelemetryExt> for Stubborn {
			fn insert(self, catalog: &mut Catalog<TelemetryExt>, name: &str) {
				catalog.ext.stubborn.insert(name.to_string(), self);
			}
			fn get_mut<'a>(catalog: &'a mut Catalog<TelemetryExt>, name: &str) -> Option<&'a mut Self> {
				catalog.ext.stubborn.get_mut(name)
			}
			fn remove(catalog: &mut Catalog<TelemetryExt>, name: &str) {
				catalog.ext.stubborn.remove(name);
				panic!("remove exploded");
			}
		}

		/// A panicking `remove` must not cost the rendition its name. The name is released before any
		/// caller code runs, so the slot is reservable again even though the hook blew up on the way
		/// out.
		#[test]
		fn a_panicking_remove_still_frees_the_name() {
			let (_broadcast, catalog) = produce();
			let reserved = catalog.reserve();

			let mut rendition = reserved.init::<Stubborn>("gps").unwrap();
			rendition.set(Stubborn).unwrap();

			std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| drop(rendition)))
				.expect_err("the config's remove panics");

			reserved
				.init::<Stubborn>("gps")
				.expect("a name held by a blown-up cleanup hook would be lost forever");
		}

		/// An insert that writes and then panics must leave no entry behind or strand its name.
		#[test]
		fn an_unwinding_rendition_retires_a_partially_written_entry() {
			let (_broadcast, catalog) = produce();
			let reserved = catalog.reserve();

			std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
				let mut boom = reserved.init::<Exploding>("gps").unwrap();
				boom.set(Exploding { wrote: true }).unwrap();
			}))
			.expect_err("the config's insert panics after writing");

			assert!(
				catalog.snapshot().ext.exploding.is_empty(),
				"the entry the panicking insert wrote is retired with its owner"
			);
			reserved
				.init::<Exploding>("gps")
				.expect("a stranded entry would refuse this forever");
		}

		/// A custom kind gets the same detection and drop-removal as video/audio.
		#[test]
		fn detects_and_advertises() {
			let (_broadcast, catalog) = produce();
			let reserved = catalog.reserve();
			let mut rendition = reserved.init::<Telemetry>("gps").unwrap();
			drop(reserved);

			rendition.set(telemetry(None)).unwrap();
			feed(&mut rendition);

			let snapshot = catalog.snapshot();
			let config = snapshot.ext.telemetry.get("gps").unwrap();
			assert!(config.bitrate.is_some(), "absent bitrate should be auto-detected");

			drop(rendition);
			assert!(
				!catalog.snapshot().ext.telemetry.contains_key("gps"),
				"the rendition should be removed on drop"
			);
		}

		/// The whole point of opting in: write a custom track through a
		/// [`container::Producer`](crate::container::Producer) and its catalog bitrate/jitter keep
		/// themselves current, with no per-frame bookkeeping in the caller.
		#[test]
		fn measures_a_custom_track() {
			let (broadcast, catalog) = produce();
			let reserved = catalog.reserve();
			let net = broadcast.create_track("gps", timed()).unwrap();
			let mut track = reserved
				.track(
					net,
					crate::catalog::hang::Container::Legacy(crate::container::Kind::Data),
					None,
				)
				.unwrap();
			drop(reserved);
			assert!(matches!(track.modify(), Err(crate::Error::NotPublished)));
			track.set(telemetry(None)).unwrap();

			// 40ms records of 5 kB: 1 Mbps, over more than the bitrate window.
			for i in 0..60u64 {
				track
					.write(crate::container::Frame {
						timestamp: ts(i * 40_000),
						duration: None,
						payload: bytes::Bytes::from(vec![0u8; 5_000]),
						keyframe: true,
					})
					.unwrap();
			}

			let snapshot = catalog.snapshot();
			let config = snapshot.ext.telemetry.get("gps").unwrap();
			assert_eq!(config.bitrate, Some(1_000_000));
		}

		/// A field the config supplies is authoritative even though the kind opts into detection.
		#[test]
		fn keeps_supplied_bitrate() {
			let (_broadcast, catalog) = produce();
			let reserved = catalog.reserve();
			let mut rendition = reserved.init::<Telemetry>("gps").unwrap();
			drop(reserved);

			rendition.set(telemetry(Some(4_200))).unwrap();
			feed(&mut rendition);

			let snapshot = catalog.snapshot();
			assert_eq!(snapshot.ext.telemetry.get("gps").unwrap().bitrate, Some(4_200));
		}

		/// Custom and media renditions share one reservation gate, so the first snapshot carries both.
		#[test]
		fn gates_with_media_tracks() {
			let (_broadcast, catalog) = produce();
			let mut consumer = catalog.consume().unwrap();

			let reserved = catalog.reserve();
			let mut video = reserved.init::<hang::catalog::VideoConfig>("v").unwrap();
			let mut gps = reserved.init::<Telemetry>("gps").unwrap();
			drop(reserved);

			let waiter = kio::Waiter::noop();
			video.set(config(None, None)).unwrap();
			assert!(
				matches!(consumer.poll_next(&waiter), std::task::Poll::Pending),
				"the catalog stays withheld while the telemetry rendition is unresolved"
			);

			gps.set(telemetry(None)).unwrap();

			let mut latest = None;
			while let std::task::Poll::Ready(Ok(Some(catalog))) = consumer.poll_next(&waiter) {
				latest = Some(catalog);
			}
			let published = latest.expect("catalog published");
			assert!(published.video.renditions.contains_key("v"));
			assert!(published.ext.telemetry.contains_key("gps"));
		}
	}
}