autumn_media_plugin/

lib.rs

//! # autumn-media-plugin
//!
//! Live-streaming media plugin for `autumn-web` applications. It packages the
//! two primitives an interactive streaming product needs:
//!
//! - **Broadcast** — one creator ingests (RTMP / WHIP / browser WebRTC) and a
//!   fan-out audience watches over low-latency WebRTC/WHEP with HLS fallback,
//!   backed by `MediaMTX`; recordings become VODs and clip/highlight encodes.
//! - **Room** — a small (mesh, no SFU) multi-participant call capped at
//!   [`config::DEFAULT_ROOM_MAX_PARTICIPANTS`] participants.
//!
//! Both primitives are off by default. Enable each one on the builder:
//!
//! | Builder call | Installs |
//! |---|---|
//! | [`with_broadcast`](MediaPlugin::with_broadcast) | [`MediaMtxClient`] and [`MediaUrls`] extensions |
//! | [`with_rooms`](MediaPlugin::with_rooms) | Room routes, [`RoomService`] and the room reaper |
//! | Either one | [`MediaStorage`], [`MediaWorkflows`] and the encode jobs |
//!
//! The retention sweep also starts if you set
//! [`recordings_root`](MediaPlugin::recordings_root). If the storage config is
//! not valid, the plugin installs no storage, workflows, jobs or sweep.
//!
//! If you enable neither primitive, the plugin installs no routes, extensions or
//! jobs. It logs a warning.
//!
//! Because Autumn resolves config *after* [`Plugin::build`] runs, the plugin
//! cannot read `[media]` from inside `build`; the application loads a
//! [`MediaConfig`](config::MediaConfig) up front and passes it in via
//! [`MediaPlugin::config`] (mirroring the `autumn-storage-s3`
//! `from_config(&cfg) -> …with_blob_store(store)` pattern).
//!
//! ```rust,ignore
//! use autumn_media_plugin::{prelude::*, MediaPlugin};
//!
//! let media = MediaConfig::from_autumn_toml("autumn.toml")?;
//! autumn_web::app()
//!     .plugin(MediaPlugin::new().config(media).with_broadcast())
//!     .run()
//!     .await;
//! ```
//!
//! # Naming convention
//!
//! First-party plugin: `autumn-<name>-plugin`.

pub mod config;
pub mod encode;
pub mod error;
pub mod retention;
pub mod rooms;
pub mod rooms_db;
pub mod sink;
pub mod storage;
pub mod transport;
pub mod workflows;

pub use config::{
    MediaConfig, MediaConfigError, MediaMtxConfig, MediaStorageBackend, MediaStorageConfig,
    RecordingConfig, RoomStoreBackend,
};
pub use encode::{
    FfmpegClipTailCommand, FfmpegHighlightCommand, FfmpegLiveThumbnailCommand, FfmpegPosterCommand,
    FfmpegPreviewSpriteCommand, FfmpegRoomCompositeCommand, PREVIEW_CELL_HEIGHT,
    PREVIEW_CELL_WIDTH, PREVIEW_FRAME_INTERVAL_SECONDS, PREVIEW_SPRITE_COLUMNS,
    ROOM_COMPOSITE_CELL_HEIGHT, ROOM_COMPOSITE_CELL_WIDTH, build_preview_webvtt,
    newest_recording_file, newest_recording_files, newest_recording_files_since,
    recording_segments_covering_window, slugify,
};
pub use error::MediaError;
pub use retention::{
    RetentionDefer, RetentionReport, is_expired, recording_expires_at, spawn_retention_sweep_loop,
    sweep_recordings_root, within_root,
};
pub use rooms::{
    HeartbeatRequest, HeartbeatResponse, InMemoryRoomStore, JoinRecord, JoinRequest, JoinResponse,
    LeaveRequest, ParticipantView, PublishTarget, ReapFuture, ReapStats, RoomError,
    RoomLeaveResponse, RoomService, RoomSnapshot, RoomStore, RoomStoreFuture, SessionToken,
    SubscribeTarget, renewed_expiry, room_participant_path, room_route_infos, room_router,
    spawn_room_reaper_loop, validate_room_segment,
};
pub use rooms_db::DbRoomStore;
pub use sink::{
    MediaArtifact, MediaArtifactFile, MediaArtifactKind, MediaArtifactSink, MediaArtifactSinkExt,
    MediaSinkFuture,
};
pub use storage::{MediaStorage, S3MediaStorage, StoredObject};
pub use transport::{
    IngestStatus, MediaMtxClient, MediaUrls, StreamQualityStats, StreamStatus, ViewerCount,
    duration_seconds_param, ingest_statuses_from_paths_json, quality_stats_from_path_json,
    recording_available, recording_mediamtx_path, stream_status_from_path_json,
    viewer_count_from_path_json, viewer_counts_from_paths_json,
};
pub use workflows::{
    FinalizeRecordingJobArgs, MediaWorkflowDelegate, MediaWorkflowDelegateExt,
    MediaWorkflowRequest, MediaWorkflows, PreviewJobArgs, RoomCompositeJobArgs, ThumbnailJobArgs,
    TranscodeJobArgs, media_job_infos,
};

/// Common downstream imports for configuring and mounting the media plugin.
pub mod prelude {
    pub use crate::{
        FfmpegClipTailCommand, FfmpegHighlightCommand, FfmpegLiveThumbnailCommand,
        FfmpegPosterCommand, FfmpegPreviewSpriteCommand, FfmpegRoomCompositeCommand,
        FinalizeRecordingJobArgs, MediaArtifact, MediaArtifactFile, MediaArtifactKind,
        MediaArtifactSink, MediaArtifactSinkExt, MediaConfig, MediaConfigError, MediaError,
        MediaMtxConfig, MediaPlugin, MediaStorage, MediaStorageBackend, MediaStorageConfig,
        MediaWorkflowDelegate, MediaWorkflowDelegateExt, MediaWorkflowRequest, MediaWorkflows,
        PREVIEW_CELL_HEIGHT, PREVIEW_CELL_WIDTH, PREVIEW_FRAME_INTERVAL_SECONDS,
        PREVIEW_SPRITE_COLUMNS, PreviewJobArgs, ROOM_COMPOSITE_CELL_HEIGHT,
        ROOM_COMPOSITE_CELL_WIDTH, RecordingConfig, RetentionDefer, RoomCompositeJobArgs,
        S3MediaStorage, StoredObject, ThumbnailJobArgs, TranscodeJobArgs, build_preview_webvtt,
        media_job_infos, newest_recording_file, newest_recording_files,
        newest_recording_files_since, recording_segments_covering_window, slugify,
    };
    pub use crate::{
        HeartbeatResponse, InMemoryRoomStore, JoinRecord, JoinResponse, ParticipantView, ReapStats,
        RoomError, RoomService, RoomSnapshot, RoomStore, SessionToken, renewed_expiry,
        room_participant_path, room_route_infos, room_router, spawn_room_reaper_loop,
        validate_room_segment,
    };
    pub use crate::{
        IngestStatus, MediaMtxClient, MediaUrls, StreamQualityStats, StreamStatus, ViewerCount,
        duration_seconds_param, ingest_statuses_from_paths_json, quality_stats_from_path_json,
        recording_available, recording_mediamtx_path, stream_status_from_path_json,
        viewer_count_from_path_json, viewer_counts_from_paths_json,
    };
}

use std::borrow::Cow;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;

use autumn_web::app::AppBuilder;
use autumn_web::plugin::Plugin;

