moq/

video.rs

//! Native video encode/decode via [`moq_video`].
//!
//! The video counterpart to [`audio`](crate::audio): publish raw pictures as an
//! encoded video track, and subscribe to one and hand back decoded raw frames,
//! with the codec running inside the FFI boundary (VideoToolbox on macOS, Media
//! Foundation on Windows, NVENC/NVDEC on Linux, openh264 as the software
//! fallback; no ffmpeg). Siblings to `moq_publish_media_*` /
//! `moq_consume_video`, which carry already-encoded frames for a caller that
//! brings its own codec.
//!
//! Decode is H.264 only; a non-H.264 rendition fails the subscribe with a
//! terminal error on the callback. Encode covers H.264 and H.265 (see
//! [`moq_video_codec`]).

use std::ffi::{c_char, c_void};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use tokio::sync::oneshot;

use crate::ffi::OnStatus;
use crate::{Error, Id, NonZeroSlab, Shared, State, ffi};

// ---- C-visible types ----

/// Pixel layout of the raw frames handed to [`moq_encode_video_frame`].
///
/// The enum is exposed in the C header for readability, but ABI fields that
/// carry it are typed `u32`. A C caller passing an unknown discriminant gets
/// `Error::InvalidCode` instead of UB.
#[repr(C)]
#[allow(non_camel_case_types)]
#[derive(Clone, Copy, Debug)]
pub enum moq_video_pixel_format {
	/// Tightly-packed planar I420: Y, then U, then V, no row padding.
	/// `width * height * 3 / 2` bytes, the same layout [`moq_decode_video`]
	/// hands back.
	MOQ_VIDEO_PIXEL_FORMAT_I420 = 0,
	/// Tightly-packed RGBA, `width * height * 4` bytes, no row padding.
	MOQ_VIDEO_PIXEL_FORMAT_RGBA = 1,
}

/// Output video codec for [`moq_encode_video`].
///
/// Not every codec has a backend on every machine: H.265 is hardware-only, so
/// publishing it fails where no hardware encoder is available.
#[repr(C)]
#[allow(non_camel_case_types)]
#[derive(Clone, Copy, Debug)]
pub enum moq_video_codec {
	/// H.264 / AVC, published as an `avc3` track.
	MOQ_VIDEO_CODEC_H264 = 0,
	/// H.265 / HEVC, published as a `hev1` track.
	MOQ_VIDEO_CODEC_H265 = 1,
}

/// Which encoder implementation [`moq_encode_video`] should use.
#[repr(C)]
#[allow(non_camel_case_types)]
#[derive(Clone, Copy, Debug)]
pub enum moq_video_encoder_kind {
	/// Prefer a platform hardware encoder, falling back to software.
	MOQ_VIDEO_ENCODER_KIND_AUTO = 0,
	/// Hardware only; fails if none is available.
	MOQ_VIDEO_ENCODER_KIND_HARDWARE = 1,
	/// Software only (openh264, H.264 only).
	MOQ_VIDEO_ENCODER_KIND_SOFTWARE = 2,
	/// A specific backend, named by `moq_video_encoder_output::encoder`.
	MOQ_VIDEO_ENCODER_KIND_NAMED = 3,
}

/// Raw frame layout the caller hands to [`moq_encode_video_frame`], plus
/// the resolution and rate the encoder is opened at. Every published frame must
/// match `width` x `height`; scale before publishing if your source moves.
#[repr(C)]
#[allow(non_camel_case_types)]
pub struct moq_video_encoder_input {
	/// `moq_video_pixel_format` discriminant.
	pub format: u32,
	/// Encoded width in pixels. Must be even (I420 chroma is subsampled 2x2).
	pub width: u32,
	/// Encoded height in pixels. Must be even.
	pub height: u32,
	/// Nominal frames per second, used for the codec time base and the default
	/// bitrate and keyframe interval. Must be non-zero.
	pub framerate: u32,
}

/// Codec-side configuration for [`moq_encode_video`]. Every knob spells
/// "unset" as 0.
#[repr(C)]
#[allow(non_camel_case_types)]
pub struct moq_video_encoder_output {
	/// `moq_video_codec` discriminant.
	pub codec: u32,
	/// Target bitrate in bits per second. 0 derives one from the resolution and
	/// framerate.
	pub bitrate: u64,
	/// Keyframe interval in frames: a subscriber joining mid-stream waits at
	/// most this many frames before it can decode. 0 uses ~2 seconds.
	pub gop: u32,
	/// `moq_video_encoder_kind` discriminant.
	pub kind: u32,
	/// Backend name, UTF-8, e.g. `"videotoolbox"`, `"nvenc"`, `"mediafoundation"`,
	/// `"openh264"`. Read only when `kind` is `MOQ_VIDEO_ENCODER_KIND_NAMED`.
	pub encoder: *const c_char,
	pub encoder_len: usize,
}

