//! Per-app audio routing.
//!
//! Two cooperating layers:
//!
//! - **Null-sink + loopback** (pactl shell-out): a per-PID null-sink
//!   `pixelpass_capture_<pid>` plus a `module-loopback` that mirrors the
//!   default sink's monitor into it. gst captures from the null-sink's
//!   monitor, so the viewer hears whatever the user hears — by default.
//!
//! - **Per-stream rerouting** (libpipewire on a dedicated OS thread):
//!   when [`HostOpts::app`] is set, a [`StreamRouter`] subscribes to the
//!   PipeWire registry, finds `Stream/Output/Audio` nodes whose
//!   `application.name` matches the filter, and writes
//!   `target.object` to the "default" metadata so WirePlumber reroutes
//!   them to our null-sink. Once at least one stream is actually routed,
//!   the loopback is unloaded — otherwise the viewer would hear the
//!   filtered audio twice (once via the routed stream, once via the
//!   default-sink monitor loopback).
//!
//! - **Local monitor** (pactl shell-out, app mode only): rerouting *moves*
//!   the chosen app off the sharer's speakers into the null-sink, so without
//!   this the sharer would go deaf to the very content they're sharing. We
//!   mirror the null-sink's monitor back to `@DEFAULT_SINK@` so the sharer
//!   hears it too. Only the chosen app is in the null-sink — never the
//!   desktop/call — so this can't echo back into the capture. It is loaded on
//!   the first routed stream (after the default-sink loopback is gone, so the
//!   two never coexist and feed back) and unloaded when the app stops.
//!
//! pactl is the right tool for the one-shot null-sink/loopback graph
//! mutations. libpipewire is dragged in only when per-stream filtering
//! is requested, because that needs registry-event subscription.

use anyhow::{Context, Result, bail};
use std::cell::RefCell;
use std::collections::BTreeMap;
use std::io::{self, Read};
use std::process::{Child, Command, ExitStatus, Stdio};
use std::rc::Rc;
use std::sync::Arc;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use crate::cli::HostOpts;
use crate::host::ledger::{self, LedgerError, ModuleLedger, UnloadOutcome};
use crate::repair::plan::{self as repair_plan, Fingerprint, Shape};

/// How long a `pactl load-module` worker may run before it is killed and reaped.
///
/// A bound, not a calibration: a local Pulse socket answers in milliseconds, and
/// this exists only so a wedged server cannot hang teardown forever. It is
/// deliberately generous because exceeding it is no longer destructive: the
/// ledger records the attempt, and reconciliation finds whatever the server
/// actually did. Unloads use PulseSession's independently bounded native
/// connect/list/unload requests instead of a second pactl connection.
const PACTL_BUDGET: Duration = Duration::from_secs(5);
const PACTL_REAP_BUDGET: Duration = Duration::from_secs(1);

/// Owns the pactl-loaded modules plus, when filtering is active, the
/// libpipewire stream-router thread. Drop unloads modules as a backstop;
/// prefer [`Routing::shutdown`] explicitly, which is the only path that can
/// reconcile a load whose outcome was never observed.
pub struct Routing {
    /// Every module this host has loaded, is loading, or must ask the server
    /// about. Shared with the event task, which loads and unloads the two
    /// loopbacks as the routed app comes and goes.
    ///
    /// This replaced three `Option<u32>`s. The reason is in [`crate::host::ledger`]:
    /// an `Option` cannot say "a load is in flight", so a cancelled load looked
    /// exactly like no load at all and its module was left behind.
    ledger: Arc<ModuleLedger>,
    sink_name: String,
    stream_router: Option<StreamRouter>,
    event_task: Option<tokio::task::JoinHandle<()>>,
}

