audiofox 1.0.1

// SPDX-License-Identifier: GPL-2.0-or-later WITH Classpath-exception-2.0
// Audiofox, the audio capturer!
// Copyright (c) 2026 foxberries

use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::rc::Rc;
use std::str::FromStr;
use std::sync::{OnceLock, mpsc};

use anyhow::{Context, anyhow};
use log::Level::{Debug, Trace, Warn};
use log::{debug, error, info, log};
use pw::keys;
use pw::registry::{GlobalObject, Listener, RegistryBox};
use pw::spa::utils::dict::DictRef;
use pw::types::ObjectType;
use uuid::Uuid;

use crate::common::audiofox::AUDIOFOX;
use crate::common::result::{Level, LeveledResult, Result};
use crate::common::source::AudioSource;
use crate::platform::linux::command::{CmdListener, Response, StreamStates};
use crate::platform::linux::wrapper::{Channel, Direction};

/// The system's name for the link factory object.
pub static LINK_FACTORY: OnceLock<Box<str>> = OnceLock::new();

/// An ID of a node detected by the registry, valid for the lifetime of the node.
pub type RegId = u32;

/// The name of an audio port, used to link its parent node to Audiofox.
pub type PortName = Rc<str>;

/// The name of a node detected by the registry.
pub type RegName = Rc<str>;

/// A node detected by the registry.
#[derive(Default)]
pub struct RegNode {
    pub name: RegName,
    pub out_left: PortName,
    pub out_right: PortName,
}

/// Values found by the registry. Its fields are fully populated before [`start()`](crate::start())
/// returns, and are updated throughout the lifetime of the [`AudioManager`](crate::AudioManager) instance.
#[derive(Default)]
pub struct RegState {
    pub nodes: RefCell<HashMap<RegId, RegNode>>,
    /// Internal flag for if an Audiofox node is currently being initialized.
    init: Cell<bool>,
    /// Internal flag for if a node is currently being reconnected to a stream.
    recon: Cell<bool>,
}

/// Port information required to initialize a stream.
struct RegPort {
    node_id: u32,
    name: PortName,
    direction: Direction,
    channel: Channel,
}

impl RegPort {
    /// Get information about this port.
    fn get_info(id: u32, props: &DictRef) -> Result<Self> {
        let node_id = props.get(*keys::NODE_ID).with_context(|| {
            format!("port id={id}: missing required property node.id")
        })?.parse::<u32>().context("failed to parse node.id as u32")?;

        let name = props.get(*keys::PORT_NAME).with_context(|| {
            format!("port node.id={node_id}: missing required property node.name")
        })?.into();

        let direction = props.get(*keys::PORT_DIRECTION).with_context(|| {
            format!("port node.id={node_id}, name={name}: missing required property port.direction")
        })?.parse::<Direction>()?;

        let channel = props.get(*keys::AUDIO_CHANNEL).with_context(|| {
            format!("port node.id={node_id}, name={name}, direction={direction}: missing required property audio.channel")
        })?.parse::<Channel>()?;

        Ok(Self { node_id, name, direction, channel })
    }
}

impl RegState {
    /// Create a new registry listener with this `RegState`.
    pub fn monitor(
        self: Rc<Self>, registry: &RegistryBox, streams: &Rc<StreamStates>,
        sender: &mpsc::Sender<Response>, listener: &Rc<CmdListener>,
    ) -> Listener {
        registry
            .add_listener_local()
            .global(self.clone().start(streams.clone(), sender.clone(), listener.clone()))
            .global_remove(self.on_remove())
            .register()
    }

    /// Initialize a new `RegState`, logging any errors received throughout its lifetime.
    fn start(
        self: Rc<Self>, streams: Rc<StreamStates>, sender: mpsc::Sender<Response>,
        listener: Rc<CmdListener>,
    ) -> impl Fn(&GlobalObject<&DictRef>) {
        move |global| {
            if let Err(err) = self.on_add(global, &streams, &sender, &listener) {
                log!(err.level, "{}", err.error);
            }
        }
    }