/// One raw frame handed to [`moq_encode_video_frame`].
///
/// Pixel format and resolution are fixed by [`moq_video_encoder_input`] at
/// publish time, so a frame carries neither: `data` is exactly one picture in
/// that layout, borrowed for the duration of the call (the encoder copies before
/// returning). The decode side has its own [`moq_video_frame`], which does carry
/// dimensions, since there they are what the stream turned out to be.
#[repr(C)]
#[allow(non_camel_case_types)]
pub struct moq_video_encoder_frame {
	/// Presentation timestamp, in microseconds.
	pub timestamp_us: u64,
	pub data: *const u8,
	pub data_size: usize,
}

/// Decode-side configuration the caller passes to [`moq_decode_video`].
///
/// `format` selects the CPU pixel layout of each [`moq_video_frame`] (`I420`
/// is `width * height * 3 / 2` bytes, `RGBA` is `width * height * 4` bytes),
/// and `width`/`height` select its size: zero both for the stream's native
/// size, otherwise both must be even and non-zero.
///
/// This struct is versioned by recompilation, not by reserved fields: adding a
/// field changes its layout, so rebuild callers against the `moq.h` that ships
/// with the `libmoq.a` they link.
#[repr(C)]
#[allow(non_camel_case_types)]
pub struct moq_video_decoder_output {
	/// Upper bound on buffering before skipping a stalled group, in
	/// microseconds. Same congestion-control knob as
	/// `moq_consume_video`'s `max_age_us`. 0 = skip aggressively
	/// (the moq-mux default); set to your playout buffer for a softer skip.
	pub max_age_us: u64,
	/// `moq_video_pixel_format` discriminant. Unknown values fail
	/// [`moq_decode_video`] rather than decoding into an assumed layout.
	pub format: u32,
	/// Target width in pixels. 0 with `height` 0 means the native size.
	pub width: u32,
	/// Target height in pixels. 0 with `width` 0 means the native size.
	pub height: u32,
}

/// One decoded video frame from [`moq_decode_video`]: pixels plus a
/// presentation timestamp.
///
/// The pixel layout is what [`moq_video_decoder_output`]'s `format` asked
/// for: I420 is the Y plane (`width * height`), then U, then V (`width/2 *
/// height/2` each), no row padding, BT.601 limited range; RGBA is tightly
/// packed `width * height * 4` bytes, no row padding.
///
/// `data` is owned by the consume slab and stays valid until the same id is
/// released with [`moq_decode_video_frame_free`].
///
/// The publish side has its own [`moq_video_encoder_frame`], which carries no
/// dimensions because the encoder already fixed them.
#[repr(C)]
#[allow(non_camel_case_types)]
pub struct moq_video_frame {
	pub timestamp_us: u64,
	pub width: u32,
	pub height: u32,
	pub data: *const u8,
	pub data_size: usize,
}

// ---- State extension (used internally by lib.rs) ----

/// Raw-video state: encoders being published, plus decoder tasks and their
/// buffered decoded frames.
#[derive(Default)]
pub struct Video {
	producers: NonZeroSlab<Shared<VideoEncoder>>,
	consumer_tasks: NonZeroSlab<Option<VideoTaskEntry>>,
	frames: NonZeroSlab<VideoFrame>,
}

/// Wait out an encode-thread round trip from a C entry point.
///
/// The C ABI hands back a status code, so there is no executor to yield to and
/// this is where [`Sink`](moq_video::encode::Sink)'s futures stop. Blocking is
/// also what paces the caller: a raw frame is megabytes, so a publish free to run
/// ahead of the codec would queue pictures without bound.
///
/// `pollster` rather than a tokio helper because those panic when the calling
/// thread is driving a runtime, which the one dispatching a callback is.
fn block_on<T>(future: impl std::future::Future<Output = T>) -> T {
	pollster::block_on(future)
}

fn video_ceiling(
	config: &moq_video::encode::Config,
	rendition: &hang::catalog::VideoConfig,
) -> moq_net::bandwidth::Rate {
	config
		.bitrate
		.or_else(|| rendition.bitrate.map(moq_net::bandwidth::Rate::from_bps))
		.unwrap_or_else(|| {
			moq_net::bandwidth::Rate::from_bps(
				(config.size().pixels() as f64 * config.framerate.as_f64() * 0.07) as u64,
			)
		})
}

async fn follow_reservation(
	inner: Shared<VideoEncoder>,
	mut consumer: moq_net::bandwidth::Consumer,
	ceiling: Arc<AtomicU64>,
) {
	use moq_mux::rate::{Control, Policy};

	let mut max = moq_net::bandwidth::Rate::from_bps(ceiling.load(Ordering::SeqCst));
	let mut control = Control::new(Policy::new(max));
	loop {
		let estimate = match consumer.changed().await {
			Ok(estimate) => estimate,
			Err(_) => return,
		};
		let next = moq_net::bandwidth::Rate::from_bps(ceiling.load(Ordering::SeqCst));
		if next != max {
			max = next;
			control = Control::new(Policy::new(max));
		}
		let Some(bitrate) = control.update(estimate, Instant::now()) else {
			continue;
		};

		let mut guard = inner.lock();
		let Some(producer) = guard.as_mut() else {
			return;
		};
		match block_on(producer.encoder.set_bitrate(bitrate)) {
			Ok(()) => tracing::debug!(bitrate = bitrate.as_bps(), "adjusted encoder bitrate"),
			Err(moq_video::Error::BitrateUnsupported(name)) => {
				tracing::warn!(encoder = name, "encoder cannot follow the bandwidth estimate");
				return;
			}
			Err(err) => {
				tracing::warn!(error = %err, bitrate = bitrate.as_bps(), "failed to adjust encoder bitrate");
			}
		}
	}
}