use crate::config::DEFAULT_ROOM_MAX_PARTICIPANTS;
use crate::retention::RetentionDefer as RetentionDeferHook;
use crate::sink::MediaArtifactSink as MediaArtifactSinkTrait;
use crate::workflows::MediaWorkflowDelegate as MediaWorkflowDelegateHook;

/// The live-streaming media plugin.
///
/// Configure the two primitives with [`with_broadcast`](Self::with_broadcast)
/// and [`with_rooms`](Self::with_rooms), supply a
/// [`MediaConfig`](config::MediaConfig) with [`config`](Self::config), then
/// install with `app.plugin(...)`.
///
/// See the crate docs for what each primitive installs.
pub struct MediaPlugin {
    /// Resolved `[media]` configuration.
    config: MediaConfig,
    /// Whether the broadcast primitive is enabled.
    enable_broadcast: bool,
    /// Whether the rooms primitive is enabled.
    enable_rooms: bool,
    /// Hard cap on mesh-room participants.
    room_max_participants: usize,
    /// Job queue name the built-in media encode jobs are registered on.
    queue: String,
    /// URL prefix for the plugin's API routes.
    api_prefix: String,
    /// App-supplied artifact completion callback (parallels the
    /// `OutboundWebhookHandler` store).
    artifact_sink: Option<Arc<dyn MediaArtifactSinkTrait>>,
    /// Optional runtime delegate that overrides the built-in `#[job]` engine
    /// (e.g. an external Harvest adapter).
    workflow_delegate: Option<MediaWorkflowDelegateHook>,
    /// Retention window override (days). `None` uses `config.recording.retention_days`.
    retention_days: Option<u32>,
    /// Filesystem root the retention sweep operates on. `None` disables the
    /// sweep (there is nothing to sweep without a recordings root).
    recordings_root: Option<PathBuf>,
    /// App-overridable retention-defer predicate.
    retention_defer: Option<RetentionDeferHook>,
}

impl MediaPlugin {
    /// Create a media plugin with default configuration and both primitives
    /// disabled (opt in with [`with_broadcast`](Self::with_broadcast) /
    /// [`with_rooms`](Self::with_rooms)).
    #[must_use]
    pub fn new() -> Self {
        Self {
            config: MediaConfig::default(),
            enable_broadcast: false,
            enable_rooms: false,
            room_max_participants: DEFAULT_ROOM_MAX_PARTICIPANTS,
            queue: "media".to_owned(),
            api_prefix: "/api/media".to_owned(),
            artifact_sink: None,
            workflow_delegate: None,
            retention_days: None,
            recordings_root: None,
            retention_defer: None,
        }
    }

    /// Build a fully-wired broadcast [`MediaPlugin`] from an Arroyo operator's
    /// existing `ARROYO_*` environment — the ratified migration shim (issue
    /// #1974, slice 5). One call maps the operator's env onto a plugin ready to
    /// install, so adopting `autumn-media-plugin` changes no ops config:
    ///
    /// - the `[media]` config surface — `MediaMTX` origins, `FFmpeg` bin, the
    ///   local/S3 storage selection (including Arroyo's Tigris `auto` region /
    ///   `t3.storage.dev` endpoint / `highlights` key-prefix defaults), and the
    ///   retention window — via
    ///   [`MediaConfig::from_arroyo_env`](config::MediaConfig::from_arroyo_env);
    /// - the **broadcast** primitive enabled (Arroyo is one-to-many and uses no
    ///   rooms), so the plugin is watch-path-ready without a further builder
    ///   call;
    /// - the retention sweep's recordings root from `ARROYO_RECORDINGS_ROOT`
    ///   (default `recordings`, matching Arroyo's own
    ///   `configured_recordings_root()` fallback), so the hourly retention loop
    ///   is wired exactly as before.
    ///
    /// The returned value is a normal builder: chain
    /// [`artifact_sink`](Self::artifact_sink) (the app writes its own tables on
    /// completion), [`workflow_delegate`](Self::workflow_delegate),
    /// [`retention_defer`](Self::retention_defer), etc. This reads the process
    /// environment; the pure mapping lives in
    /// [`from_arroyo_env_pairs`](Self::from_arroyo_env_pairs) for testability.
    #[must_use]
    pub fn from_arroyo_env() -> Self {
        let env: HashMap<String, String> = std::env::vars().collect();
        Self::from_arroyo_env_pairs(&env)
    }

    /// Pure core of [`from_arroyo_env`](Self::from_arroyo_env): map the supplied
    /// `ARROYO_*` environment map onto a wired [`MediaPlugin`] without touching
    /// process-global environment (mirroring
    /// [`MediaConfig::from_arroyo_env_pairs`](config::MediaConfig::from_arroyo_env_pairs)).
    #[must_use]
    pub fn from_arroyo_env_pairs(env: &HashMap<String, String>) -> Self {
        let config = MediaConfig::from_arroyo_env_pairs(env);
        Self::new()
            .config(config)
            .with_broadcast()
            .recordings_root(arroyo_recordings_root(env))
    }

    /// Supply the resolved `[media]` configuration.
    ///
    /// The plugin cannot read `[media]` itself because Autumn loads config
    /// after [`Plugin::build`] runs; resolve it up front with
    /// [`MediaConfig::from_autumn_toml`](config::MediaConfig::from_autumn_toml)
    /// or [`MediaConfig::from_arroyo_env`](config::MediaConfig::from_arroyo_env)
    /// and pass it here. Adopts the config's `room_max_participants`.
    #[must_use]
    pub fn config(mut self, config: MediaConfig) -> Self {
        self.room_max_participants = config.room_max_participants;
        self.config = config;
        self
    }

    /// Enable the broadcast primitive (ingest → fan-out playback → VOD).
    ///
    /// Installs [`MediaMtxClient`] and [`MediaUrls`] extensions, built from
    /// `config.mediamtx`. The plugin installs them even if the storage config
    /// is not valid.
    #[must_use]
    pub const fn with_broadcast(mut self) -> Self {
        self.enable_broadcast = true;
        self
    }

    /// Enable the rooms primitive (small mesh calls).
    ///
    /// Mounts the room routes, installs [`RoomService`] and starts the room
    /// reaper.
    #[must_use]
    pub const fn with_rooms(mut self) -> Self {
        self.enable_rooms = true;
        self
    }

    /// Override the hard cap on mesh-room participants.
    #[must_use]
    pub const fn room_max_participants(mut self, max: usize) -> Self {
        self.room_max_participants = max;
        self
    }

    /// Override the mesh-room session-token lifetime, in seconds (shortcut for
    /// `config.room_token_ttl_seconds`).
    #[must_use]
    pub const fn room_token_ttl_seconds(mut self, seconds: u32) -> Self {
        self.config.room_token_ttl_seconds = seconds;
        self
    }

    /// Override the maximum room session length, in seconds (shortcut for
    /// `config.room_session_max_seconds`). It must be at least the token TTL.
    #[must_use]
    pub const fn room_session_max_seconds(mut self, seconds: u32) -> Self {
        self.config.room_session_max_seconds = seconds;
        self
    }

    /// Override the mesh-room `MediaMTX` path namespace (shortcut for
    /// `config.room_namespace`).
    #[must_use]
    pub fn room_namespace(mut self, namespace: impl Into<String>) -> Self {
        self.config.room_namespace = Some(namespace.into());
        self
    }

    /// Override the job/Harvest queue name used for media encode work.
    #[must_use]
    pub fn queue(mut self, queue: impl Into<String>) -> Self {
        self.queue = queue.into();
        self
    }

