pixelpass/src/host/observer/adapter.rs

//! PipeWire I/O adapter for the pure registry observer.
//!
//! This module owns a read-only PipeWire main-loop thread, translates registry
//! callbacks into [`RegEvent`]s, and publishes the latest [`Projection`] for
//! consumers running outside the PipeWire thread.

use super::classify::DeviceClaim;
use super::{EventKind, LinkEndpoints, NodeObservation, Projection, RegEvent, RegistryModel};
use crate::host::audio::parse_object_serial;
use crate::host::taint::snapshot::{
    ClientSnapshot, GlobalId, MediaRole, NodeProps, PortDirection, PortSnapshot, Serial,
};
use anyhow::{Context, Result};
use pipewire::{self as pw, types::ObjectType};
use std::cell::{Cell, RefCell};
use std::collections::{BTreeMap, VecDeque};
use std::rc::Rc;
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

const READINESS_TIMEOUT_MILLIS: u64 = 2_000;
const TICK_INTERVAL: Duration = Duration::from_millis(250);

/// A consumer that sees **every** projection, one per applied registry event,
/// on the observer thread.
///
/// This exists because polling [`RegistryObserverHandle::latest`] coalesces, and
/// some consumers cannot tolerate that. Phase 4's AEC validator is the concrete
/// case: it detects a module unload by observing the *empty gap* before the next
/// module appears, and PipeWire reuses module indices verbatim across an
/// unload/reload (v3.4 §5.2 correction 3), so a consumer that misses the gap
/// silently aliases a fresh module onto a dead module's validated identity.
///
/// **Implementations run inline on the PipeWire loop thread.** Whatever they do
/// delays the next registry callback, so they must be bounded and must not
/// block. The phase-5 audit is the only implementor and measures its own cost
/// for exactly this reason.
pub trait ProjectionSink: Send {
    /// `now_us` is monotonic microseconds since the observer started — the same
    /// clock that drives [`RegEvent::Tick`], so a sink's notion of time cannot
    /// drift from the readiness epoch's.
    fn on_projection(&mut self, projection: &Projection, kind: EventKind, now_us: u64);
}

/// Tokio-side access to the observer's most recent coherent projection.
pub struct RegistryObserverHandle {
    latest: Arc<Mutex<Option<Projection>>>,
    shutdown_tx: pw::channel::Sender<()>,
    thread: Option<JoinHandle<()>>,
}

impl RegistryObserverHandle {
    /// Spawn the read-only PipeWire registry observer.
    pub fn spawn() -> Result<Self> {
        Self::spawn_with_sink(None)
    }

    /// Spawn the observer with a per-event [`ProjectionSink`] attached.
    ///
    /// The sink is moved onto the observer thread and dropped when that thread
    /// exits, which is what lets a sink emit a final summary on shutdown without
    /// the caller arranging one.
    pub fn spawn_with_sink(sink: Option<Box<dyn ProjectionSink>>) -> Result<Self> {
        let latest = Arc::new(Mutex::new(None));
        let latest_for_thread = Arc::clone(&latest);
        let (shutdown_tx, shutdown_rx) = pw::channel::channel::<()>();
        let thread = std::thread::Builder::new()
            .name("pixelpass-pw-observer".to_string())
            .spawn(move || {
                if let Err(e) = run_observer(latest_for_thread, shutdown_rx, sink) {
                    tracing::warn!(
                        "registry observer: libpipewire thread exited with error: {e:#}"
                    );
                }
            })
            .context("failed to spawn libpipewire registry observer thread")?;

        Ok(Self {
            latest,
            shutdown_tx,
            thread: Some(thread),
        })
    }

    /// Return a clone of the latest projection, or `None` before the first
    /// registry event has been applied.
    pub fn latest(&self) -> Option<Projection> {
        self.latest
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .clone()
    }
}

impl Drop for RegistryObserverHandle {
    fn drop(&mut self) {
        let _ = self.shutdown_tx.send(());
        if let Some(thread) = self.thread.take()
            && let Err(e) = thread.join()
        {
            tracing::warn!("registry observer: pw thread join failed: {e:?}");
        }
    }
}

struct BoundLink {
    _proxy: pw::link::Link,
    _listener: pw::link::LinkListener,
}