/// An encoder paired with the track publishing its output, plus the pixel format
/// its caller feeds it (fixed at publish time, so a frame carries only pixels and
/// a timestamp).
///
/// The encoder is a [`Sink`](moq_video::encode::Sink) rather than a bare
/// `Encoder` because a C caller drives a handle from whichever thread it likes,
/// so a bare `Encoder` would be built on one thread and dropped on another,
/// unbalancing the per-thread COM apartment the Windows backend opens. The sink
/// owns the thread instead, so every caller is welcome.
pub(crate) struct VideoEncoder {
	encoder: moq_video::encode::Sink,
	producer: moq_video::encode::Producer<moq_mux::catalog::hang::Extra>,
	format: moq_video_pixel_format,
	/// The encoded resolution, from the publish config. Frames carry only pixels,
	/// so this is what says how to read them.
	size: moq_video::Size,
	reservation: Option<Arc<moq_net::bandwidth::Reservation>>,
	follow: Option<oneshot::Sender<()>>,
	ceiling: Option<Arc<AtomicU64>>,
}

/// A delivered frame, flattened to CPU bytes at delivery time in the layout
/// [`moq_decode_video`] was asked for: the C ABI hands out a stable byte
/// pointer, so a GPU-decoded frame (e.g. NVDEC) is downloaded exactly once here.
struct VideoFrame {
	timestamp_us: u64,
	width: u32,
	height: u32,
	data: bytes::Bytes,
}

/// What [`moq_decode_video`] delivers per frame: the requested CPU pixel format
/// and target size, validated up front so the delivery loop never second-guesses.
#[derive(Clone, Copy)]
pub struct DecoderOutput {
	format: moq_video_pixel_format,
	size: Option<moq_video::Size>,
}

/// End a video track, given the result of draining its encoder into it.
///
/// A clean finish is a promise that the track holds everything the publisher
/// produced, so a lost tail has to end the track as an abort instead. Finishing
/// anyway would leave a truncated stream indistinguishable from a complete one,
/// and only the local caller would ever learn otherwise.
fn finalize(
	mut producer: moq_video::encode::Producer<moq_mux::catalog::hang::Extra>,
	drained: Result<(), moq_video::Error>,
) -> Result<(), Error> {
	match drained {
		Ok(()) => Ok(producer.finish()?),
		Err(err) => {
			producer.abort(moq_net::Error::Transport(err.to_string()));
			Err(err.into())
		}
	}
}

/// A spawned task entry: `close` signals shutdown, `callback` delivers status.
///
/// Same lifetime contract as the audio decoder: the task delivers one final
/// terminal callback and then removes itself, so `user_data` stays valid until
/// that callback fires. `close` is an `Option` so `consume_close` can drop just
/// the sender without removing the entry.
struct VideoTaskEntry {
	close: Option<oneshot::Sender<()>>,
	callback: OnStatus,
}

impl VideoEncoder {
	fn publish_frame(&mut self, timestamp_us: u64, data: &[u8]) -> Result<(), Error> {
		// A buffer that isn't one picture at the configured size is rejected here,
		// by the surface constructors, rather than reinterpreted.
		let size = self.size;
		let surface = match self.format {
			moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_I420 => {
				moq_video::Surface::I420(moq_video::I420::new(size, data.to_vec())?)
			}
			moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_RGBA => moq_video::Surface::rgba(data, size)?,
		};

		let frame = moq_video::Frame::new(surface, moq_net::Timestamp::from_micros(timestamp_us)?);
		// A backend that pipelines hands back an earlier frame's output, so this is
		// zero or more access units rather than one per call.
		let encoded = block_on(self.encoder.encode(frame))?;
		self.producer.publish(&encoded)?;
		Ok(())
	}

	fn publish_cut(&mut self) -> Result<(), Error> {
		// A keyframe is what a cut is on the wire: the importer closes the open
		// group and starts a new one at it.
		Ok(block_on(self.encoder.cut())?)
	}

	fn publish_bitrate(&mut self, bitrate: u64) -> Result<(), Error> {
		block_on(self.encoder.set_bitrate(moq_net::bandwidth::Rate::from_bps(bitrate)))?;
		// Ceiling first: `update` wakes the follower, which must not read the old
		// floor and retune above this cap before parking on an unchanged grant.
		if let Some(ceiling) = &self.ceiling {
			ceiling.store(bitrate, Ordering::SeqCst);
		}
		if let Some(reservation) = &self.reservation {
			reservation.update(moq_net::bandwidth::Rate::from_bps(bitrate));
		}
		Ok(())
	}

	fn publish_finish(self) -> Result<(), Error> {
		let VideoEncoder {
			encoder, mut producer, ..
		} = self;
		// Drain the codec into the track before ending it, so the last frames land
		// in it rather than being dropped with the encoder.
		let drained = block_on(encoder.finish()).and_then(|encoded| producer.publish(&encoded));
		finalize(producer, drained)
	}
}