impl Routing {
    /// Create the per-PID null-sink + loopback. If `opts.app` is set,
    /// also spawn the libpipewire thread that reroutes matching streams.
    pub async fn start(opts: &HostOpts) -> Result<Self> {
        let pid = std::process::id();
        let sink_name = repair_plan::sink_name_for(pid);
        let ledger = ModuleLedger::new();
        // Construct the owner before the first mutation. Any error or cancellation
        // below now drops a real `Routing`, whose backstop closes, quiesces,
        // reconciles, and unloads this ledger. Previously the owner did not exist
        // until both initial modules had loaded, so constructor failure leaked
        // everything loaded up to that point.
        let mut routing = Self {
            ledger: Arc::clone(&ledger),
            sink_name: sink_name.clone(),
            stream_router: None,
            event_task: None,
        };

        // Every module this host loads carries an ownership token, minted per
        // load, so `--repair` can tell whose pid the name refers to instead of
        // assuming the number means the same thing everywhere. Without it a repair
        // run in another pid namespace can unload a live host's audio; see
        // `repair::plan::OwnerToken`. That same per-load nonce is what lets
        // reconciliation identify a module whose load was interrupted before its
        // index was ever read.
        load_module(&ledger, Shape::LegacyCaptureSink, pid)
            .await
            .context("failed to load module-null-sink")?;

        // In strict per-app mode we never mirror the default sink: the viewer
        // must hear *only* the chosen app, never the whole desktop (which would
        // leak e.g. a voice call the sharer is in back to viewers — the echo
        // bug A23). Without strict mode (whole-desktop share, or best-effort
        // app filtering) we load the monitor loopback so the viewer hears
        // system audio immediately and during any gap before the app routes.
        // 20ms loopback latency keeps the mirrored audio tight; pactl's
        // default of 200ms is enough to be perceptible.
        let strict_app = opts.app.is_some() && opts.strict_audio;
        if !strict_app {
            load_module(&ledger, Shape::LoopbackIntoCapture, pid)
                .await
                .context("failed to load module-loopback (null-sink cleaned up on Drop)")?;
        }

        tracing::info!(
            strict_app,
            %sink_name,
            "audio routing: null-sink ready (loopback skipped in strict app mode)"
        );

        if let Some(app) = &opts.app {
            let (router, mut event_rx) = StreamRouter::spawn(app.clone(), sink_name.clone())?;
            let ledger_for_task = Arc::clone(&ledger);
            let strict = opts.strict_audio;
            let event_task = tokio::spawn(async move {
                use crate::common::output::{self, AppAudioState};
                while let Some(ev) = event_rx.recv().await {
                    match ev {
                        Event::FirstRoutedStream => {
                            tracing::info!(
                                "audio routing: first stream routed → unloading default-sink loopback"
                            );
                            let mirror_absent =
                                unload_module(&ledger_for_task, Shape::LoopbackIntoCapture).await;
                            // Mirror the routed app back to the sharer's own
                            // speakers so they hear the content they're sharing.
                            // Loaded *after* the default-sink loopback is gone so
                            // the two never coexist (which would feed back), and
                            // sourced from the null-sink monitor — the chosen app
                            // only, never the desktop/call — so it can't echo into
                            // the capture.
                            if mirror_absent {
                                ensure_loaded(&ledger_for_task, Shape::LoopbackOutOfCapture, pid)
                                    .await;
                            } else {
                                tracing::warn!(
                                    "audio routing: default-sink mirror absence was not confirmed; \
                                     refusing to load the inverse local monitor"
                                );
                            }
                            // Tell the front-end the chosen app's audio is live.
                            output::emit(output::Event::AppAudio {
                                state: AppAudioState::Routed,
                            });
                        }
                        Event::LastRoutedStreamGone => {
                            // Routed app exited/paused mid-session. Notify the
                            // front-end either way; the recovery differs by mode.
                            output::emit(output::Event::AppAudio {
                                state: AppAudioState::Lost,
                            });
                            // The shared app is gone, so its null-sink is silent:
                            // stop mirroring it to the sharer's speakers. Re-loads
                            // on the next FirstRoutedStream if the app resumes.
                            let local_monitor_absent =
                                unload_module(&ledger_for_task, Shape::LoopbackOutOfCapture).await;
                            if strict {
                                // Strict mode: do NOT restore the whole-desktop
                                // loopback. Viewers hear silence until the app
                                // produces audio again — never the rest of the
                                // desktop (call included).
                                tracing::info!(
                                    "audio routing: strict mode — last routed stream gone, leaving viewers silent"
                                );
                                continue;
                            }
                            if !local_monitor_absent {
                                tracing::warn!(
                                    "audio routing: local-monitor absence was not confirmed; \
                                     refusing to restore the inverse default-sink mirror"
                                );
                                continue;
                            }
                            // Best-effort mode: restore the default-sink loopback
                            // so the viewer hears system audio again instead of
                            // silence. Already loaded is not an error — the ledger
                            // refuses the load and `ensure_loaded` says so quietly.
                            tracing::info!(
                                "audio routing: last routed stream gone → restoring default-sink loopback"
                            );
                            ensure_loaded(&ledger_for_task, Shape::LoopbackIntoCapture, pid).await;
                        }
                    }
                }
            });
            routing.stream_router = Some(router);
            routing.event_task = Some(event_task);
        }

        // Strict per-app mode suppresses the default-sink loopback, so until the
        // chosen app's first stream routes the viewer hears *silence*. Emit an
        // initial `lost` at capture start (capture is lazy — this runs on the
        // first viewer) so the front-end can warn from the outset rather than
        // only after an app that *was* routed later stops (audit A23 P2/F1):
        // `LastRoutedStreamGone`→`lost` never fires for an app that never routed.
        if let Some(state) = initial_app_audio_state(opts) {
            crate::common::output::emit(crate::common::output::Event::AppAudio { state });
        }

        Ok(routing)
    }

    pub fn sink_name(&self) -> &str {
        &self.sink_name
    }

    /// Stop the stream router and the event task, settle anything the ledger is
    /// unsure about, then unload every module in shape order — the loopbacks
    /// before the sink they reference, because PipeWire can leave zombie links if
    /// a sink is destroyed with active inputs.
    ///
    /// The event task is **awaited, not merely aborted**. Aborting and walking
    /// away is what left orphans behind: the task's load is an await point now,
    /// so dropping its future marks the slot ambiguous rather than losing the
    /// module — but only a path that then reconciles can actually clean it up.
    /// `Drop` cannot await, which is why it is the narrower backstop.
    pub async fn shutdown(mut self) {
        // Closing is synchronous and happens first: after this point the event
        // task cannot register another mutation even if it receives one last
        // router event while shutdown is in progress.
        self.ledger.close();
        if let Some(router) = self.stream_router.take() {
            // ⚠️ Still an unbounded join: a wedged PipeWire thread parks this
            // task indefinitely. That is the pre-existing defect S3b exists for.
            // Nothing here makes it worse, and the ledger is what will make
            // bounding it safe when it lands.
            router.shutdown();
        }
        if let Some(mut task) = self.event_task.take() {
            // The router's exit drops the event senders, so the task normally
            // ends by itself. Abort is the fallback, and it is awaited through
            // `&mut JoinHandle` so the future is genuinely dropped — and with it
            // any in-flight permit — before reconciliation reads the ledger.
            // Dropping the handle instead would *detach* the task, which is how a
            // load could still land after teardown believed it was finished.
            if tokio::time::timeout(PACTL_BUDGET, &mut task).await.is_err() {
                tracing::warn!(
                    "audio routing: the event task did not finish within {PACTL_BUDGET:?}; \
                     cancelling it"
                );
                task.abort();
                let _ = task.await;
            }
        }

        // A cancelled `spawn_blocking` await detaches its worker. Every worker is
        // registered before it can be spawned, so this is a real ordering
        // boundary: reconciliation cannot overtake late module creation/removal.
        let ledger_for_wait = Arc::clone(&self.ledger);
        if let Err(e) = tokio::task::spawn_blocking(move || {
            ledger_for_wait.wait_for_operations();
        })
        .await
        {
            tracing::warn!("audio routing: module-operation wait task failed: {e}");
        }
        cleanup_modules(&self.ledger).await;

        if !self.ledger.is_clean() {
            tracing::warn!(
                settled = self.ledger.is_settled(),
                "audio routing: some audio modules could not be removed safely; \
                 `pixelpass --repair` will clean up anything left behind"
            );
        }
    }
}