#[derive(Default)]
struct LiveGlobal {
    bound_link: Option<BoundLink>,
}

struct ObserverState {
    model: RegistryModel,
    latest: Arc<Mutex<Option<Projection>>>,
    last_candidate: Option<u32>,
    live_globals: BTreeMap<GlobalId, VecDeque<LiveGlobal>>,
    sink: Option<Box<dyn ProjectionSink>>,
    started_at: Instant,
}

impl ObserverState {
    /// `started_at` is the observer's single time origin, shared with the
    /// readiness tick timer — so a sink's `now_us` and a `RegEvent::Tick`'s
    /// `now` are the same clock, not two that drift.
    fn new(
        latest: Arc<Mutex<Option<Projection>>>,
        sink: Option<Box<dyn ProjectionSink>>,
        started_at: Instant,
    ) -> Self {
        Self {
            model: RegistryModel::new(0, READINESS_TIMEOUT_MILLIS),
            latest,
            last_candidate: None,
            live_globals: BTreeMap::new(),
            sink,
            started_at,
        }
    }

    fn apply(&mut self, event: RegEvent) {
        // Taken before the model consumes the event: the sink is told what kind
        // of observation produced the projection, and deriving that from the
        // event itself is what stops the two from ever disagreeing.
        let kind = event.kind();
        self.model.apply(event);

        let candidate = self.model.pulse_pid_candidate();
        if candidate != self.last_candidate {
            self.last_candidate = candidate;
            if let Some(pid) = candidate {
                let comm = std::fs::read_to_string(format!("/proc/{pid}/comm"))
                    .ok()
                    .map(|comm| comm.trim_end_matches(['\r', '\n']).to_string());
                // Folded into the model directly rather than through `apply`, so
                // one registry event still yields exactly one sink call — the
                // no-coalescing contract cuts both ways, and a *duplicated*
                // observation would make the O5 event rate a fiction.
                self.model.apply(RegEvent::ProcCommProbed { pid, comm });
            }
        }

        self.publish(kind);
    }

    fn publish(&mut self, kind: EventKind) {
        let projection = self.model.project();
        if let Some(sink) = self.sink.as_mut() {
            let now_us = u64::try_from(self.started_at.elapsed().as_micros()).unwrap_or(u64::MAX);
            sink.on_projection(&projection, kind, now_us);
        }
        // Published after the sink has seen it, so the projection is moved
        // rather than cloned — the snapshot is the largest thing the observer
        // owns and this runs on every event.
        *self
            .latest
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(projection);
    }

    /// Record the global's id and apply its add event as one step, so the
    /// bound-link FIFO stays provably lockstep with the model's own `live_ids`
    /// index. Recording only on *applied* adds (never on unknown object types
    /// or globals dropped for a missing serial) is what keeps the two id
    /// queues the same length per id — otherwise a phantom slot ahead of a
    /// bound Link would be popped on removal, leaking that Link's proxy.
    fn add(&mut self, id: GlobalId, event: RegEvent) {
        self.live_globals
            .entry(id)
            .or_default()
            .push_back(LiveGlobal::default());
        self.apply(event);
    }

    fn attach_bound_link(&mut self, id: GlobalId, bound_link: BoundLink) {
        let Some(global) = self.live_globals.get_mut(&id).and_then(VecDeque::back_mut) else {
            tracing::warn!(
                global_id = id.0,
                "registry observer: link bind completed without a live global slot"
            );
            return;
        };
        global.bound_link = Some(bound_link);
    }

    fn remove_global(&mut self, id: GlobalId) -> Option<BoundLink> {
        let (bound_link, empty) = {
            let globals = self.live_globals.get_mut(&id)?;
            let bound_link = globals.pop_front().and_then(|global| global.bound_link);
            (bound_link, globals.is_empty())
        };
        if empty {
            self.live_globals.remove(&id);
        }
        bound_link
    }
}