impl Video {
	/// Advertise a track for an already-opened encoder.
	///
	/// The encoder and the rendition it will emit are both resolved by the caller,
	/// and before this, so a config this machine can't encode fails without leaving
	/// a track advertised that will never carry frames.
	pub fn publish(
		&mut self,
		broadcast: &moq_net::broadcast::Producer,
		catalog: moq_mux::catalog::Producer<moq_mux::catalog::hang::Extra>,
		format: moq_video_pixel_format,
		config: &moq_video::encode::Config,
		rendition: hang::catalog::VideoConfig,
		encoder: moq_video::encode::Sink,
	) -> Result<Id, Error> {
		let producer = moq_video::encode::Producer::new(broadcast.clone(), catalog, rendition)?;
		self.producers.insert(Shared::new(VideoEncoder {
			encoder,
			producer,
			format,
			size: config.size(),
			reservation: None,
			follow: None,
			ceiling: None,
		}))
	}

	pub(crate) fn follow(
		&self,
		id: Id,
		allocator: &moq_net::bandwidth::Allocator,
		ceiling: moq_net::bandwidth::Rate,
	) -> Result<(), Error> {
		let shared = self.producer(id)?;
		let max = ceiling;
		let ceiling = Arc::new(AtomicU64::new(max.as_bps()));
		let reservation = {
			let mut guard = shared.lock();
			let encoder = guard.as_mut().ok_or(Error::MediaNotFound)?;
			let reservation = Arc::new(allocator.reserve(&encoder.producer.demand(), max));
			encoder.reservation = Some(reservation.clone());
			encoder.ceiling = Some(ceiling.clone());
			reservation
		};
		let (close, closed) = oneshot::channel();
		shared.lock().as_mut().ok_or(Error::MediaNotFound)?.follow = Some(close);
		let follower = shared.clone();
		tokio::spawn(async move {
			tokio::select! {
				biased;
				_ = closed => {}
				_ = follow_reservation(follower, reservation.consumer(), ceiling) => {}
			}
		});
		Ok(())
	}

	pub(crate) fn reservation(&self, id: Id) -> Result<Option<Arc<moq_net::bandwidth::Reservation>>, Error> {
		Ok(self
			.producer(id)?
			.lock()
			.as_ref()
			.ok_or(Error::MediaNotFound)?
			.reservation
			.clone())
	}

	/// A watch-only handle to the encoded track's subscriber demand.
	pub(crate) fn demand(&self, id: Id) -> Result<moq_net::track::Demand, Error> {
		Ok(self
			.producer(id)?
			.lock()
			.as_ref()
			.ok_or(Error::MediaNotFound)?
			.producer
			.demand())
	}

	/// Resolve a producer handle, so the caller can encode with the global lock
	/// released.
	///
	/// Bind the result before locking it: a temporary [`State`] guard lives to the
	/// end of the statement that created it, so resolving and locking in one
	/// expression would put the encode back under the global lock.
	pub(crate) fn producer(&self, id: Id) -> Result<Shared<VideoEncoder>, Error> {
		self.producers.get(id).cloned().ok_or(Error::MediaNotFound)
	}

	/// Resolve a producer and drop its id, so nothing can be published to it after.
	pub(crate) fn remove(&mut self, id: Id) -> Result<Shared<VideoEncoder>, Error> {
		self.producers.remove(id).ok_or(Error::MediaNotFound)
	}

	pub fn consume(
		&mut self,
		broadcast: &moq_net::broadcast::Consumer,
		catalog: &hang::catalog::VideoConfig,
		name: &str,
		options: moq_video::decode::Options,
		output: DecoderOutput,
		on_frame: OnStatus,
	) -> Result<Id, Error> {
		let broadcast = broadcast.clone();
		let catalog = catalog.clone();
		let name = name.to_string();

		let channel = oneshot::channel();
		let entry = VideoTaskEntry {
			close: Some(channel.0),
			callback: on_frame,
		};
		let id = self.consumer_tasks.insert(Some(entry))?;

		// `Consumer::new` subscribes (blocking on SUBSCRIBE_OK), so run it inside
		// the task to keep this entrypoint non-blocking.
		tokio::spawn(async move {
			let res = async move {
				let consumer = moq_video::decode::Consumer::new(&broadcast, &catalog, name, options).await?;
				Self::run(on_frame, consumer, channel.1, output).await
			}
			.await;

			// Deliver one final terminal callback (code <= 0), then drop the entry.
			// Pull it out from under the lock so the callback never runs while held.
			let entry = State::lock().video.consumer_tasks.remove(id).flatten();
			if let Some(entry) = entry {
				entry.callback.call(res);
			}
		});

		Ok(id)
	}