    /// Override the URL prefix for the plugin's API routes.
    #[must_use]
    pub fn api(mut self, prefix: impl Into<String>) -> Self {
        self.api_prefix = prefix.into();
        self
    }

    /// Override the `FFmpeg` binary path (shortcut for
    /// `config.ffmpeg.bin`).
    #[must_use]
    pub fn ffmpeg_bin(mut self, bin: impl Into<String>) -> Self {
        self.config.ffmpeg.bin = bin.into();
        self
    }

    /// Install the app-supplied [`MediaArtifactSink`](sink::MediaArtifactSink)
    /// the built-in workflow jobs invoke on completion.
    ///
    /// Without a sink, a completed workflow persists its output but logs that
    /// no sink recorded it (parallels leaving `OutboundWebhookHandler` unset).
    #[must_use]
    pub fn artifact_sink(mut self, sink: Arc<dyn MediaArtifactSinkTrait>) -> Self {
        self.artifact_sink = Some(sink);
        self
    }

    /// Install a runtime workflow delegate that overrides the built-in `#[job]`
    /// engine (e.g. an external Harvest adapter maintained outside this
    /// workspace). When set, every [`MediaWorkflows`] `queue_*` call routes
    /// through the delegate instead of enqueuing the built-in job.
    #[must_use]
    pub fn workflow_delegate(mut self, delegate: MediaWorkflowDelegateHook) -> Self {
        self.workflow_delegate = Some(delegate);
        self
    }

    /// Override the recording-retention window, in days (`0` disables the
    /// sweep). Defaults to `config.recording.retention_days`.
    #[must_use]
    pub const fn retention_days(mut self, days: u32) -> Self {
        self.retention_days = Some(days);
        self
    }

    /// Set the filesystem root the retention sweep deletes expired recordings
    /// from. Required to spawn the sweep — without it, no retention loop runs.
    #[must_use]
    pub fn recordings_root(mut self, root: impl Into<PathBuf>) -> Self {
        self.recordings_root = Some(root.into());
        self
    }

    /// Install the app-overridable retention-defer predicate: return `true` for
    /// a path to hold it back this sweep (e.g. a still-encoding workflow
    /// references it).
    #[must_use]
    pub fn retention_defer(mut self, defer: RetentionDeferHook) -> Self {
        self.retention_defer = Some(defer);
        self
    }
}

impl Default for MediaPlugin {
    fn default() -> Self {
        Self::new()
    }
}

impl Plugin for MediaPlugin {
    /// This plugin ships in lockstep with `autumn-web` — see
    /// [`lockstep_contract`](autumn_web::plugin_contract::lockstep_contract).
    fn contract(&self) -> Option<autumn_web::plugin_contract::PluginContract> {
        Some(autumn_web::plugin_contract::lockstep_contract(
            env!("CARGO_PKG_NAME"),
            env!("CARGO_PKG_VERSION"),
        ))
    }