impl Drop for Routing {
    /// Synchronous backstop for the paths that never reach [`Routing::shutdown`]
    /// — an error on the way up, or a panic. It cannot await, so it can neither
    /// wait for the event task nor reconcile; it unloads what the ledger can name
    /// and says so plainly when something is left unexplained.
    ///
    /// After a completed `shutdown` the ledger holds nothing and this does nothing.
    fn drop(&mut self) {
        self.ledger.close();
        if let Some(router) = self.stream_router.take() {
            router.shutdown();
        }
        if let Some(task) = self.event_task.take() {
            task.abort();
        }
        self.ledger.close_and_wait();
        cleanup_modules_blocking(&self.ledger);
        if !self.ledger.is_clean() {
            tracing::warn!(
                settled = self.ledger.is_settled(),
                "audio routing: torn down with modules that could not be removed safely; \
                 run `pixelpass --repair` to clean up anything left behind"
            );
        }
    }
}

/// The app-audio state to announce at capture start, if any. Only strict per-app
/// mode warrants one: there the loopback is suppressed, so the viewer hears
/// silence until the chosen app's first stream routes — surface that as an
/// initial `lost`. In every other mode (whole-desktop, or best-effort app
/// filtering) the loopback keeps audio flowing from the outset, so there is no
/// initial gap to report. Pure: no I/O, so the emit decision is unit-testable.
pub(super) fn initial_app_audio_state(
    opts: &HostOpts,
) -> Option<crate::common::output::AppAudioState> {
    (opts.app.is_some() && opts.strict_audio).then_some(crate::common::output::AppAudioState::Lost)
}

// ──────────────────────────────────────────────────────────────────────
// App enumeration (interactive picker source)
// ──────────────────────────────────────────────────────────────────────

/// One deduplicated app currently producing audio. The picker in
/// interactive mode shows these as the per-app capture choices.
#[derive(Debug, Clone)]
pub struct App {
    pub name: String,
    pub stream_count: u32,
}

/// Enumerate apps currently sending audio to any sink, deduplicated by
/// `application.name`. Returns an empty Vec if nothing is playing.
pub fn list_playing_apps() -> Result<Vec<App>> {
    let output = Command::new("pactl")
        .args(["-f", "json", "list", "sink-inputs"])
        .output()
        .context("failed to run `pactl -f json list sink-inputs`")?;
    if !output.status.success() {
        bail!(
            "pactl list sink-inputs failed: {}",
            String::from_utf8_lossy(&output.stderr).trim()
        );
    }
    parse_sink_inputs(&output.stdout)
}

fn parse_sink_inputs(stdout: &[u8]) -> Result<Vec<App>> {
    let entries: Vec<SinkInput> =
        serde_json::from_slice(stdout).context("pactl returned unparseable JSON")?;
    let mut counts: BTreeMap<String, u32> = BTreeMap::new();
    for entry in entries {
        let Some(name) = entry.properties.application_name else {
            continue;
        };
        let trimmed = name.trim();
        if trimmed.is_empty() {
            continue;
        }
        *counts.entry(trimmed.to_string()).or_insert(0) += 1;
    }
    Ok(counts
        .into_iter()
        .map(|(name, stream_count)| App { name, stream_count })
        .collect())
}

#[derive(serde::Deserialize)]
struct SinkInput {
    properties: SinkInputProperties,
}

#[derive(serde::Deserialize)]
struct SinkInputProperties {
    #[serde(rename = "application.name")]
    application_name: Option<String>,
}

// ──────────────────────────────────────────────────────────────────────
// pactl module helpers
// ──────────────────────────────────────────────────────────────────────

/// Mint an ownership token for one module load.
///
/// **Per load, not per session.** The nonce is what makes two loads by the same pid
/// render different arguments, which is what lets a fingerprint tell a module from
/// its replacement at the same index. A token minted once and reused for every
/// reload would be a host-session nonce and would not do that, so the counter is
/// bumped on every call and mixed with the clock.
fn owner_token(pid: u32) -> Result<repair_plan::OwnerToken> {
    use std::sync::atomic::{AtomicU64, Ordering};
    static LOADS: AtomicU64 = AtomicU64::new(0);

    let local = crate::repair::local_identity()?;
    // A nonce only has to be unlikely to repeat, not unguessable. The counter makes
    // two loads within the same clock tick distinct; the clock keeps two runs of the
    // same process distinct.
    let counter = LOADS.fetch_add(1, Ordering::Relaxed);
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64)
        .unwrap_or(0);
    Ok(repair_plan::OwnerToken {
        machine: local.machine,
        boot: local.boot,
        pid_ns: local.pid_ns,
        nonce: nanos ^ (counter << 48) ^ (u64::from(pid) << 32),
    })
}