	async fn run(
		callback: OnStatus,
		mut consumer: moq_video::decode::Consumer,
		mut close: oneshot::Receiver<()>,
		output: DecoderOutput,
	) -> Result<(), Error> {
		loop {
			// `biased` so a pending close always wins over a ready frame.
			let frame = tokio::select! {
				biased;
				_ = &mut close => return Ok(()),
				frame = consumer.read() => match frame? {
					Some(frame) => frame,
					None => return Ok(()),
				},
			};

			// The decoder's scale hint is best effort: a backend without a
			// scaler ignores it, so enforce the requested size here rather
			// than trusting it. The frame is already CPU pixels, so this
			// scales on the CPU; convert outside the lock, then hold the lock
			// only to buffer it, and release before the callback.
			let mut frame = frame;
			if let Some(size) = output.size
				&& frame.size() != size
			{
				frame = frame.resize(size, &moq_video::resize::Config::default())?;
			}
			let size = frame.size();
			let data = match output.format {
				moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_I420 => {
					bytes::Bytes::from(frame.surface.into_i420()?.into_data())
				}
				moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_RGBA => bytes::Bytes::from(
					frame
						.surface
						.to_rgba(&moq_video::convert::Config::default())?
						.into_data(),
				),
			};
			let frame = VideoFrame {
				// The C ABI carries microseconds; the decoded frame's Timestamp is
				// constrained to a QUIC VarInt, so the microsecond value fits a u64.
				timestamp_us: frame.timestamp.as_micros() as u64,
				width: size.width,
				height: size.height,
				data,
			};
			let frame_id = State::lock().video.frames.insert(frame)?;
			callback.call(Ok(frame_id));
		}
	}

	pub fn consume_close(&mut self, id: Id) -> Result<(), Error> {
		// Signal shutdown; the task delivers a final callback and removes itself.
		self.consumer_tasks
			.get_mut(id)
			.and_then(|entry| entry.as_mut())
			.ok_or(Error::TrackNotFound)?
			.close
			.take()
			.ok_or(Error::TrackNotFound)?;
		Ok(())
	}

	pub fn frame_info(&self, id: Id, dst: &mut moq_video_frame) -> Result<(), Error> {
		let frame = self.frames.get(id).ok_or(Error::FrameNotFound)?;
		*dst = moq_video_frame {
			timestamp_us: frame.timestamp_us,
			width: frame.width,
			height: frame.height,
			data: frame.data.as_ptr(),
			data_size: frame.data.len(),
		};
		Ok(())
	}

	pub fn frame_free(&mut self, id: Id) -> Result<(), Error> {
		self.frames.remove(id).ok_or(Error::FrameNotFound)?;
		Ok(())
	}
}

// ---- C entry points ----

fn pixel_format_from_u32(value: u32) -> Result<moq_video_pixel_format, Error> {
	Ok(match value {
		v if v == moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_I420 as u32 => {
			moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_I420
		}
		v if v == moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_RGBA as u32 => {
			moq_video_pixel_format::MOQ_VIDEO_PIXEL_FORMAT_RGBA
		}
		_ => return Err(Error::InvalidCode),
	})
}

/// Parse [`moq_video_decoder_output`]'s `width`/`height` into the target size:
/// 0x0 is the stream's native size, anything else must be even and non-zero
/// (I420 chroma is subsampled 2x2) and small enough for the packed byte math.
fn decoder_size(width: u32, height: u32) -> Result<Option<moq_video::Size>, Error> {
	if width == 0 && height == 0 {
		return Ok(None);
	}
	if width == 0 || height == 0 || !width.is_multiple_of(2) || !height.is_multiple_of(2) {
		return Err(Error::InvalidConfig(format!(
			"decode size {width}x{height}: use 0x0 for the native size or even non-zero dimensions"
		)));
	}
	let size = moq_video::Size::new(width, height);
	if size
		.pixels()
		.checked_mul(4)
		.is_none_or(|bytes| usize::try_from(bytes).is_err())
	{
		return Err(Error::InvalidConfig(format!(
			"decode size {width}x{height}: dimensions too large to represent"
		)));
	}
	Ok(Some(size))
}

fn codec_from_u32(value: u32) -> Result<moq_video::encode::Codec, Error> {
	use moq_video::encode::Codec;
	Ok(match value {
		v if v == moq_video_codec::MOQ_VIDEO_CODEC_H264 as u32 => Codec::H264,
		v if v == moq_video_codec::MOQ_VIDEO_CODEC_H265 as u32 => Codec::H265,
		_ => return Err(Error::InvalidCode),
	})
}

/// # Safety
/// - `output->encoder` must point to `output->encoder_len` bytes of UTF-8 when
///   `output->kind` is `MOQ_VIDEO_ENCODER_KIND_NAMED`.
unsafe fn encoder_kind(output: &moq_video_encoder_output) -> Result<moq_video::encode::Kind, Error> {
	use moq_video::encode::Kind;
	Ok(match output.kind {
		v if v == moq_video_encoder_kind::MOQ_VIDEO_ENCODER_KIND_AUTO as u32 => Kind::Auto,
		v if v == moq_video_encoder_kind::MOQ_VIDEO_ENCODER_KIND_HARDWARE as u32 => Kind::Hardware,
		v if v == moq_video_encoder_kind::MOQ_VIDEO_ENCODER_KIND_SOFTWARE as u32 => Kind::Software,
		v if v == moq_video_encoder_kind::MOQ_VIDEO_ENCODER_KIND_NAMED as u32 => {
			Kind::Named(unsafe { ffi::parse_str(output.encoder, output.encoder_len)? }.to_string())
		}
		_ => return Err(Error::InvalidCode),
	})
}