    fn name(&self) -> Cow<'static, str> {
        Cow::Borrowed("autumn-media-plugin")
    }

    /// Applies the plugin to the [`AppBuilder`].
    ///
    /// Declares the `[media]` config section (via
    /// [`AppBuilder::config_section`]). Then `server.strict_config = true` does
    /// not reject `[media]`. Then installs the parts of each enabled primitive
    /// (see the crate docs).
    // `build` is a long, linear plugin-assembly routine (config validation,
    // room wiring + reaper, broadcast transport, storage/workflow install,
    // retention loop); it reads best as one top-to-bottom sequence rather than
    // fragmented across helpers.
    #[allow(clippy::too_many_lines)]
    fn build(self, app: AppBuilder) -> AppBuilder {
        let Self {
            config,
            enable_broadcast,
            enable_rooms,
            room_max_participants,
            queue,
            api_prefix,
            artifact_sink,
            workflow_delegate,
            retention_days,
            recordings_root,
            retention_defer,
        } = self;

        // Declare the plugin-owned `[media]` top-level config table so a host app
        // running `server.strict_config = true` boots cleanly instead of failing
        // with `unknown key "media"`. `MediaConfig` is still read from raw TOML
        // (see `MediaConfig::from_autumn_toml`); this only tells core's strict
        // unknown-key check that `[media]` is a known, opaque section the plugin
        // owns and validates itself. Applied first so every return path below
        // (including the early room-misconfig abort) carries the declaration.
        let app = app.config_section("media");

        let retention_days = retention_days.unwrap_or(config.recording.retention_days);

        // The mesh is O(N²), so `DEFAULT_ROOM_MAX_PARTICIPANTS` (6) is an
        // ABSOLUTE ceiling with no SFU: the `[media]` room config and the
        // `room_max_participants` builder may set a per-room seat count only
        // within `1..=6`. An out-of-range value (0 or >6) is a fatal
        // misconfiguration — fail fast and LOUD at boot, never a silent clamp.
        // `Plugin::build` returns an `AppBuilder` (not a `Result`), so the
        // specific error is surfaced from an `on_startup` hook, which aborts
        // boot with `process::exit(1)` exactly as a failed init would (see
        // autumn-web's `run_startup_hooks`). `room_max_participants` is the
        // effective seat count sourced from either `config(..)` (which copies
        // `room_max_participants` into it) or the `room_max_participants(..)`
        // builder, so this single check covers both the TOML and builder paths;
        // `InMemoryRoomStore::create_room` re-checks the fixed 6 ceiling as a
        // defense-in-depth backstop, and `MediaConfig::validate` stays the
        // opt-in strict check for consumers who validate config up front.
        //
        // A configured `room_namespace` is prepended to every room path, so an
        // invalid one (a slash, a dot segment, whitespace) would likewise mount
        // the router fine yet make every `POST /rooms` fail `InvalidSegment` —
        // another config that can never serve a request. It fails fast here too,
        // via the same `on_startup` abort but its own specific message.
        // The storage backend keeps its own degrade path below (unchanged), so
        // this fails fast only on the room cap, the room namespace and the
        // session limit.
        //
        // Gated on `enable_rooms`: a broadcast-only plugin (`with_broadcast()`
        // without `with_rooms()`) mounts no room router / `RoomService`, so a
        // stray/irrelevant room setting — or a room env override — must never
        // abort boot and take the broadcast path down with it. The rooms path
        // keeps the existing fail-fast (see `room_config_boot_error`).
        if let Some(message) = room_config_boot_error(enable_rooms, room_max_participants, &config)
        {
            tracing::error!(
                room_max_participants,
                ceiling = DEFAULT_ROOM_MAX_PARTICIPANTS,
                "🍂 Autumn Media: {message}"
            );
            return app.on_startup(move |_state| {
                let message = message.clone();
                async move { Err(autumn_web::AutumnError::internal_server_error_msg(message)) }
            });
        }

        tracing::info!(
            broadcast = %enable_broadcast,
            rooms = %enable_rooms,
            room_max_participants,
            room_store_backend = config.room_store_backend.as_str(),
            storage_backend = config.storage.backend.as_str(),
            queue = %queue,
            api_prefix = %api_prefix,
            has_artifact_sink = artifact_sink.is_some(),
            has_workflow_delegate = workflow_delegate.is_some(),
            retention_days,
            "🍂 Autumn Media configured"
        );

        // Both primitives are off by default. With neither, install nothing.
        if !enable_broadcast && !enable_rooms {
            tracing::warn!(
                recordings_root_ignored = recordings_root.is_some(),
                "🍂 Autumn Media: no primitive enabled; call `with_broadcast()` or \
                 `with_rooms()`. The plugin installs nothing, and no retention sweep starts."
            );
            return app;
        }

        // Rooms are storage-independent, so mount the room signaling router and
        // install the `RoomService` extension up front — they must stay
        // available even if the storage backend below fails to resolve. The
        // `nest` + `declare_plugin_routes` pair keeps the room routes both
        // served and audit-visible under `api_prefix`.
        let mut app = app;
        if enable_rooms {
            // Select the room-state backend (config, default `memory`).
            //
            // The `memory` store is per-process, so the `RoomService` and the
            // reaper MUST share ONE instance — built once here and cloned into
            // both hooks. The `db` store is stateless w.r.t. process memory (all
            // room state lives in the shared database), so per-hook instances
            // over the same pool are equivalent; it is built lazily from
            // `state.pool()` inside each hook via `build_room_store`, which
            // degrades to a warned in-memory store if `db` was selected without
            // a configured database.
            let backend = config.room_store_backend;
            let shared_memory_store: Option<Arc<dyn rooms::RoomStore>> = match backend {
                config::RoomStoreBackend::Memory => Some(Arc::new(rooms::InMemoryRoomStore::new(
                    room_max_participants,
                ))),
                config::RoomStoreBackend::Db => None,
            };

            // URL/namespace/TTL for the `RoomService`, cloned into the
            // initializer so the service can be built there (the `db` store
            // needs `&AppState`).
            let room_urls = transport::MediaUrls::from_config(&config.mediamtx);
            let room_namespace = config.room_namespace.clone().unwrap_or_default();
            let room_token_ttl =
                chrono::Duration::seconds(i64::from(config.room_token_ttl_seconds));
            let room_max_session =
                chrono::Duration::seconds(i64::from(config.room_session_max_seconds));
            let room_rate_limit = config.room_rate_limit_per_minute;

            let init_store = shared_memory_store.clone();
            app = app
                .nest(&api_prefix, rooms::room_router())
                .declare_plugin_routes(rooms::room_route_infos(&api_prefix))
                .state_initializer(move |state| {
                    let store = init_store
                        .clone()
                        .unwrap_or_else(|| build_room_store(state, room_max_participants));
                    let room_service = rooms::RoomService::new(
                        store,
                        room_urls.clone(),
                        room_namespace.clone(),
                        room_token_ttl,
                        room_max_participants,
                    )
                    .with_session_max(room_max_session)
                    .with_rate_limit_per_minute(room_rate_limit);
                    state.insert_extension(room_service);
                })
                // Spawn from `on_startup` so the reaper shares the running app's
                // tokio runtime (matching the retention sweep below). For the
                // `db` backend it builds its own store instance over the same
                // pool the service uses — sweeping the same shared rows.
                .on_startup(move |state| {
                    let store = shared_memory_store
                        .clone()
                        .unwrap_or_else(|| build_room_store(&state, room_max_participants));
                    async move {
                        rooms::spawn_room_reaper_loop(store);
                        Ok(())
                    }
                });
        }

        // The broadcast transport does not need storage, so install it before
        // the storage check below.
        if enable_broadcast {
            let mtx_client = transport::MediaMtxClient::new(&config.mediamtx);
            let urls = transport::MediaUrls::from_config(&config.mediamtx);
            app = app.state_initializer(move |state| {
                state.insert_extension(mtx_client);
                state.insert_extension(urls);
            });
        }

        // Resolve the storage backend up front so a misconfiguration surfaces
        // as one error line here rather than inside a job. If storage fails, the
        // plugin keeps the room routes and broadcast transport. It installs no
        // storage, workflows, encode jobs or retention sweep.
        let storage = match storage::MediaStorage::from_config(&config.storage) {
            Ok(storage) => storage,
            Err(error) => {
                tracing::error!(
                    %error,
                    "🍂 Autumn Media: storage config invalid; encode workflows disabled"
                );
                return app;
            }
        };

        let ffmpeg_bin = config.ffmpeg.bin;
        let workflows = workflows::MediaWorkflows::new(ffmpeg_bin, queue.clone());

        // Extensions are installed via `state_initializer` (not `on_startup`)
        // so they exist BEFORE job workers start — mirroring autumn-web's
        // outbound-webhook plugin, whose manager must be present before the
        // first job runs.
        let app = app
            .state_initializer(move |state| {
                state.insert_extension(storage.clone());
                state.insert_extension(workflows.clone());
                if let Some(sink) = &artifact_sink {
                    state.insert_extension(sink::MediaArtifactSinkExt(sink.clone()));
                }
                if let Some(delegate) = &workflow_delegate {
                    state.insert_extension(workflows::MediaWorkflowDelegateExt(delegate.clone()));
                }
            })
            // Register the built-in encode jobs with the queue overridden to
            // `queue` (the JobInfo's `queue` field, not the `#[job]` literal, is
            // what the enqueue chokepoint routes on). autumn-web's worker
            // auto-registers any declared-but-unconfigured queue at lowest
            // priority, so the media queue drains without extra config.
            .jobs(workflows::media_job_infos(&queue));

        // Recording-retention sweep: spawn only when a recordings root is
        // configured and retention is enabled. Spawned from `on_startup` so it
        // shares the running app's runtime, matching how the thumbnail/retention
        // loops are spawned in an Autumn app.
        if let Some(root) = recordings_root {
            app.on_startup(move |_state| {
                let root = root.clone();
                let defer = retention_defer.clone();
                async move {
                    retention::spawn_retention_sweep_loop(root, retention_days, defer);
                    Ok(())
                }
            })
        } else {
            if retention_days > 0 {
                tracing::debug!(
                    "🍂 Autumn Media: retention window set but no recordings_root; sweep not spawned"
                );
            }
            app
        }
    }
}

/// Validate an effective mesh-room seat count against the ABSOLUTE ceiling.
///
/// Mesh WebRTC is O(N²), so [`DEFAULT_ROOM_MAX_PARTICIPANTS`] (6) is a hard cap
/// with no SFU, and a room must seat at least one participant. Returns the
/// specific, fail-fast error message — naming the offending value and the cap —
/// when `configured` is outside `1..=6`, or `None` when it is in range.
fn room_max_participants_error(configured: usize) -> Option<String> {
    if configured == 0 {
        Some(format!(
            "a room must allow at least 1 participant; got {configured}"
        ))
    } else if configured > DEFAULT_ROOM_MAX_PARTICIPANTS {
        Some(format!(
            "mesh rooms are capped at {DEFAULT_ROOM_MAX_PARTICIPANTS} participants (no SFU); got {configured}"
        ))
    } else {
        None
    }
}

/// Validate a configured mesh-room `MediaMTX` path namespace at boot.
///
/// The namespace is prepended to every room / participant `MediaMTX` path, so a
/// value [`rooms::validate_room_segment`] would reject (a slash, a `.` / `..`
/// dot segment, whitespace, or other non-`[A-Za-z0-9_-]` label) mounts the
/// router fine but makes every `POST /rooms` fail with `InvalidSegment` — a
/// config that can never serve a request must instead refuse to boot. Returns
/// the specific, fail-fast error message — naming the offending value and the
/// allowed charset — when the namespace is present-and-invalid, or `None` when
/// it is absent, empty, or a valid segment.
fn room_namespace_error(namespace: Option<&str>) -> Option<String> {
    let ns = namespace?;
    if ns.is_empty() || rooms::validate_room_segment(ns).is_ok() {
        return None;
    }
    Some(format!(
        "media.room_namespace {ns:?} is not a valid room path segment: use only ASCII letters, digits, '_' or '-' (no '/', '.', or empty)"
    ))
}