/// Load the Pulse module for one [`Shape`] and return its index.
///
/// Both the module name and its arguments come from the shape itself
/// ([`crate::repair::plan::Shape`]) rather than being written out here, so that
/// `--repair`'s exact-form matcher and this loader are one source of truth. A
/// latency or argument change that moved only one of them would leave repair
/// silently unable to recognise the modules this build loads.
async fn load_module(ledger: &Arc<ModuleLedger>, shape: Shape, pid: u32) -> Result<Fingerprint> {
    let owner = owner_token(pid).context("could not build an audio ownership token")?;
    let permit = ledger
        .begin_load(shape, pid, owner.clone())
        .map_err(anyhow::Error::new)
        .with_context(|| format!("cannot load the {} module", shape.label()))?;
    let args = shape.render_args(pid, Some(&owner));

    // The affine permit moves into the blocking worker. Dropping this await does
    // not cancel `spawn_blocking`; the worker remains registered, owns and reaps
    // its child, and settles the slot before teardown's quiescence barrier opens.
    tokio::task::spawn_blocking(move || -> Result<Fingerprint> {
        let output = permit.with_server_operation(|| {
            let mut command = Command::new("pactl");
            command
                .arg("load-module")
                .arg(shape.module_name())
                .args(args);
            bounded_output(&mut command, PACTL_BUDGET)
        });
        let output = match output {
            Ok(output) => output,
            Err(e) => {
                // Whether the server was reached is unknown. Leaving the permit
                // unsettled makes its Drop create a reconciliation question.
                drop(permit);
                return Err(e).context("failed to run pactl load-module");
            }
        };

        if output.timed_out {
            drop(permit);
            bail!("pactl load-module did not finish within {PACTL_BUDGET:?}");
        }
        if !output.status.success() {
            let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
            if output.status.code().is_some() {
                // pactl exited normally and reported the server's refusal.
                permit.abandon();
            } else {
                drop(permit);
            }
            bail!("pactl load-module failed: {stderr}");
        }

        let id_str = String::from_utf8_lossy(&output.stdout).trim().to_string();
        // Genuinely 32-bit: this is a Pulse module index, not object.serial.
        let Ok(index) = id_str.parse::<u32>() else {
            drop(permit);
            bail!("pactl returned unexpected module ID: {id_str:?}");
        };
        let fp = permit.commit(index)?;
        tracing::info!(
            module = fp.id,
            shape = shape.label(),
            "audio routing: loaded pactl module"
        );
        Ok(fp)
    })
    .await
    .context("the pactl load worker failed")?
}

/// Load `shape` unless the slot already holds it.
///
/// A busy slot is not a failure on the oscillation path: `FirstRoutedStream` and
/// `LastRoutedStreamGone` can both ask for a module that is already in the state
/// they want, and the ledger is what decides that rather than a separate flag.
async fn ensure_loaded(ledger: &Arc<ModuleLedger>, shape: Shape, pid: u32) {
    let Err(e) = load_module(ledger, shape, pid).await else {
        return;
    };
    match e.downcast_ref::<LedgerError>() {
        Some(LedgerError::Busy { state, .. }) => tracing::debug!(
            shape = shape.label(),
            state,
            "audio routing: nothing to load, the slot is already occupied"
        ),
        _ => tracing::warn!(
            "audio routing: failed to load the {} module: {e:#}",
            shape.label()
        ),
    }
}

/// Unload whatever the ledger holds for `shape`, and record how it went.
///
/// A no-op for a slot holding nothing. An outcome that cannot be confirmed is
/// recorded as uncertain rather than assumed done, so the module keeps being
/// named until the server is asked about it.
async fn unload_module(ledger: &Arc<ModuleLedger>, shape: Shape) -> bool {
    unload_module_inner(ledger, shape, false).await
}

async fn unload_module_inner(ledger: &Arc<ModuleLedger>, shape: Shape, cleanup: bool) -> bool {
    let begun = if cleanup {
        ledger.begin_cleanup_unload(shape)
    } else {
        ledger.begin_unload(shape)
    };
    let permit = match begun {
        Ok(Some(permit)) => permit,
        Ok(None) => return true,
        Err(e) => {
            tracing::warn!(
                shape = shape.label(),
                "audio routing: refusing module unload: {e}"
            );
            return false;
        }
    };
    match tokio::task::spawn_blocking(move || finish_verified_unload(permit)).await {
        Ok(confirmed_absent) => confirmed_absent,
        Err(e) => {
            // A panicking worker drops its affine permit and therefore leaves an
            // unload reconciliation question behind.
            tracing::warn!(
                shape = shape.label(),
                "audio routing: unload worker failed: {e}"
            );
            false
        }
    }
}