/// Open a video track on a broadcast, encoding the raw frames you publish to it.
///
/// The encoder is opened here, so an unsupported codec, resolution, or backend
/// fails now rather than on the first frame. The track is named after the codec
/// (`.avc3` / `.hev1`) and its catalog rendition is published immediately, read
/// out of the encoder rather than guessed, so a subscriber can find the track
/// before a frame is written to it.
///
/// Returns a non-zero handle on success or a negative error code.
///
/// # Safety
/// - `input` / `output` must point to fully populated structs.
/// - `output->encoder` must point to `output->encoder_len` bytes of UTF-8 when
///   `output->kind` is `MOQ_VIDEO_ENCODER_KIND_NAMED`.
/// - `bandwidth` is a handle from [`crate::moq_session_bandwidth`], or 0 to hold the
///   configured bitrate regardless of congestion.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn moq_encode_video(
	broadcast: u32,
	input: *const moq_video_encoder_input,
	output: *const moq_video_encoder_output,
	bandwidth: u32,
) -> i32 {
	ffi::enter(move || {
		let broadcast = ffi::parse_id(broadcast)?;
		let raw_input = unsafe { input.as_ref() }.ok_or(Error::InvalidPointer)?;
		let raw_output = unsafe { output.as_ref() }.ok_or(Error::InvalidPointer)?;

		let format = pixel_format_from_u32(raw_input.format)?;

		let framerate = moq_video::Rate::new(raw_input.framerate, 1)
			.map_err(|_| Error::Video(moq_video::Error::InvalidFramerate(raw_input.framerate).into()))?;
		let mut config = moq_video::encode::Config::new(raw_input.width, raw_input.height, framerate);
		config.codec = codec_from_u32(raw_output.codec)?;
		config.kind = unsafe { encoder_kind(raw_output)? };
		// The C ABI spells an unset knob as 0, which neither field accepts as a real
		// value: a zero bitrate or GOP is the default, not a request.
		config.bitrate = (raw_output.bitrate != 0).then(|| moq_net::bandwidth::Rate::from_bps(raw_output.bitrate));
		if raw_output.gop != 0 {
			config.gop = moq_video::encode::Gop::Keyframe {
				interval: raw_output.gop,
			};
		}

		// Both before the global lock is taken: bringing up a hardware encoder is slow
		// enough that every other call would wait behind it. The probe runs first and
		// closes its encoder before this one opens, so only one codec session is live.
		let rendition = block_on(config.probe())?;
		let encoder = block_on(moq_video::encode::Sink::open(&config))?;

		let bandwidth = ffi::parse_id_optional(bandwidth)?;
		let mut state = State::lock();
		let allocator = bandwidth.map(|id| state.bandwidth.allocator(id)).transpose()?;
		let State { publish, video, .. } = &mut *state;
		let (broadcast_producer, catalog) = publish.pair_mut(broadcast)?;

		let id = video.publish(
			broadcast_producer,
			catalog.clone(),
			format,
			&config,
			rendition.clone(),
			encoder,
		)?;
		if let Some(allocator) = allocator.as_ref() {
			video.follow(id, allocator, video_ceiling(&config, &rendition))?;
		}
		Ok(id)
	})
}

/// This encoder's bandwidth reservation, or 0 if it was published without one.
///
/// Closing the returned handle does not release the encoder's claim; that lasts
/// until [moq_encode_video_finish].
#[unsafe(no_mangle)]
pub extern "C" fn moq_encode_video_reservation(producer: u32) -> i32 {
	ffi::enter(move || {
		let producer = ffi::parse_id(producer)?;
		let mut state = State::lock();
		match state.video.reservation(producer)? {
			Some(reservation) => Ok(i32::from(state.bandwidth.hold(reservation)?)),
			None => Ok(0),
		}
	})
}

/// Watch whether the encoded video track has subscribers, so the camera and encoder
/// run only while someone watches. See [`crate::moq_publish_media_demand`] for the
/// callback contract.
///
/// Returns a non-zero watcher handle on success, or a negative code on failure.
///
/// # Safety
/// - `on_demand` must be non-NULL.
/// - The caller must keep `user_data` valid until the terminal (`<= 0`) `on_demand` callback.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn moq_encode_video_demand(
	producer: u32,
	on_demand: crate::moq_status_callback,
	user_data: *mut c_void,
) -> i32 {
	ffi::enter(move || {
		let producer = ffi::parse_id(producer)?;
		let on_demand = unsafe { OnStatus::new(user_data, on_demand)? };
		let mut state = State::lock();
		let demand = state.video.demand(producer)?;
		state.publish.demand(demand, on_demand)
	})
}