/// Resolve the fail-fast boot error for room configuration, gated on rooms
/// actually being enabled.
///
/// Room config (the `room_max_participants` seat count and the
/// `room_namespace` path segment) is only load-bearing when a room router /
/// [`rooms::RoomService`] is mounted — i.e. when `enable_rooms` is true. A
/// broadcast-only plugin (`with_broadcast()` without `with_rooms()`) serves no
/// room requests, so an out-of-range `room_max_participants` or an invalid
/// `room_namespace` (e.g. a stray value or a `media.room_namespace` env
/// override) is inert and must **not** abort boot — otherwise an irrelevant
/// room setting becomes a full broadcast outage. When `enable_rooms` is false
/// this returns `None` unconditionally; when it is true the existing
/// [`room_max_participants_error`] / [`room_namespace_error`] fail-fast checks
/// apply exactly as before.
fn room_config_boot_error(
    enable_rooms: bool,
    room_max_participants: usize,
    config: &MediaConfig,
) -> Option<String> {
    if !enable_rooms {
        return None;
    }
    room_max_participants_error(room_max_participants)
        .or_else(|| room_namespace_error(config.room_namespace.as_deref()))
        .or_else(|| room_session_max_error(config))
}

/// Reject a session limit shorter than the token TTL.
///
/// A join mints a token that lives `room_token_ttl_seconds`. A shorter session
/// limit ends the session before that token expires.
fn room_session_max_error(config: &MediaConfig) -> Option<String> {
    let session_max = config.room_session_max_seconds;
    let token_ttl = config.room_token_ttl_seconds;
    (session_max < token_ttl).then(|| {
        format!(
            "media.room_session_max_seconds ({session_max}) must be at least \
             media.room_token_ttl_seconds ({token_ttl})"
        )
    })
}

/// Build a shared, database-backed [`rooms_db::DbRoomStore`] from the running
/// app's connection pool for the `db` room-store backend.
///
/// Only called on the `db` code path (the `memory` backend shares one pre-built
/// [`rooms::InMemoryRoomStore`] across the service and reaper). If `db` was
/// selected but no database pool is configured, it logs an actionable error and
/// **degrades to a per-process in-memory store** so the app still boots and
/// rooms still work — they just will not survive across processes until a
/// database is configured.
fn build_room_store(state: &autumn_web::AppState, cap: usize) -> Arc<dyn rooms::RoomStore> {
    if let Some(pool) = state.pool().cloned() {
        Arc::new(rooms_db::DbRoomStore::new(pool, cap))
    } else {
        tracing::error!(
            "🍂 Autumn Media: room_store_backend=\"db\" but no database pool is configured; \
             falling back to the in-memory room store — rooms will NOT survive across \
             processes. Configure a `[database]` primary_url to enable the shared store."
        );
        Arc::new(rooms::InMemoryRoomStore::new(cap))
    }
}

/// Recordings root an Arroyo deployment uses when `ARROYO_RECORDINGS_ROOT` is
/// unset — mirrors Arroyo's own `configured_recordings_root()` fallback, so the
/// migration shim wires the retention sweep the same way with no ops change.
const DEFAULT_ARROYO_RECORDINGS_ROOT: &str = "recordings";

/// Resolve the Arroyo recordings root from an env map: a non-blank
/// `ARROYO_RECORDINGS_ROOT`, else [`DEFAULT_ARROYO_RECORDINGS_ROOT`]. Blank /
/// whitespace-only values are treated as unset (parity with
/// [`MediaConfig::from_arroyo_env_pairs`](config::MediaConfig::from_arroyo_env_pairs)).
fn arroyo_recordings_root(env: &HashMap<String, String>) -> PathBuf {
    env.get("ARROYO_RECORDINGS_ROOT")
        .map(|value| value.trim())
        .filter(|value| !value.is_empty())
        .map_or_else(
            || PathBuf::from(DEFAULT_ARROYO_RECORDINGS_ROOT),
            PathBuf::from,
        )
}

#[cfg(test)]
mod contract_tests {
    use super::MediaPlugin;
    use autumn_web::plugin::Plugin;

    #[test]
    fn contract_declares_lockstep_with_own_crate() {
        let contract = MediaPlugin::new().contract().expect("a contract");
        assert_eq!(contract.plugin, env!("CARGO_PKG_NAME"));
        assert_eq!(
            contract.plugin_version.as_deref(),
            Some(env!("CARGO_PKG_VERSION"))
        );
        assert_eq!(
            contract.autumn_web.as_deref(),
            Some(autumn_web::plugin_contract::lockstep_range(env!("CARGO_PKG_VERSION")).as_str())
        );
        assert!(contract.experimental_surfaces.is_empty());
    }
}

// ── Arroyo migration shim (slice 5) ─────────────────────────────────────────

#[cfg(test)]
mod arroyo_shim_tests {
    use super::{DEFAULT_ARROYO_RECORDINGS_ROOT, MediaPlugin, arroyo_recordings_root};
    use crate::config::MediaStorageBackend;
    use std::collections::HashMap;
    use std::path::PathBuf;

    fn env(pairs: &[(&str, &str)]) -> HashMap<String, String> {
        pairs
            .iter()
            .map(|(key, value)| ((*key).to_owned(), (*value).to_owned()))
            .collect()
    }

    #[test]
    fn empty_env_is_wired_broadcast_local_plugin() {
        let plugin = MediaPlugin::from_arroyo_env_pairs(&HashMap::new());
        // Broadcast is enabled (Arroyo is one-to-many); rooms are not.
        assert!(plugin.enable_broadcast, "broadcast must be enabled");
        assert!(!plugin.enable_rooms, "rooms must stay disabled");
        // Config maps through with neutral defaults for an empty env.
        assert_eq!(plugin.config.storage.backend, MediaStorageBackend::Local);
        assert_eq!(plugin.config.mediamtx.api_base, "http://127.0.0.1:9997");
        assert_eq!(plugin.config.ffmpeg.bin, "/usr/bin/ffmpeg");
        // Retention sweep is wired to Arroyo's default recordings root, so the
        // hourly loop runs exactly as it did pre-migration.
        assert_eq!(
            plugin.recordings_root.as_deref(),
            Some(std::path::Path::new(DEFAULT_ARROYO_RECORDINGS_ROOT))
        );
        // The produced config is valid to hand to MediaStorage::from_config.
        assert!(plugin.config.validate().is_ok());
    }

    #[test]
    fn maps_full_s3_environment_and_validates() {
        let vars = env(&[
            ("ARROYO_VIDEO_STORAGE_BACKEND", "s3"),
            ("BUCKET_NAME", "arroyo-bucket"),
            ("AWS_ACCESS_KEY_ID", "AKIA"),
            ("AWS_SECRET_ACCESS_KEY", "secret"),
            ("ARROYO_MEDIAMTX_API_BASE", "http://mtx:9997"),
            ("ARROYO_FFMPEG_BIN", "/opt/ffmpeg"),
            ("ARROYO_RECORDING_RETENTION_DAYS", "21"),
            ("ARROYO_RECORDINGS_ROOT", "/data/recordings"),
        ]);
        let plugin = MediaPlugin::from_arroyo_env_pairs(&vars);
        assert!(plugin.enable_broadcast);
        assert_eq!(plugin.config.storage.backend, MediaStorageBackend::S3);
        assert_eq!(
            plugin.config.storage.bucket.as_deref(),
            Some("arroyo-bucket")
        );
        // Arroyo's Tigris S3 defaults flow through the config-level shim.
        assert_eq!(plugin.config.storage.region.as_deref(), Some("auto"));
        assert_eq!(
            plugin.config.storage.endpoint_url.as_deref(),
            Some("https://t3.storage.dev")
        );
        assert_eq!(plugin.config.storage.key_prefix, "highlights");
        assert_eq!(plugin.config.mediamtx.api_base, "http://mtx:9997");
        assert_eq!(plugin.config.ffmpeg.bin, "/opt/ffmpeg");
        // Retention window flows from ARROYO_RECORDING_RETENTION_DAYS into the
        // config (the plugin resolves the effective days from it at build time).
        assert_eq!(plugin.config.recording.retention_days, 21);
        assert_eq!(
            plugin.recordings_root.as_deref(),
            Some(std::path::Path::new("/data/recordings"))
        );
        assert!(plugin.config.validate().is_ok());
    }