fn run_observer(
    latest: Arc<Mutex<Option<Projection>>>,
    shutdown_rx: pw::channel::Receiver<()>,
    sink: Option<Box<dyn ProjectionSink>>,
) -> Result<()> {
    let started_at = Instant::now();
    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(ObserverState::new(latest, sink, started_at)));

    let main_loop_for_shutdown = main_loop.clone();
    let _shutdown_receiver = shutdown_rx.attach(main_loop.loop_(), move |()| {
        main_loop_for_shutdown.quit();
    });

    let pending_sync = Rc::new(Cell::new(None));
    let pending_sync_for_done = Rc::clone(&pending_sync);
    let state_for_done = Rc::clone(&state);
    let _core_listener = core
        .add_listener_local()
        .done(move |id, seq| {
            if id == pw::core::PW_ID_CORE && pending_sync_for_done.get() == Some(seq) {
                pending_sync_for_done.set(None);
                state_for_done.borrow_mut().apply(RegEvent::ServerSynced);
            }
        })
        .error(|id, seq, res, message| {
            tracing::warn!(
                id,
                seq,
                result = res,
                %message,
                "registry observer: PipeWire core error"
            );
        })
        .register();

    let registry_weak = registry.downgrade();
    let state_for_global = Rc::clone(&state);
    let state_for_remove = Rc::clone(&state);
    let _registry_listener = registry
        .add_listener_local()
        .global(move |obj| {
            let id = GlobalId(obj.id);

            match obj.type_ {
                ObjectType::Node => {
                    let Some(props) = obj.props.as_ref() else {
                        tracing::warn!(
                            node_id = obj.id,
                            "registry observer: Node has no properties; dropping"
                        );
                        return;
                    };
                    let Some(serial) = parse_serial(obj.id, "Node", props.get("object.serial"))
                    else {
                        return;
                    };
                    let node_props = NodeProps {
                        peerspeak_owned: truthy(props.get("peerspeak.owned")),
                        pulse_module_id: props
                            .get("pulse.module.id")
                            .and_then(|value| value.parse::<u64>().ok()),
                        link_group: props.get("node.link-group").map(str::to_owned),
                        client_id: props
                            .get("client.id")
                            .and_then(|value| value.parse::<u32>().ok())
                            .map(GlobalId),
                        process_id: props
                            .get("application.process.id")
                            .and_then(|value| value.parse::<u32>().ok()),
                        passthrough: truthy(props.get("node.passthrough")),
                        session_device: false,
                    };
                    let observation = NodeObservation {
                        serial,
                        id,
                        name: props.get("node.name").map(str::to_owned),
                        role: MediaRole::parse(props.get("media.class")),
                        props: node_props,
                        device_claim: DeviceClaim {
                            device_id: props
                                .get("device.id")
                                .and_then(|value| value.parse::<u32>().ok())
                                .map(GlobalId),
                            device_api: props.get("device.api").map(str::to_owned),
                            factory_name: props.get("factory.name").map(str::to_owned),
                            alsa_driver_name: props.get("alsa.driver_name").map(str::to_owned),
                        },
                    };
                    state_for_global
                        .borrow_mut()
                        .add(id, RegEvent::NodeAdded(observation));
                }
                ObjectType::Port => {
                    let Some(props) = obj.props.as_ref() else {
                        tracing::warn!(
                            port_id = obj.id,
                            "registry observer: Port has no properties; dropping"
                        );
                        return;
                    };
                    let Some(serial) = parse_serial(obj.id, "Port", props.get("object.serial"))
                    else {
                        return;
                    };
                    let Some(node) = props
                        .get("node.id")
                        .and_then(|value| value.parse::<u32>().ok())
                        .map(GlobalId)
                    else {
                        tracing::warn!(
                            port_id = obj.id,
                            node_id = props.get("node.id").unwrap_or("<absent>"),
                            "registry observer: Port has no usable node.id; dropping"
                        );
                        return;
                    };
                    let direction = match props.get("port.direction") {
                        Some("in") => PortDirection::In,
                        Some("out") => PortDirection::Out,
                        direction => {
                            tracing::warn!(
                                port_id = obj.id,
                                direction = direction.unwrap_or("<absent>"),
                                "registry observer: Port has no usable direction; dropping"
                            );
                            return;
                        }
                    };
                    state_for_global.borrow_mut().add(
                        id,
                        RegEvent::PortAdded(PortSnapshot {
                            serial,
                            id,
                            node,
                            direction,
                            exclusive: truthy(props.get("port.exclusive")),
                            monitor: truthy(props.get("port.monitor")),
                        }),
                    );
                }
                ObjectType::Client => {
                    let Some(props) = obj.props.as_ref() else {
                        tracing::warn!(
                            client_id = obj.id,
                            "registry observer: Client has no properties; dropping"
                        );
                        return;
                    };
                    let Some(serial) = parse_serial(obj.id, "Client", props.get("object.serial"))
                    else {
                        return;
                    };
                    state_for_global.borrow_mut().add(
                        id,
                        RegEvent::ClientAdded(ClientSnapshot {
                            serial,
                            id,
                            sec_pid: props
                                .get("pipewire.sec.pid")
                                .and_then(|value| value.parse::<u32>().ok()),
                        }),
                    );
                }
                ObjectType::Device => {
                    state_for_global
                        .borrow_mut()
                        .add(id, RegEvent::DeviceAdded { id });
                }
                ObjectType::Link => {
                    let Some(props) = obj.props.as_ref() else {
                        tracing::warn!(
                            link_id = obj.id,
                            "registry observer: Link has no properties; dropping"
                        );
                        return;
                    };
                    let Some(serial) = parse_serial(obj.id, "Link", props.get("object.serial"))
                    else {
                        return;
                    };
                    let endpoints = link_endpoints_from_props(props);
                    state_for_global.borrow_mut().add(
                        id,
                        RegEvent::LinkAdded {
                            serial,
                            id,
                            endpoints,
                        },
                    );
                    if endpoints.is_some() {
                        return;
                    }

                    let Some(registry) = registry_weak.upgrade() else {
                        return;
                    };
                    let link: pw::link::Link = match registry.bind(obj) {
                        Ok(link) => link,
                        Err(e) => {
                            tracing::warn!(
                                link_id = obj.id,
                                "registry observer: failed to bind Link for endpoints: {e}"
                            );
                            return;
                        }
                    };
                    let resolved = Rc::new(Cell::new(false));
                    let resolved_for_info = Rc::clone(&resolved);
                    let state_for_info = Rc::downgrade(&state_for_global);
                    let listener = link
                        .add_listener_local()
                        .info(move |info| {
                            if resolved_for_info.replace(true) {
                                return;
                            }
                            let endpoints = LinkEndpoints {
                                output_node: GlobalId(info.output_node_id()),
                                input_node: GlobalId(info.input_node_id()),
                                output_port: optional_global_id(info.output_port_id()),
                                input_port: optional_global_id(info.input_port_id()),
                            };
                            if let Some(state) = state_for_info.upgrade() {
                                state
                                    .borrow_mut()
                                    .apply(RegEvent::LinkEndpointsResolved { serial, endpoints });
                            }
                        })
                        .register();
                    state_for_global.borrow_mut().attach_bound_link(
                        id,
                        BoundLink {
                            _proxy: link,
                            _listener: listener,
                        },
                    );
                }
                _ => {}
            }
        })
        .global_remove(move |id| {
            let id = GlobalId(id);
            let bound_link = state_for_remove.borrow_mut().remove_global(id);
            state_for_remove
                .borrow_mut()
                .apply(RegEvent::Removed { id });
            drop(bound_link);
        })
        .register();

    pending_sync.set(Some(
        core.sync(0)
            .context("registry observer: initial core.sync failed")?,
    ));

    let state_for_tick = Rc::clone(&state);
    let timer = main_loop.loop_().add_timer(move |_| {
        let now = u64::try_from(started_at.elapsed().as_millis()).unwrap_or(u64::MAX);
        state_for_tick.borrow_mut().apply(RegEvent::Tick { now });
    });
    timer
        .update_timer(Some(TICK_INTERVAL), Some(TICK_INTERVAL))
        .into_result()
        .context("registry observer: failed to arm readiness timer")?;

    tracing::info!("registry observer: pw thread running");
    main_loop.run();
    tracing::info!("registry observer: pw thread exiting");
    Ok(())
}