fn finish_verified_unload(permit: ledger::UnloadPermit) -> bool {
    let fp = permit.fingerprint().clone();
    let result = permit.with_server_operation(|| verified_unload(&fp));
    match result {
        Ok(()) => {
            permit.finish(UnloadOutcome::Confirmed);
            true
        }
        Err(e) => {
            permit.finish(UnloadOutcome::Uncertain(format!("{e:#}")));
            false
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum UnloadPresence {
    Exact,
    Absent,
    Replaced,
}

fn unload_presence(fp: &Fingerprint, current: &[repair_plan::ModuleObservation]) -> UnloadPresence {
    match current.iter().find(|module| module.id == fp.id) {
        None => UnloadPresence::Absent,
        Some(observed) if fp.still_matches(observed) => UnloadPresence::Exact,
        Some(_) => UnloadPresence::Replaced,
    }
}

/// Re-list and unload through one verified-local Pulse connection. The exact
/// fingerprint is checked immediately before destruction; if the index vanished
/// or was reused, our module is already absent and the replacement is left alone.
fn verified_unload(fp: &Fingerprint) -> Result<()> {
    let mut session = crate::repair::introspect::PulseSession::connect()
        .context("could not connect to verify a module unload")?;
    let current = session
        .list_modules()
        .context("could not list modules immediately before unload")?;
    match unload_presence(fp, &current) {
        UnloadPresence::Exact => {}
        UnloadPresence::Absent => {
            tracing::info!(
                module = fp.id,
                shape = fp.shape.label(),
                "audio routing: tracked module was already absent"
            );
            return Ok(());
        }
        UnloadPresence::Replaced => {
            tracing::warn!(
                module = fp.id,
                shape = fp.shape.label(),
                "audio routing: module index was reused; leaving the replacement alone"
            );
            return Ok(());
        }
    }
    session
        .unload_module(fp.id)
        .with_context(|| format!("the server did not confirm unloading module #{}", fp.id))?;
    tracing::info!(
        module = fp.id,
        shape = fp.shape.label(),
        "audio routing: unloaded verified Pulse module"
    );
    Ok(())
}

/// One bounded child result. On deadline the child is killed and synchronously
/// reaped before this returns, so server reconciliation cannot overtake a late
/// `pactl` request merely because its async waiter was cancelled.
struct BoundedOutput {
    status: ExitStatus,
    stdout: Vec<u8>,
    stderr: Vec<u8>,
    timed_out: bool,
}

/// Ensures every early-return and panic after spawn kills and reaps the child.
/// The normal path marks it reaped after `try_wait`/`wait` obtained the status.
struct ReapedChild {
    child: Child,
    reaped: bool,
}

impl Drop for ReapedChild {
    fn drop(&mut self) {
        if self.reaped {
            return;
        }
        let _ = self.child.kill();
        match self.reap_within(PACTL_REAP_BUDGET) {
            Ok(Some(_)) => {}
            Ok(None) => tracing::warn!(
                "audio routing: killed pactl child was not reaped within {PACTL_REAP_BUDGET:?}"
            ),
            Err(e) => tracing::warn!("audio routing: could not reap killed pactl child: {e}"),
        }
    }
}

impl ReapedChild {
    fn reap_within(&mut self, budget: Duration) -> io::Result<Option<ExitStatus>> {
        let deadline = Instant::now() + budget;
        loop {
            if let Some(status) = self.child.try_wait()? {
                self.reaped = true;
                return Ok(Some(status));
            }
            if Instant::now() >= deadline {
                return Ok(None);
            }
            std::thread::sleep(Duration::from_millis(5));
        }
    }
}

fn bounded_output(command: &mut Command, budget: Duration) -> io::Result<BoundedOutput> {
    command.stdout(Stdio::piped()).stderr(Stdio::piped());
    let mut child = ReapedChild {
        child: command.spawn()?,
        reaped: false,
    };
    let stdout = child
        .child
        .stdout
        .take()
        .ok_or_else(|| io::Error::other("pactl stdout was not piped"))?;
    let stderr = child
        .child
        .stderr
        .take()
        .ok_or_else(|| io::Error::other("pactl stderr was not piped"))?;
    let stdout_reader = std::thread::Builder::new()
        .name("pixelpass-pactl-stdout".to_string())
        .spawn(move || {
            let mut bytes = Vec::new();
            let mut stdout = stdout;
            stdout.read_to_end(&mut bytes).map(|_| bytes)
        })?;
    let stderr_reader = std::thread::Builder::new()
        .name("pixelpass-pactl-stderr".to_string())
        .spawn(move || {
            let mut bytes = Vec::new();
            let mut stderr = stderr;
            stderr.read_to_end(&mut bytes).map(|_| bytes)
        })?;

    let deadline = Instant::now() + budget;
    let (status, timed_out) = loop {
        if let Some(status) = child.child.try_wait()? {
            child.reaped = true;
            break (status, false);
        }
        if Instant::now() >= deadline {
            let _ = child.child.kill();
            let Some(status) = child.reap_within(PACTL_REAP_BUDGET)? else {
                return Err(io::Error::new(
                    io::ErrorKind::TimedOut,
                    format!("pactl did not exit within {PACTL_REAP_BUDGET:?} after SIGKILL"),
                ));
            };
            break (status, true);
        }
        std::thread::sleep(Duration::from_millis(5));
    };
    let stdout = stdout_reader
        .join()
        .map_err(|_| io::Error::other("pactl stdout reader panicked"))??;
    let stderr = stderr_reader
        .join()
        .map_err(|_| io::Error::other("pactl stderr reader panicked"))??;
    Ok(BoundedOutput {
        status,
        stdout,
        stderr,
        timed_out,
    })
}

/// Async teardown: keep reconciling and unloading while a pass changes state.
/// This replaces the arbitrary two-round count. With loads closed, the state
/// graph is monotonic except for `Ambiguous(Unload) -> Loaded -> Ambiguous` when
/// the same unload remains uncertain; that produces an identical snapshot and
/// stops here for `--repair` rather than spinning.
async fn cleanup_modules(ledger: &Arc<ModuleLedger>) {
    loop {
        let before = ledger.snapshot();
        if ledger.is_clean() {
            break;
        }
        if let Err(e) = ledger::reconcile_pending(ledger).await {
            tracing::warn!("audio routing: could not reconcile the module ledger: {e:#}");
        }
        for fp in ledger.loaded() {
            unload_module_inner(ledger, fp.shape, true).await;
        }
        if ledger.is_clean() || ledger.snapshot() == before {
            break;
        }
    }
}

/// Synchronous teardown backstop. PulseSession bounds every connect/list/unload
/// request, and `close_and_wait` has already drained any registered pactl worker.
fn cleanup_modules_blocking(ledger: &Arc<ModuleLedger>) {
    loop {
        let before = ledger.snapshot();
        if ledger.is_clean() {
            break;
        }
        if let Err(e) = ledger::reconcile_pending_blocking(ledger) {
            tracing::warn!("audio routing: could not reconcile the module ledger: {e:#}");
        }
        for fp in ledger.loaded() {
            let permit = match ledger.begin_cleanup_unload(fp.shape) {
                Ok(Some(permit)) => permit,
                Ok(None) => continue,
                Err(e) => {
                    tracing::warn!(
                        shape = fp.shape.label(),
                        "audio routing: refusing blocking module unload: {e}"
                    );
                    continue;
                }
            };
            finish_verified_unload(permit);
        }
        if ledger.is_clean() || ledger.snapshot() == before {
            break;
        }
    }
}

// ──────────────────────────────────────────────────────────────────────
// Per-stream routing (libpipewire thread)
// ──────────────────────────────────────────────────────────────────────

/// Command from tokio → libpipewire thread.
enum Cmd {
    /// Clear `target.object` for everything we routed, then quit the
    /// MainLoop so the thread joins.
    Shutdown,
}

/// Event from libpipewire thread → tokio. The pair drives loopback
/// oscillation: unload on `FirstRoutedStream`, re-load on
/// `LastRoutedStreamGone`. Both fire on count-transitions (0→N and N→0
/// respectively), not on every change.
enum Event {
    /// At least one stream is now routed to our sink. Receiver unloads
    /// the default-sink loopback so the filtered audio isn't doubled.
    FirstRoutedStream,
    /// The last routed stream just disappeared (app closed, paused,
    /// switched output). Receiver re-loads the default-sink loopback so
    /// the viewer doesn't go silent.
    LastRoutedStreamGone,
}

/// Handle to the libpipewire stream-router thread.
pub struct StreamRouter {
    cmd_tx: pipewire::channel::Sender<Cmd>,
    thread: Option<JoinHandle<()>>,
}

impl StreamRouter {
    /// Spawn the libpipewire thread. Returns the router handle and the
    /// event receiver tokio side polls.
    fn spawn(
        filter_name: String,
        sink_name: String,
    ) -> Result<(Self, tokio::sync::mpsc::UnboundedReceiver<Event>)> {
        let (cmd_tx, cmd_rx) = pipewire::channel::channel::<Cmd>();
        let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel::<Event>();

        let thread = std::thread::Builder::new()
            .name("pixelpass-pw-router".to_string())
            .spawn(move || {
                if let Err(e) = run_router(filter_name, sink_name, cmd_rx, event_tx) {
                    tracing::warn!("audio routing: libpipewire thread exited with error: {e:#}");
                }
            })
            .context("failed to spawn libpipewire router thread")?;

        Ok((
            Self {
                cmd_tx,
                thread: Some(thread),
            },
            event_rx,
        ))
    }

    fn shutdown(mut self) {
        // Best-effort: if the send fails the thread is already gone.
        let _ = self.cmd_tx.send(Cmd::Shutdown);
        if let Some(t) = self.thread.take()
            && let Err(e) = t.join()
        {
            tracing::warn!("audio routing: pw thread join failed: {e:?}");
        }
    }
}

/// Body of the libpipewire thread. Owns MainLoop, registry listener, and
/// all PipeWire proxies for the duration of the routing session.
fn run_router(
    filter_name: String,
    sink_name: String,
    cmd_rx: pipewire::channel::Receiver<Cmd>,
    event_tx: tokio::sync::mpsc::UnboundedSender<Event>,
) -> Result<()> {
    use pipewire::{self as pw, types::ObjectType};

    let main_loop =
        pw::main_loop::MainLoopRc::new(None).context("pw main loop construction failed")?;
    let context =
        pw::context::ContextRc::new(&main_loop, None).context("pw context construction failed")?;
    let core = context
        .connect_rc(None)
        .context("pw core connect failed (is the daemon running?)")?;
    let registry = core.get_registry_rc().context("pw get_registry failed")?;

    let state = Rc::new(RefCell::new(RouterState {
        sink_serial: None,
        default_metadata: None,
        routed_node_ids: Vec::new(),
        pending: Vec::new(),
    }));

    // Cmd handler: clear metadata for routed streams, then quit.
    let main_loop_for_cmd = main_loop.clone();
    let state_for_cmd = Rc::clone(&state);
    let _cmd_recv = cmd_rx.attach(main_loop.loop_(), move |cmd| match cmd {
        Cmd::Shutdown => {
            let s = state_for_cmd.borrow();
            if let Some(meta) = &s.default_metadata {
                for &nid in &s.routed_node_ids {
                    meta.set_property(nid, "target.object", None, None);
                }
                if !s.routed_node_ids.is_empty() {
                    tracing::info!(
                        n = s.routed_node_ids.len(),
                        "audio routing: cleared target.object on routed streams before quitting"
                    );
                }
            }
            main_loop_for_cmd.quit();
        }
    });

    let filter_lower = filter_name.to_ascii_lowercase();
    let sink_name_owned = sink_name.clone();
    let registry_weak = registry.downgrade();
    let state_for_reg = Rc::clone(&state);
    let event_tx_for_reg = event_tx.clone();
    let state_for_remove = Rc::clone(&state);
    let event_tx_for_remove = event_tx.clone();

    let _reg_listener = registry
        .add_listener_local()
        .global(move |obj| {
            let Some(reg) = registry_weak.upgrade() else {
                return;
            };
            match obj.type_ {
                ObjectType::Node => {
                    let Some(props) = obj.props.as_ref() else {
                        return;
                    };
                    if props.get("node.name") == Some(sink_name_owned.as_str()) {
                        match props.get("object.serial").and_then(parse_object_serial) {
                            Some(serial) => {
                                state_for_reg.borrow_mut().sink_serial = Some(serial);
                                tracing::info!(serial, "audio routing: pixelpass sink registered");
                                try_flush(&state_for_reg, &event_tx_for_reg);
                            }
                            // Never silently: without a serial `try_flush` can
                            // never route anything, so the whole app-filter mode
                            // is dead and the only symptom is missing audio.
                            None => tracing::warn!(
                                node_id = obj.id,
                                serial = props.get("object.serial").unwrap_or("<absent>"),
                                "audio routing: pixelpass sink has no usable object.serial; \
                                 stream rerouting disabled"
                            ),
                        }
                        return;
                    }
                    if props.get("media.class") != Some("Stream/Output/Audio") {
                        return;
                    }
                    let Some(app) = props.get("application.name") else {
                        return;
                    };
                    if !app.eq_ignore_ascii_case(&filter_lower) {
                        return;
                    }
                    tracing::info!(
                        node_id = obj.id,
                        %app,
                        "audio routing: matched stream, queued for route"
                    );
                    state_for_reg.borrow_mut().pending.push(obj.id);
                    try_flush(&state_for_reg, &event_tx_for_reg);
                }
                ObjectType::Metadata => {
                    let Some(props) = obj.props.as_ref() else {
                        return;
                    };
                    if props.get("metadata.name") != Some("default") {
                        return;
                    }
                    let metadata: pw::metadata::Metadata = match reg.bind(obj) {
                        Ok(m) => m,
                        Err(e) => {
                            tracing::warn!("audio routing: bind default metadata failed: {e}");
                            return;
                        }
                    };
                    state_for_reg.borrow_mut().default_metadata = Some(metadata);
                    tracing::info!("audio routing: default metadata bound");
                    try_flush(&state_for_reg, &event_tx_for_reg);
                }
                _ => {}
            }
        })
        .global_remove(move |id| {
            handle_global_remove(&state_for_remove, &event_tx_for_remove, id);
        })
        .register();

    tracing::info!(filter = %filter_name, "audio routing: pw thread running");
    main_loop.run();
    tracing::info!("audio routing: pw thread exiting");
    Ok(())
}

/// Parse a PipeWire `object.serial` property value.
///
/// `object.serial` is a **64-bit** monotonically-increasing counter
/// (`pw_global`'s serial is `uint64_t`); it is *not* a `pw` object id
/// (those are `u32` and get recycled — the serial exists precisely so
/// that recycled ids can be disambiguated). Parsing it as `u32` silently
/// yields `None` past `u32::MAX`, which on a long-lived daemon means the
/// sink is never registered and no stream is ever routed.
///
/// Strict on purpose: PipeWire emits a bare decimal, so anything else
/// (empty, signed, whitespace-padded, non-numeric, overflowing) is a
/// property we do not understand and must not guess at. Leading zeroes
/// are accepted — they are unambiguous and parse to the same value.
pub(crate) fn parse_object_serial(raw: &str) -> Option<u64> {
    if raw.is_empty() || !raw.bytes().all(|b| b.is_ascii_digit()) {
        return None;
    }
    raw.parse::<u64>().ok()
}

struct RouterState {
    /// See [`parse_object_serial`] — 64-bit, and not interchangeable with
    /// the `u32` node ids in `routed_node_ids` / `pending`.
    sink_serial: Option<u64>,
    default_metadata: Option<pipewire::metadata::Metadata>,
    routed_node_ids: Vec<u32>,
    pending: Vec<u32>,
}

/// Drop the vanished node from `routed_node_ids` and `pending`. If it
/// was the last routed stream, emit `LastRoutedStreamGone` so the
/// tokio side restores the default-sink loopback.
fn handle_global_remove(
    state: &Rc<RefCell<RouterState>>,
    event_tx: &tokio::sync::mpsc::UnboundedSender<Event>,
    id: u32,
) {
    let mut s = state.borrow_mut();
    let was_routed = !s.routed_node_ids.is_empty();
    s.routed_node_ids.retain(|&x| x != id);
    s.pending.retain(|&x| x != id);
    if was_routed && s.routed_node_ids.is_empty() {
        tracing::info!(
            node_id = id,
            "audio routing: last routed stream disappeared"
        );
        let _ = event_tx.send(Event::LastRoutedStreamGone);
    }
}

/// Drain pending streams to the sink, but only once both prerequisites
/// (sink serial known + default metadata bound) are in place. Emits
/// `FirstRoutedStream` when routed count crosses 0→N (so it fires
/// each time the count comes back up from zero, not just the first
/// time — pairs with `LastRoutedStreamGone` to oscillate the loopback).
fn try_flush(
    state: &Rc<RefCell<RouterState>>,
    event_tx: &tokio::sync::mpsc::UnboundedSender<Event>,
) {
    let mut s = state.borrow_mut();
    let Some(serial) = s.sink_serial else { return };
    if s.default_metadata.is_none() {
        return;
    }
    if s.pending.is_empty() {
        return;
    }
    let was_empty = s.routed_node_ids.is_empty();
    let serial_str = serial.to_string();
    let pending = std::mem::take(&mut s.pending);
    if let Some(meta) = &s.default_metadata {
        for nid in &pending {
            meta.set_property(*nid, "target.object", Some("Spa:Id"), Some(&serial_str));
            tracing::info!(
                node_id = *nid,
                sink_serial = serial,
                "audio routing: stream routed to pixelpass sink"
            );
        }
    }
    s.routed_node_ids.extend(pending);
    if was_empty && !s.routed_node_ids.is_empty() {
        let _ = event_tx.send(Event::FirstRoutedStream);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::host::ledger::SlotState;
    use crate::repair::plan::{ModuleObservation, classify};

    /// Whole-desktop routing: no app filter, so no PipeWire thread and no event
    /// task — just the null-sink and its default-sink loopback.
    fn whole_desktop_opts() -> HostOpts {
        HostOpts {
            window: false,
            app: None,
            strict_audio: false,
            display_server: None,
            quality: crate::cli::Quality::Auto,
            bitrate: None,
            framerate: None,
            max_height: None,
            no_hwencode: false,
            max_viewers: None,
            interactive: false,
            relay: None,
        }
    }

    /// The module table exactly as `--repair` observes it.
    fn module_snapshot() -> Vec<(u32, String, String)> {
        let mut session =
            crate::repair::introspect::PulseSession::connect().expect("a local Pulse server");
        session
            .list_modules()
            .expect("the server lists its modules")
            .into_iter()
            .map(|m| (m.id, m.name, m.args))
            .collect()
    }

    #[test]
    fn normal_unload_requires_the_full_fingerprint_not_only_the_index() {
        fn observation(id: u32, pid: u32, nonce: u64) -> ModuleObservation {
            let token = repair_plan::OwnerToken {
                machine: "abc123".to_string(),
                boot: "def456".to_string(),
                pid_ns: 4_026_531_836,
                nonce,
            };
            ModuleObservation::new(
                id,
                Shape::LoopbackIntoCapture.module_name(),
                &repair_plan::recorded_argument(
                    &Shape::LoopbackIntoCapture.render_args(pid, Some(&token)),
                ),
            )
        }

        let ours = observation(5, 42, 7);
        let fp = classify(&ours).expect("the fixture is canonical");
        assert_eq!(
            unload_presence(&fp, std::slice::from_ref(&ours)),
            UnloadPresence::Exact
        );
        assert_eq!(unload_presence(&fp, &[]), UnloadPresence::Absent);

        // Same live index, but another perfectly canonical host module. This is
        // the non-vacuous reuse case: an id-only normal unload would destroy it.
        let replacement = observation(5, 99, 8);
        assert_eq!(
            unload_presence(&fp, std::slice::from_ref(&replacement)),
            UnloadPresence::Replaced
        );
    }

    /// A/B against the live graph: routing must leave the module table exactly
    /// as it found it. The same shape as `--repair`'s field gate, because the
    /// property is the same one — nothing of ours outlives the session.
    #[tokio::test]
    #[ignore = "loads real Pulse modules; run with --ignored --test-threads=1"]
    async fn live_teardown_leaves_the_module_table_as_it_found_it() {
        let before = module_snapshot();
        let routing = Routing::start(&whole_desktop_opts())
            .await
            .expect("routing starts");

        let during = module_snapshot();
        let ours: Vec<_> = during
            .iter()
            .filter(|m| !before.iter().any(|b| b.0 == m.0))
            .collect();
        assert_eq!(
            ours.len(),
            2,
            "the null-sink and its default-sink loopback must both be loaded"
        );
        for (id, name, args) in ours {
            let fp = classify(&ModuleObservation::new(*id, name, args))
                .expect("a module we loaded must match one of our canonical forms");
            assert!(
                fp.owner.is_some(),
                "every module we load carries an owner token, or --repair cannot \
                 attribute it to this host's pid namespace"
            );
        }

        routing.shutdown().await;
        assert_eq!(
            module_snapshot(),
            before,
            "teardown must leave the module table byte-identical"
        );
    }

    /// The orphan race, staged against a real server: a load cancelled while
    /// `pactl` is in flight must still be findable and removable.
    ///
    /// Whether the server got as far as creating the module is genuinely racy,
    /// and that is the point — the gate does not care which way it went, only
    /// that the ledger can account for both. With the permit's `Drop` disarmed
    /// and the module created, the final comparison fails.
    #[tokio::test]
    #[ignore = "loads real Pulse modules; run with --ignored --test-threads=1"]
    async fn live_a_cancelled_load_is_reconciled_not_orphaned() {
        let before = module_snapshot();
        let ledger = ModuleLedger::new();
        let pid = std::process::id();

        let ledger_for_task = Arc::clone(&ledger);
        let task = tokio::spawn(async move {
            let _ = load_module(&ledger_for_task, Shape::LegacyCaptureSink, pid).await;
        });
        // Wait until the affine permit is registered. A single yield is not a
        // scheduling guarantee and made the old version of this gate capable of
        // aborting before the task had started.
        tokio::time::timeout(PACTL_BUDGET, async {
            while matches!(ledger.state(Shape::LegacyCaptureSink), SlotState::Vacant) {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("the load registers its operation");
        task.abort();
        let _ = task.await;

        // Cancellation detaches `spawn_blocking`; closing plus this registered-
        // operation wait is the ordering boundary that prevents reconciliation
        // from overtaking the worker's late server mutation.
        ledger.close();
        let ledger_for_wait = Arc::clone(&ledger);
        tokio::task::spawn_blocking(move || ledger_for_wait.wait_for_operations())
            .await
            .expect("the operation wait runs");

        // The worker either committed a known module or left a reconciliation
        // question. Both are correct; vacancy here would mean the live operation
        // disappeared from the ledger.
        assert_eq!(
            ledger.pending().len() + ledger.loaded().len(),
            1,
            "the cancelled load must retain exactly one tracked outcome"
        );

        cleanup_modules(&ledger).await;

        assert!(
            ledger.is_clean(),
            "every tracked module must be removed, not merely explained"
        );
        assert_eq!(
            module_snapshot(),
            before,
            "a cancelled load must leave nothing behind"
        );
    }

    #[test]
    fn object_serial_parses_past_u32() {
        // The regression this fix exists for: a serial one past `u32::MAX`
        // used to parse as `None` and silently disable rerouting.
        let beyond = u64::from(u32::MAX) + 1;
        assert_eq!(parse_object_serial(&beyond.to_string()), Some(beyond));
        assert_eq!(
            parse_object_serial(&u64::MAX.to_string()),
            Some(u64::MAX),
            "the full 64-bit range must round-trip"
        );
    }

    #[test]
    fn object_serial_accepts_ordinary_serials() {
        // Without this the valid cases are only 1, 10 and 20 digits long, and
        // a length-gated mutant (`if (2..10).contains(&raw.len()) { None }`)
        // survives the whole suite while rejecting every serial a freshly
        // started daemon actually hands out. (Codex, round 1.)
        for serial in 0_u64..=1024 {
            assert_eq!(parse_object_serial(&serial.to_string()), Some(serial));
        }
        assert_eq!(parse_object_serial("123456789"), Some(123_456_789));
        assert_eq!(
            parse_object_serial("007"),
            Some(7),
            "leading zeroes are fine"
        );
    }

    #[test]
    fn object_serial_boundary_values() {
        assert_eq!(parse_object_serial("0"), Some(0));
        assert_eq!(parse_object_serial("1"), Some(1));
        let max32 = u64::from(u32::MAX);
        assert_eq!(parse_object_serial(&max32.to_string()), Some(max32));
        assert_eq!(
            parse_object_serial(&(max32 - 1).to_string()),
            Some(max32 - 1)
        );
    }

    #[test]
    fn object_serial_round_trips_through_the_metadata_string() {
        // `try_flush` writes the serial back out as a decimal string for
        // `target.object`; widening must not introduce a formatting change.
        for raw in ["0", "4294967296", "18446744073709551615"] {
            let parsed = parse_object_serial(raw).expect("valid serial");
            assert_eq!(parsed.to_string(), raw);
        }
    }

    #[test]
    fn object_serial_rejects_malformed() {
        for raw in [
            "",
            " 12",
            "12 ",
            "+12",
            "-1",
            "1.0",
            "0x10",
            "12a",
            "abc",
            // u64::MAX + 1 — overflow must be rejected, not wrapped.
            "18446744073709551616",
        ] {
            assert_eq!(parse_object_serial(raw), None, "should reject {raw:?}");
        }
    }
}