    #[test]
    fn recordings_root_honors_env_and_defaults() {
        assert_eq!(
            arroyo_recordings_root(&env(&[("ARROYO_RECORDINGS_ROOT", "/mnt/rec")])),
            PathBuf::from("/mnt/rec")
        );
        // Blank / whitespace-only is treated as unset → Arroyo's default.
        assert_eq!(
            arroyo_recordings_root(&env(&[("ARROYO_RECORDINGS_ROOT", "   ")])),
            PathBuf::from(DEFAULT_ARROYO_RECORDINGS_ROOT)
        );
        assert_eq!(
            arroyo_recordings_root(&HashMap::new()),
            PathBuf::from(DEFAULT_ARROYO_RECORDINGS_ROOT)
        );
    }

    #[test]
    fn returned_plugin_stays_chainable() {
        // The shim returns a normal builder, so an app can layer its own
        // overrides (queue name, retention window) on top of the mapped env.
        let plugin = MediaPlugin::from_arroyo_env_pairs(&HashMap::new())
            .queue("arroyo-media")
            .retention_days(30);
        assert!(plugin.enable_broadcast);
        assert_eq!(plugin.queue, "arroyo-media");
        assert_eq!(plugin.retention_days, Some(30));
    }
}

// ── Absolute room-cap enforcement (Fix 1) ───────────────────────────────────

#[cfg(test)]
mod room_cap_tests {
    use super::{DEFAULT_ROOM_MAX_PARTICIPANTS, MediaPlugin, room_max_participants_error};
    use crate::config::MediaConfig;

    #[test]
    fn out_of_range_room_cap_is_a_specific_fail_fast_error() {
        // >6 fails loud, naming the offending value and the ceiling — never a
        // clamp.
        let over = room_max_participants_error(50).expect("50 > 6 must be rejected");
        assert!(over.contains("50"), "names the offending value: {over}");
        assert!(over.contains('6'), "names the ceiling: {over}");
        assert!(over.contains("no SFU"), "explains why: {over}");
        // 0 fails loud with its own message.
        let zero = room_max_participants_error(0).expect("0 must be rejected");
        assert!(zero.contains("at least 1 participant"), "0 message: {zero}");
        assert!(zero.contains('0'), "names the value: {zero}");
    }

    #[test]
    fn in_range_room_cap_is_accepted() {
        assert!(room_max_participants_error(1).is_none());
        assert!(room_max_participants_error(4).is_none());
        assert!(room_max_participants_error(DEFAULT_ROOM_MAX_PARTICIPANTS).is_none());
    }

    #[test]
    fn builder_and_config_paths_feed_the_same_effective_cap_that_boot_rejects() {
        // The `room_max_participants(50)` builder stores what it was given;
        // `build()` is what rejects it fail-fast (no clamp), so an out-of-range
        // builder value produces the specific error at boot.
        let over = MediaPlugin::new().with_rooms().room_max_participants(50);
        assert_eq!(over.room_max_participants, 50);
        assert!(room_max_participants_error(over.room_max_participants).is_some());

        // `room_max_participants(4)` stays in range → a real 4-seat room.
        let ok = MediaPlugin::new().with_rooms().room_max_participants(4);
        assert_eq!(ok.room_max_participants, 4);
        assert!(room_max_participants_error(ok.room_max_participants).is_none());

        // The `[media] room_max_participants = 50` config path flows into the
        // same effective field via `config(..)`, so it is rejected identically.
        let config = MediaConfig {
            room_max_participants: 50,
            ..MediaConfig::default()
        };
        let from_config = MediaPlugin::new().config(config);
        assert_eq!(from_config.room_max_participants, 50);
        assert!(room_max_participants_error(from_config.room_max_participants).is_some());
    }
}

// ── Room-namespace fail-fast (Fix P2-a) ─────────────────────────────────────

#[cfg(test)]
mod room_namespace_tests {
    use super::{MediaPlugin, room_namespace_error};

    #[test]
    fn absent_empty_and_valid_namespaces_are_accepted() {
        // Absent and empty mean "no namespace" → no boot error; a valid segment
        // passes through untouched.
        assert!(room_namespace_error(None).is_none());
        assert!(room_namespace_error(Some("")).is_none());
        assert!(room_namespace_error(Some("tenant-a")).is_none());
        assert!(room_namespace_error(Some("room_1")).is_none());
    }

    #[test]
    fn invalid_namespace_is_a_specific_fail_fast_error() {
        // A slash, either dot segment, or embedded whitespace fails loud, naming
        // the offending value and the allowed charset — never a silent mount
        // that 500s every request.
        for bad in ["tenant/a", ".", "..", "a b"] {
            let message = room_namespace_error(Some(bad))
                .unwrap_or_else(|| panic!("{bad:?} must be rejected"));
            assert!(
                message.contains(bad),
                "names the offending value: {message}"
            );
            assert!(
                message.contains("letters") && message.contains('-'),
                "mentions the allowed charset: {message}"
            );
        }
    }

    #[test]
    fn room_namespace_builder_feeds_the_effective_value_that_boot_rejects() {
        // The `room_namespace("tenant/a")` builder stores what it was given;
        // `build()` is what rejects it fail-fast, so an invalid builder value
        // produces the specific error at boot.
        let bad = MediaPlugin::new().with_rooms().room_namespace("tenant/a");
        assert_eq!(bad.config.room_namespace.as_deref(), Some("tenant/a"));
        assert!(room_namespace_error(bad.config.room_namespace.as_deref()).is_some());

        // A valid namespace flows through untouched → a real namespaced room.
        let ok = MediaPlugin::new().with_rooms().room_namespace("tenant-a");
        assert_eq!(ok.config.room_namespace.as_deref(), Some("tenant-a"));
        assert!(room_namespace_error(ok.config.room_namespace.as_deref()).is_none());
    }
}

// ── Room-config fail-fast gated on rooms being enabled (Codex P2) ────────────
//
// The boot-time room-config abort must be conditional on `enable_rooms`: a
// broadcast-only plugin mounts no room router / `RoomService`, so a stray or
// invalid room setting (an out-of-range `room_max_participants`, a bad
// `room_namespace` env override) is inert and must never abort boot and take
// the broadcast path down with it. Rooms-enabled keeps the existing fail-fast.

#[cfg(test)]
mod room_config_gate_tests {
    use super::{MediaConfig, MediaPlugin, room_config_boot_error};

    fn with_namespace(namespace: &str) -> MediaConfig {
        MediaConfig {
            room_namespace: Some(namespace.to_owned()),
            ..MediaConfig::default()
        }
    }