fn parse_serial(id: u32, kind: &str, raw: Option<&str>) -> Option<Serial> {
    match raw.and_then(parse_object_serial) {
        Some(serial) => Some(Serial(serial)),
        None => {
            tracing::warn!(
                global_id = id,
                object_type = kind,
                serial = raw.unwrap_or("<absent>"),
                "registry observer: global has no usable object.serial; dropping"
            );
            None
        }
    }
}

fn truthy(value: Option<&str>) -> bool {
    value.is_some_and(|value| value != "false" && value != "0")
}

fn link_endpoints_from_props(props: &pw::spa::utils::dict::DictRef) -> Option<LinkEndpoints> {
    let output_node = props.get("link.output.node")?.parse::<u32>().ok()?;
    let input_node = props.get("link.input.node")?.parse::<u32>().ok()?;
    Some(LinkEndpoints {
        output_node: GlobalId(output_node),
        input_node: GlobalId(input_node),
        output_port: props
            .get("link.output.port")
            .and_then(|value| value.parse::<u32>().ok())
            .map(GlobalId),
        input_port: props
            .get("link.input.port")
            .and_then(|value| value.parse::<u32>().ok())
            .map(GlobalId),
    })
}

fn optional_global_id(id: u32) -> Option<GlobalId> {
    (id != pw::constants::ID_ANY).then_some(GlobalId(id))
}

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

    struct PactlModule {
        id: Option<u32>,
    }

    impl PactlModule {
        fn load(name: &str, args: &[String]) -> Self {
            let output = Command::new("pactl")
                .arg("load-module")
                .arg(name)
                .args(args)
                .output()
                .expect("pactl must be installed for the live observer test");
            assert!(
                output.status.success(),
                "pactl load-module {name} failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
            let id = String::from_utf8(output.stdout)
                .expect("pactl module id must be UTF-8")
                .trim()
                .parse::<u32>()
                .expect("pactl module id must be a u32");
            Self { id: Some(id) }
        }

        fn unload(mut self) {
            let id = self.id.take().expect("module must still be loaded");
            let output = Command::new("pactl")
                .arg("unload-module")
                .arg(id.to_string())
                .output()
                .expect("pactl must be installed for the live observer test");
            assert!(
                output.status.success(),
                "pactl unload-module {id} failed: {}",
                String::from_utf8_lossy(&output.stderr).trim()
            );
        }
    }

    impl Drop for PactlModule {
        fn drop(&mut self) {
            if let Some(id) = self.id.take() {
                let _ = Command::new("pactl")
                    .arg("unload-module")
                    .arg(id.to_string())
                    .output();
            }
        }
    }

    fn wait_for(
        observer: &RegistryObserverHandle,
        predicate: impl Fn(&Projection) -> bool,
    ) -> Projection {
        let deadline = Instant::now() + Duration::from_secs(10);
        while Instant::now() < deadline {
            if let Some(projection) = observer.latest()
                && predicate(&projection)
            {
                return projection;
            }
            std::thread::sleep(Duration::from_millis(50));
        }
        panic!("timed out waiting for the registry projection");
    }

    fn has_node(projection: &Projection, name: &str) -> bool {
        projection
            .snapshot
            .nodes()
            .any(|node| node.name.as_deref() == Some(name))
    }

    #[test]
    #[ignore = "needs live pipewire"]
    fn live_topology_diff_tracks_null_sink_and_loopback() {
        pw::init();
        let observer = RegistryObserverHandle::spawn().expect("observer thread must spawn");
        let baseline = wait_for(&observer, |projection| projection.graph_ready);
        let baseline_links = baseline.snapshot.links().count();

        let unique = format!("pixelpass_observer_test_{}", std::process::id());
        let capture_name = format!("{unique}_capture");
        let playback_name = format!("{unique}_playback");
        let null_sink = PactlModule::load("module-null-sink", &[format!("sink_name={unique}")]);
        let with_sink = wait_for(&observer, |projection| has_node(projection, &unique));
        let sink_links = with_sink.snapshot.links().count();

        let loopback = PactlModule::load(
            "module-loopback",
            &[
                format!("source={unique}.monitor"),
                format!("sink={unique}"),
                format!("source_output_properties=node.name={capture_name}"),
                format!("sink_input_properties=node.name={playback_name}"),
            ],
        );
        wait_for(&observer, |projection| {
            has_node(projection, &capture_name)
                && has_node(projection, &playback_name)
                && projection.snapshot.links().count() > sink_links
        });

        loopback.unload();
        null_sink.unload();
        wait_for(&observer, |projection| {
            !has_node(projection, &unique)
                && !has_node(projection, &capture_name)
                && !has_node(projection, &playback_name)
                && projection.snapshot.links().count() <= baseline_links
        });
    }
}