/// Encode and publish one raw frame.
///
/// `frame->data` is borrowed for the duration of the call and must be exactly one
/// picture in the pixel format and at the resolution declared by
/// [`moq_video_encoder_input`].
/// A backend that pipelines publishes an earlier frame's output here, so a call
/// that emits nothing is normal rather than an error.
///
/// # Safety
/// - `frame` must point to a valid [`moq_video_encoder_frame`].
/// - `frame->data` must point to `frame->data_size` bytes.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn moq_encode_video_frame(producer: u32, frame: *const moq_video_encoder_frame) -> i32 {
	ffi::enter(move || {
		let producer = ffi::parse_id(producer)?;
		let frame = unsafe { frame.as_ref() }.ok_or(Error::InvalidPointer)?;
		let data = unsafe { ffi::parse_slice(frame.data, frame.data_size)? };

		let producer = State::lock().video.producer(producer)?;
		producer
			.lock()
			.as_mut()
			.ok_or(Error::MediaNotFound)?
			.publish_frame(frame.timestamp_us, data)
	})
}

/// Cut a new group at the next published frame.
///
/// Optional. The encoder already keyframes every `moq_video_encoder_output.gop`
/// frames, and each of those cuts a group, so a subscriber can always join
/// without you calling this. Reach for it only to place the boundaries yourself:
/// aligning groups with something the encoder cannot see, such as a scene change,
/// a source switch, or resuming after an idle gap.
///
/// The next frame is encoded as a keyframe, which closes the open group and
/// starts a new one at it. Calling this repeatedly before that frame arrives cuts
/// once, not several times.
///
/// Fails when the selected encoder cannot force a keyframe (a V4L2 driver
/// without the control): nothing is queued, and groups keep falling every
/// `gop` frames.
#[unsafe(no_mangle)]
pub extern "C" fn moq_encode_video_cut(producer: u32) -> i32 {
	ffi::enter(move || {
		let producer = ffi::parse_id(producer)?;
		let producer = State::lock().video.producer(producer)?;
		producer.lock().as_mut().ok_or(Error::MediaNotFound)?.publish_cut()
	})
}

/// Retune a live encoder to `bitrate` bits per second, taking effect from
/// roughly the next frame. No keyframe is forced, so this is cheap enough to
/// drive from a congestion controller.
///
/// The configured bitrate is a ceiling on some backends (openh264 rejects a raise
/// above the rate it opened at), so set `bitrate` to the highest you will ask
/// for and adapt downwards from there.
///
/// When this encoder was published against a bandwidth allocator, the reservation
/// and follower ceiling move with it, so a later grant cannot retune above this
/// value.
///
/// Returns a negative code if this backend cannot retune while running. That is
/// not fatal: the encoder keeps running at its current rate, so stop adapting
/// rather than stop publishing.
#[unsafe(no_mangle)]
pub extern "C" fn moq_encode_video_bitrate(producer: u32, bitrate: u64) -> i32 {
	ffi::enter(move || {
		let producer = ffi::parse_id(producer)?;
		let producer = State::lock().video.producer(producer)?;
		producer
			.lock()
			.as_mut()
			.ok_or(Error::MediaNotFound)?
			.publish_bitrate(bitrate)
	})
}

/// Flush any frames the codec is still holding and finalize the video track.
///
/// The handle is released, so nothing can be published to it afterwards.
#[unsafe(no_mangle)]
pub extern "C" fn moq_encode_video_finish(producer: u32) -> i32 {
	ffi::enter(move || {
		let producer = ffi::parse_id(producer)?;
		// The id is dropped first, so nothing new queues behind the drain; whatever
		// is mid-encode still finishes before this takes the encoder.
		let producer = State::lock().video.remove(producer)?;
		producer.take().ok_or(Error::MediaNotFound)?.publish_finish()
	})
}

/// Subscribe to a video track and decode it into raw frames in the requested
/// CPU pixel format and size (see [`moq_video_decoder_output`]).
///
/// The catalog `index` selects which video rendition to subscribe to, matching
/// the existing `moq_consume_video` selection model. Only H.264 is
/// supported; a non-H.264 rendition fails on the terminal callback.
///
/// An unknown `output->format` or an invalid `output->width`/`height` fails
/// here, before subscribing: an accepted request always produces the requested
/// layout or fails on the terminal callback instead of delivering it silently.
///
/// Returns a non-zero handle on success or a negative error code.
///
/// `on_frame` is called with a positive frame id per decoded frame, then exactly
/// once more with a terminal code: `0` (closed cleanly) or a negative error.
/// After the terminal (`<= 0`) callback, `on_frame` is never called again and
/// `user_data` is never touched again, so release `user_data` there. The terminal
/// callback fires even after [`moq_decode_video_cancel`].
///
/// Starts at the newest cached group so reopening live playback skips the backlog.
///
/// # Safety
/// - `output` must point to a valid [`moq_video_decoder_output`].
/// - `user_data` must stay valid until the terminal (`<= 0`) `on_frame` callback.
#[unsafe(no_mangle)]
pub unsafe extern "C" fn moq_decode_video(
	catalog: u32,
	index: u32,
	output: *const moq_video_decoder_output,
	on_frame: crate::moq_status_callback,
	user_data: *mut c_void,
) -> i32 {
	ffi::enter(move || {
		let raw = unsafe { output.as_ref() }.ok_or(Error::InvalidPointer)?;

		// Validate the request before resolving anything: a C caller can probe
		// support without a live track, and a bad request never opens a
		// subscription it would immediately drop.
		let format = pixel_format_from_u32(raw.format)?;
		let size = decoder_size(raw.width, raw.height)?;
		let catalog = ffi::parse_id(catalog)?;

		let mut options = moq_video::decode::Options::new();
		options.start = moq_video::decode::Start::Latest;
		options.max_age = Duration::from_micros(raw.max_age_us);
		// The C caller takes packed pixels, so let a backend that can decode
		// straight to the CPU do that rather than downloading afterwards.
		options.decoder.output = moq_video::Output::Cpu;
		// A backend with a hardware scaler (NVDEC) honors this for free; the
		// delivery loop still enforces it, since other backends ignore it.
		options.decoder.scale_hint = size;
		let output = DecoderOutput { format, size };
		let on_frame = unsafe { OnStatus::new(user_data, on_frame)? };

		let mut state = State::lock();
		let (broadcast, video_cfg, name) = state.consume.video_rendition(catalog, index as usize)?;

		let State { video, .. } = &mut *state;
		video.consume(&broadcast, &video_cfg, &name, options, output, on_frame)
	})
}