    /// Called when an object is added to the registry.
    fn on_add(
        &self, global: &GlobalObject<&DictRef>, streams: &StreamStates,
        sender: &mpsc::Sender<Response>, listener: &CmdListener,
    ) -> LeveledResult<()> {
        // Get the global object's properties
        let Some(props) = global.props else {
            send_logged!(sender, Response::Error(anyhow!("global object has no properties")));
            return Ok(());
        };

        // If the added object is a node
        if global.type_ == ObjectType::Node {
            // Get this node's required properties
            let id = global.id;

            let name = props.get(*keys::NODE_NAME).with_context(|| {
                format!("skipping node id={id}: missing required property node.name")
            }).level(Trace)?;

            let class = props.get(*keys::MEDIA_CLASS).with_context(|| {
                format!("node id={id}, name={name}: missing required property media.class")
            }).level(Trace)?;

            // If this node is named audiofox and has a NODE_SESSION attribute, it may be our target
            if name == AUDIOFOX && let Some(session) = props.get(*keys::NODE_SESSION) {
                self.init_stream_node(id, session, streams)?;
            }
            // If this is a capturable node, add it to the state
            else if class.contains("Stream/Output/Audio") {
                info!("adding node id={id}, name={name}");

                // Place inside the HashMap for internal use
                self.nodes.borrow_mut().insert(id, RegNode { name: name.into(), ..RegNode::default() });

                // Check if this node needs to be reconnected to an existing stream
                if streams.borrow().values().flat_map(|s| s.names.iter())
                    .any(|n| n.as_ref() == name) {
                    self.recon.set(true);
                }
            }
            // We do not support other types of node
            else {
                debug!("node id={id}, name={name}, class={class}: unsupported media.class");
            }
        }

        // If the added object is a port
        else if global.type_ == ObjectType::Port {
            let port = RegPort::get_info(global.id, props).level(Trace)?;

            // If we are currently initializing audiofox, this port may be our target
            if self.init.get() {
                self.init_stream_port(port, streams, sender);
            }
            // Otherwise, process as a standard port
            else {
                let RegPort { node_id, name, channel, direction } = port;
                let mut nodes = self.nodes.borrow_mut();
                let reg_node = nodes.get_mut(&node_id).with_context(|| {
                    format!("port id={node_id}, name={name}, channel={channel}: not associated with capturable node")
                }).level(Trace)?;

                if direction == Direction::Out {
                    match channel {
                        Channel::FL => reg_node.out_left = name,
                        Channel::FR => reg_node.out_right = name,
                    }
                }

                // If we are currently reconnecting to a stream, this port may be our target
                if self.recon.get() && !(reg_node.out_left.is_empty() || reg_node.out_right.is_empty()) {
                    let name = reg_node.name.clone();
                    drop(nodes); // Drop so CmdListener can safely query

                    self.reconnect(node_id, &name, streams, listener)?;
                }
            }
        }

        // If the added object is a link factory
        else if LINK_FACTORY.get().is_none() && global.type_ == ObjectType::Factory {
            // Initialize the name for later use
            if props.get(*keys::FACTORY_TYPE_NAME) == Some(ObjectType::Link.to_str()) {
                if let Some(name) = props.get(*keys::FACTORY_NAME) {
                    LINK_FACTORY.get_or_init(|| name.into());
                }
            }
        }
        Ok(())
    }

    /// Called when an object is removed from the registry.
    fn on_remove(self: Rc<Self>) -> impl Fn(u32) {
        move |id| {
            if let Some(node) = self.nodes.borrow_mut().remove(&id) {
                info!("removing node id={id}, name={}", node.name);
            }
        }
    }

    /// Called when the registry detects that a node may be an Audiofox stream.
    fn init_stream_node(&self, id: u32, session: &str, streams: &StreamStates) -> LeveledResult<()> {
        // Parse its session attribute to see if it is a UUID
        let uuid = Uuid::from_str(session).with_context(|| {
            format!("node id={id}, name={AUDIOFOX}: NODE_SESSION is not a valid UUID")
        }).level(Debug)?;

        // Get the most recently added UUID and StreamState
        let mut streams = streams.borrow_mut();
        let (uuid, stream) = streams.last_mut().with_context(|| {
            format!("candidate id={id}, uuid={uuid}: no links active or initializing")
        }).level(Debug)?;

        // If this node's UUID matches, initialize a link on this node
        if uuid.eq(uuid) {
            if stream.id == RegId::default() {
                self.init.set(true);
                stream.id = id;
                info!("audiofox created successfully with id={id}");
            } else {
                error!("candidate id={id}, uuid={uuid}: uuid matches existing audiofox id={}", stream.id);
            }
        } else {
            debug!("candidate id={id}: unrecognized uuid");
        }

        Ok(())
    }

    /// Called when the registry detects that a port may be an Audiofox port.
    fn init_stream_port(&self, port: RegPort, streams: &StreamStates, sender: &mpsc::Sender<Response>) {
        if port.direction == Direction::In {
            // Get the most recently added UUID and StreamState
            let mut streams = streams.borrow_mut();
            if let Some((_uuid, stream)) = streams.last_mut() {
                // If this port is of the existing link, store its name
                if stream.id == port.node_id {
                    match port.channel {
                        Channel::FL => stream.in_left = port.name,
                        Channel::FR => stream.in_right = port.name,
                    }
                }
                // If we have gathered both input ports, the audiofox node is ready for linking
                if !(stream.in_left.is_empty() || stream.in_right.is_empty()) {
                    info!("audiofox node ready with 2 input ports");
                    send_logged!(sender, Response::Init);
                    self.init.set(false);
                }
            } else {
                send_logged!(sender, Response::Error(anyhow!("no links active or initializing")));
                self.init.set(false);
            }
        }
    }

    /// Called when the registry detects that a node needs to be reconnected to a stream.
    fn reconnect(&self, id: RegId, name: &RegName, streams: &StreamStates, listener: &CmdListener) -> LeveledResult<()> {
        // Disable the reconnecting flag
        self.recon.set(false);

        // Get the existing streams that target this name
        let uuids: Vec<_> = streams.borrow()
            .iter()
            .filter_map(|(uuid, state)| {
                state.names.contains(name).then_some(*uuid)
            }).collect();

        // Reinitialize a link with each target UUID
        for uuid in uuids {
            let node = &[AudioSource { id: id.into(), name: name.as_ref().into() }];
            listener.on_link(uuid, node).with_context(|| {
                format!("node id={id}, name={name}: failed to reconnect to stream uuid={uuid}")
            }).level(Warn)?;
        }

        Ok(())
    }
}