    #[test]
    fn broadcast_only_ignores_invalid_room_config() {
        // `with_broadcast()` without `with_rooms()` → rooms disabled. An invalid
        // room cap (0) or namespace ("bad/ns") is irrelevant and must NOT abort
        // boot, so no failing startup hook is installed and broadcast stays up.
        assert!(!MediaPlugin::new().with_broadcast().enable_rooms);
        assert!(room_config_boot_error(false, 0, &MediaConfig::default()).is_none());
        assert!(room_config_boot_error(false, 50, &with_namespace("bad/ns")).is_none());
        assert!(room_config_boot_error(false, 4, &with_namespace("tenant-a")).is_none());
        let short_session = MediaConfig {
            room_session_max_seconds: 1,
            ..MediaConfig::default()
        };
        assert!(room_config_boot_error(false, 4, &short_session).is_none());
    }

    #[test]
    fn rooms_enabled_still_fails_fast_on_invalid_room_config() {
        // `with_rooms()` → rooms enabled. The same invalid settings that a
        // broadcast-only plugin ignores must still abort boot on the rooms path,
        // preserving the pre-existing fail-fast (naming the offending value).
        assert!(MediaPlugin::new().with_rooms().enable_rooms);

        let bad_cap = room_config_boot_error(true, 0, &MediaConfig::default())
            .expect("rooms-enabled must fail fast on room_max_participants = 0");
        assert!(bad_cap.contains('0'), "names the offending cap: {bad_cap}");

        let bad_ns = room_config_boot_error(true, 4, &with_namespace("bad/ns"))
            .expect("rooms-enabled must fail fast on an invalid room_namespace");
        assert!(
            bad_ns.contains("bad/ns"),
            "names the offending namespace: {bad_ns}"
        );

        // A valid room config on the rooms path boots cleanly (no abort).
        assert!(room_config_boot_error(true, 4, &with_namespace("tenant-a")).is_none());
    }

    #[test]
    fn rooms_enabled_fails_fast_on_a_session_max_less_than_the_token_ttl() {
        let short_session = MediaConfig {
            room_token_ttl_seconds: 300,
            room_session_max_seconds: 60,
            ..MediaConfig::default()
        };
        let message = room_config_boot_error(true, 4, &short_session)
            .expect("a session shorter than its token must abort boot");
        assert!(message.contains("60"), "names the session limit: {message}");
        assert!(message.contains("300"), "names the token TTL: {message}");
    }
}

// ── Conformance reference tests ─────────────────────────────────────────────
//
// The room signaling routes `MediaPlugin::build` declares under `/api/media`
// (via `rooms::room_route_infos`) are run through autumn-web's plugin
// conformance harness — the same checks `autumn-admin-plugin`'s
// `conformance_tests` uses — so the plugin's naming / prefix / attribution
// conventions stay clean and a future slice that touches the room routes must
// consciously keep them conformant.

#[cfg(test)]
mod conformance_tests {
    use autumn_web::plugin_conformance::{
        CheckStatus, ConformanceConfig, check_collisions, check_duplicate_registration,
        check_route_attribution, check_route_prefix, check_sensitive_surfaces, run_conformance,
    };
    use autumn_web::route_listing::{RouteInfo, RouteSource};

    const PLUGIN_NAME: &str = "autumn-media-plugin";
    const API_PREFIX: &str = "/api/media";

    /// The real room routes `build` declares, attributed to the plugin exactly
    /// as `declare_plugin_routes` attributes them at runtime.
    fn declared_room_routes() -> Vec<RouteInfo> {
        super::rooms::room_route_infos(API_PREFIX)
            .into_iter()
            .map(|mut route| {
                route.source = RouteSource::Plugin(PLUGIN_NAME.to_owned());
                route
            })
            .collect()
    }

    #[test]
    fn build_declares_the_five_room_routes_when_rooms_enabled() {
        let routes = super::rooms::room_route_infos(API_PREFIX);
        assert_eq!(routes.len(), 5, "rooms declare exactly five routes");
        assert!(
            routes
                .iter()
                .all(|route| route.path.starts_with(API_PREFIX)),
            "every declared room route lives under the API prefix"
        );
    }

    #[test]
    fn room_routes_are_attributed_to_plugin_name() {
        let result = check_route_attribution(PLUGIN_NAME, &declared_room_routes());
        assert_eq!(
            result.status,
            CheckStatus::Pass,
            "attribution failed: {}",
            result.message
        );
    }

    #[test]
    fn room_routes_live_under_api_prefix() {
        let result = check_route_prefix(PLUGIN_NAME, API_PREFIX, &[], &declared_room_routes());
        assert_eq!(
            result.status,
            CheckStatus::Pass,
            "prefix check failed: {}",
            result.message
        );
    }

    #[test]
    fn room_routes_have_no_collisions_in_isolation() {
        let (result, _) = check_collisions(&declared_room_routes());
        assert_eq!(
            result.status,
            CheckStatus::Pass,
            "unexpected collision: {}",
            result.message
        );
    }

    #[test]
    fn room_routes_have_no_undeclared_sensitive_surfaces() {
        let result = check_sensitive_surfaces(PLUGIN_NAME, &declared_room_routes(), &[]);
        assert_eq!(
            result.status,
            CheckStatus::Pass,
            "sensitive-surfaces check failed: {}",
            result.message
        );
    }

    #[test]
    fn room_single_registration_passes_duplicate_check() {
        let result = check_duplicate_registration(PLUGIN_NAME, &declared_room_routes());
        assert_eq!(
            result.status,
            CheckStatus::Pass,
            "single registration should pass: {}",
            result.message
        );
    }

    #[test]
    fn room_routes_pass_full_conformance() {
        let config = ConformanceConfig::new(PLUGIN_NAME).prefix(API_PREFIX);
        let report = run_conformance(&config, &declared_room_routes());
        assert!(
            report.passed(),
            "MediaPlugin conformance failed:\n{}",
            report.to_text_report()
        );
    }

    #[test]
    fn duplicate_registration_detected() {
        // Installing the plugin twice would double its routes → FAIL.
        let mut routes = declared_room_routes();
        routes.extend(declared_room_routes());
        let result = check_duplicate_registration(PLUGIN_NAME, &routes);
        assert_eq!(
            result.status,
            CheckStatus::Fail,
            "expected duplicate-registration FAIL when installed twice"
        );
    }

    #[test]
    fn collision_with_host_route_detected() {
        // Sanity check the harness is wired: a host route colliding with a
        // plugin route is flagged.
        let mut routes = declared_room_routes();
        routes.push(RouteInfo {
            method: "GET".to_owned(),
            path: format!("{API_PREFIX}/rooms/{{room_id}}"),
            handler: "host::roster".to_owned(),
            source: RouteSource::User,
            ..Default::default()
        });
        let (result, _) = check_collisions(&routes);
        assert_eq!(
            result.status,
            CheckStatus::Fail,
            "expected collision to be detected"
        );
    }
}

#[cfg(test)]
mod config_section_tests {
    use super::MediaPlugin;

    // #1974 item 7: `MediaPlugin::build` must declare its `[media]` top-level
    // config section on the builder so a host app with
    // `server.strict_config = true` boots without core rejecting `[media]` as an
    // unknown key. Registering the real plugin and asserting the section is
    // declared proves the wiring end-to-end (not just a stand-in).
    #[test]
    fn build_declares_media_config_section() {
        let builder = autumn_web::app().plugin(MediaPlugin::new());
        assert!(
            builder.has_config_section("media"),
            "MediaPlugin::build must declare the [media] config section for strict_config"
        );
    }
}

// Each primitive installs only its own surface (#1974).
#[cfg(test)]
mod primitive_surface_tests {
    use std::collections::{HashMap, HashSet};
    use std::path::Path;
    use std::sync::{Arc, Mutex};
    use std::time::{Duration, SystemTime};