/// Stop a video (raw) consumer's background task.
///
/// Returns immediately: zero on success, or a negative code if already closed.
/// Does NOT free `user_data`; the on-frame callback still fires once more with a
/// terminal `0` (or a negative error), which is where `user_data` should be
/// released. Frame ids already delivered are likewise not freed; release each
/// with [`moq_decode_video_frame_free`].
#[unsafe(no_mangle)]
pub extern "C" fn moq_decode_video_cancel(consumer: u32) -> i32 {
	ffi::enter(move || {
		let consumer = ffi::parse_id(consumer)?;
		State::lock().video.consume_close(consumer)
	})
}

/// Copy a delivered frame's metadata into `dst`.
///
/// The written `dst->data` pointer remains valid until the same `id` is released
/// with [`moq_decode_video_frame_free`].
///
/// # Safety
/// - `dst` must point to a writable [`moq_video_frame`].
#[unsafe(no_mangle)]
pub unsafe extern "C" fn moq_decode_video_frame(id: u32, dst: *mut moq_video_frame) -> i32 {
	ffi::enter(move || {
		let id = ffi::parse_id(id)?;
		let dst = unsafe { dst.as_mut() }.ok_or(Error::InvalidPointer)?;
		State::lock().video.frame_info(id, dst)
	})
}

/// Free a frame previously delivered through the consume callback. Required for
/// every delivered frame id; closing the parent consumer is not enough.
#[unsafe(no_mangle)]
pub extern "C" fn moq_decode_video_frame_free(id: u32) -> i32 {
	ffi::enter(move || {
		let id = ffi::parse_id(id)?;
		State::lock().video.frame_free(id)
	})
}
#[cfg(test)]
mod tests {
	use super::*;

	/// A video track wired up without an encoder, plus a subscriber on it: enough
	/// to pin what [`finalize`] shows the far end.
	async fn track_under_test() -> (
		moq_video::encode::Producer<moq_mux::catalog::hang::Extra>,
		moq_net::track::Subscriber,
	) {
		let mut broadcast = moq_net::broadcast::Info::new().produce();
		let config = moq_mux::catalog::Config::default()
			.with_catalog(moq_mux::catalog::hang::Catalog::<moq_mux::catalog::hang::Extra>::default());
		let catalog = moq_mux::catalog::Producer::new(&mut broadcast, config).unwrap();
		let consumer = broadcast.consume();
		// Probed rather than hand-built, so the test track carries what a real one would.
		let rendition = moq_video::encode::Config::new(320, 240, moq_video::Rate::new(30, 1).unwrap())
			.probe()
			.await
			.unwrap();
		let producer = moq_video::encode::Producer::new(broadcast, catalog, rendition).unwrap();

		let name = producer.demand().name().to_string();
		let track = consumer.track(&name).unwrap().subscribe(None).await.unwrap();
		(producer, track)
	}

	/// A clean finish reaches the subscriber as the end of the track, which is what
	/// makes the abort case below meaningful rather than vacuous.
	#[tokio::test]
	async fn a_successful_drain_ends_the_track_cleanly() {
		let (producer, mut track) = track_under_test().await;
		finalize(producer, Ok(())).unwrap();
		assert!(matches!(track.recv_group().await, Ok(None)), "expected a clean end");
	}

	/// Regression: a lost tail must reach the subscriber as an abort. Finishing the
	/// track anyway would report a truncated stream as a complete one, and only the
	/// publisher would ever know otherwise.
	#[tokio::test]
	async fn a_failed_drain_aborts_the_track() {
		let (producer, mut track) = track_under_test().await;
		let err = moq_video::Error::Codec(anyhow::anyhow!("the codec lost the tail"));
		finalize(producer, Err(err)).unwrap_err();

		let Err(err) = track.recv_group().await else {
			panic!("expected an abort, not a clean end");
		};
		assert!(
			err.to_string().contains("the codec lost the tail"),
			"the abort should carry the drain failure: {err}"
		);
	}
}