    use autumn_web::AppState;
    use autumn_web::test::TestApp;

    use super::MediaPlugin;
    use crate::config::{MediaConfig, MediaStorageBackend};
    use crate::{
        MediaMtxClient, MediaStorage, MediaUrls, MediaWorkflows, RetentionDefer, RoomService,
        media_job_infos,
    };

    /// The extensions a built app holds.
    #[allow(clippy::struct_excessive_bools)] // one flag per extension
    #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
    struct Installed {
        storage: bool,
        workflows: bool,
        mtx_client: bool,
        urls: bool,
        rooms: bool,
    }

    const NOTHING: Installed = Installed {
        storage: false,
        workflows: false,
        mtx_client: false,
        urls: false,
        rooms: false,
    };

    /// Build `plugin` in a test app and return what `read` sees in its state.
    fn probe<T: Send + 'static>(
        plugin: MediaPlugin,
        read: impl Fn(&AppState) -> T + Send + Sync + 'static,
    ) -> T {
        let seen = Arc::new(Mutex::new(None));
        let sink = Arc::clone(&seen);
        let _client = TestApp::new()
            .plugin(plugin)
            .state_initializer(move |state| *sink.lock().unwrap() = Some(read(state)))
            .build();
        seen.lock().unwrap().take().expect("probe ran")
    }

    #[tokio::test]
    async fn rooms_take_the_session_limit_and_rate_limit_from_config() {
        let config = MediaConfig {
            room_session_max_seconds: 3600,
            room_rate_limit_per_minute: 30,
            ..MediaConfig::default()
        };
        let (rate, session) = probe(MediaPlugin::new().config(config).with_rooms(), |state| {
            let rooms = state.extension::<RoomService>().expect("rooms installed");
            (rooms.rate_limit_per_minute(), rooms.session_max())
        });
        assert_eq!(rate, 30);
        assert_eq!(session, chrono::Duration::hours(1));
    }

    /// The extensions `plugin` installs.
    fn installed(plugin: MediaPlugin) -> Installed {
        probe(plugin, |state| Installed {
            storage: state.extension::<MediaStorage>().is_some(),
            workflows: state.extension::<MediaWorkflows>().is_some(),
            mtx_client: state.extension::<MediaMtxClient>().is_some(),
            urls: state.extension::<MediaUrls>().is_some(),
            rooms: state.extension::<RoomService>().is_some(),
        })
    }

    #[tokio::test]
    async fn no_primitive_installs_nothing() {
        assert_eq!(installed(MediaPlugin::new()), NOTHING);
    }

    #[tokio::test]
    async fn broadcast_installs_transport_and_encode() {
        let got = installed(MediaPlugin::new().with_broadcast());
        assert_eq!(
            got,
            Installed {
                storage: true,
                workflows: true,
                mtx_client: true,
                urls: true,
                rooms: false,
            }
        );
    }

    #[tokio::test]
    async fn rooms_install_rooms_and_encode_but_no_broadcast_transport() {
        let got = installed(MediaPlugin::new().with_rooms());
        assert_eq!(
            got,
            Installed {
                storage: true,
                workflows: true,
                mtx_client: false,
                urls: false,
                rooms: true,
            }
        );
    }

    #[tokio::test]
    async fn both_primitives_install_everything() {
        let got = installed(MediaPlugin::new().with_broadcast().with_rooms());
        assert_eq!(
            got,
            Installed {
                storage: true,
                workflows: true,
                mtx_client: true,
                urls: true,
                rooms: true,
            }
        );
    }

    #[tokio::test]
    async fn arroyo_shim_installs_broadcast_transport() {
        let got = installed(MediaPlugin::from_arroyo_env_pairs(&HashMap::new()));
        assert!(got.mtx_client && got.urls, "{got:?}");
        assert!(!got.rooms, "{got:?}");
    }

    #[tokio::test]
    async fn broadcast_transport_survives_a_storage_error() {
        // S3 without a bucket fails `MediaStorage::from_config`.
        let mut config = MediaConfig::default();
        config.storage.backend = MediaStorageBackend::S3;
        let got = installed(MediaPlugin::new().config(config).with_broadcast());
        assert!(got.mtx_client && got.urls, "{got:?}");
        assert!(!got.storage && !got.workflows, "{got:?}");
    }

    #[tokio::test]
    async fn transport_uses_the_configured_api_base() {
        let mut config = MediaConfig::default();
        config.mediamtx.api_base = "http://mediamtx.internal:9997".to_owned();
        let api_base = probe(
            MediaPlugin::new().config(config).with_broadcast(),
            |state| {
                state
                    .extension::<MediaMtxClient>()
                    .map(|client| client.api_base().to_owned())
            },
        );
        assert_eq!(api_base.as_deref(), Some("http://mediamtx.internal:9997"));
    }

    /// The media job names `plugin` registers.
    fn media_jobs(plugin: MediaPlugin) -> HashSet<String> {
        // Jobs register after the probe runs. The registry clone shares state,
        // so read it after `build`.
        let state = probe(plugin, AppState::clone);
        let registered = state.job_registry().snapshot();
        media_job_infos("media")
            .into_iter()
            .map(|info| info.name)
            .filter(|name| registered.contains_key(name))
            .collect()
    }

    #[tokio::test]
    async fn encode_jobs_register_only_with_a_primitive() {
        let all: HashSet<String> = media_job_infos("media")
            .into_iter()
            .map(|info| info.name)
            .collect();
        assert!(media_jobs(MediaPlugin::new()).is_empty());
        assert_eq!(media_jobs(MediaPlugin::new().with_broadcast()), all);
        assert_eq!(media_jobs(MediaPlugin::new().with_rooms()), all);
    }

    #[test]
    fn only_rooms_declare_routes() {
        let routes = |plugin| {
            autumn_web::app()
                .plugin(plugin)
                .plugin_route_infos()
                .expect("route infos")
        };
        assert!(routes(MediaPlugin::new()).is_empty());
        assert!(routes(MediaPlugin::new().with_broadcast()).is_empty());
        assert!(!routes(MediaPlugin::new().with_rooms()).is_empty());
    }

    /// Build `plugin` over a root with one expired file. Return `true` if the
    /// retention sweep looks at the file within `wait`.
    async fn sweep_runs(plugin: MediaPlugin, wait: Duration) -> bool {
        let root = tempfile::tempdir().expect("tempdir");
        let old = root.path().join("old.mp4");
        std::fs::write(&old, b"old").expect("write");
        std::fs::File::options()
            .write(true)
            .open(&old)
            .and_then(|file| file.set_modified(SystemTime::now() - Duration::from_hours(72)))
            .expect("set_modified");

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        // Keep the file; only report that the sweep saw it.
        let defer: RetentionDefer = Arc::new(move |path| {
            let _ = tx.send(path);
            Box::pin(async { true })
        });
        let plugin = plugin
            .recordings_root(root.path())
            .retention_days(1)
            .retention_defer(defer);
        let _client = TestApp::new().plugin(plugin).build();
        let seen = tokio::time::timeout(wait, rx.recv()).await;
        assert!(Path::new(&old).exists(), "the defer hook keeps the file");
        matches!(seen, Ok(Some(_)))
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn retention_sweep_starts_only_with_a_primitive() {
        assert!(sweep_runs(MediaPlugin::new().with_rooms(), Duration::from_secs(5)).await);
        assert!(!sweep_runs(MediaPlugin::new(), Duration::from_millis(300)).await);
    }
}