audiofox 1.1.0

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

use std::cell::RefCell;
use std::collections::HashMap;
use std::ops::Deref;
use std::rc::Rc;
use std::sync::mpsc;
use uuid::Uuid;

use anyhow::Context;
use indexmap::IndexMap;
use log::info;
use pw::core::CoreRc;
use pw::keys;
use pw::main_loop::MainLoopRc;
use pw::properties::properties;

use crate::common::callback::AudioCallback;
use crate::common::result::{Error, Result};
use crate::common::source::{AudioSource, AudioSources};
use crate::platform::linux::manager::Command;
use crate::platform::linux::registry::LINK_FACTORY;
use crate::platform::linux::registry::RegState;
use crate::platform::linux::stream::StreamState;

/// Valid responses from the PipeWire loop.
#[derive(Debug)]
pub enum Response {
    Ready,
    Error(Error),
    List(AudioSources),
    Init,
    Link,
}

/// Ordered map of [`StreamState`] by their UUIDs.
#[derive(Default)]
pub struct StreamStates(RefCell<IndexMap<Uuid, StreamState>>);

impl Deref for StreamStates {
    type Target = RefCell<IndexMap<Uuid, StreamState>>;
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

#[derive(Clone)]
pub struct CmdListener {
    pub core: CoreRc,
    streams: Rc<StreamStates>,
    sender: mpsc::Sender<Response>,
    mainloop: MainLoopRc,
    registry: Rc<RegState>,
}

impl CmdListener {
    pub fn new(
        mainloop: &MainLoopRc,
        core: &CoreRc,
        registry: &Rc<RegState>,
        sender: &mpsc::Sender<Response>,
        streams: &Rc<StreamStates>,
    ) -> Self {
        Self {
            mainloop: mainloop.clone(),
            core: core.clone(),
            registry: registry.clone(),
            sender: sender.clone(),
            streams: streams.clone(),
        }
    }

    /// Bridge between the `PlatformManager` and `Command`-specific logic.
    pub fn recv(self: Rc<Self>) -> impl Fn(Command) {
        move |cmd| {
            let response = match cmd {
                // Infallible, unresponsive
                Command::Stop => {
                    self.mainloop.quit();
                    return;
                }
                // Infallible, responsive
                Command::List => Response::List(
                    self.registry.nodes.borrow()
                        .iter()
                        .fold(HashMap::new(), |mut acc, (reg_id, reg_node)| {
                            acc.entry(reg_node.name.as_ref())
                                .or_insert_with(Vec::new)
                                .push(*reg_id);
                            acc
                        })
                        .into_iter()
                        .map(|(name, ids)| AudioSource {
                            id: ids.into(),
                            name: name.into(),
                        })
                        .collect()
                ),
                // Fallible, responsive
                Command::Init(uuid, callback, frames) => {
                    match self.on_init(uuid, callback, frames).map_err(Response::Error) {
                        Ok(()) => return,
                        Err(response) => response,
                    }
                }
                // Fallible, responsive
                Command::Link(uuid, targets) => {
                    match self.on_link(uuid, targets.as_ref()).map_err(Response::Error) {
                        Ok(()) => Response::Link,
                        Err(e) => e,
                    }
                }
                // Infallible, unresponsive
                Command::End(uuid) => {
                    self.on_end(uuid);
                    return;
                }
            };
            // Send the response
            send_logged!(self.sender, response);
        }
    }

    /// Called when a stream is initialized by the `PlatformManager`.
    fn on_init(&self, uuid: Uuid, callback: AudioCallback, frames: usize) -> Result<()> {
        // Create a new stream state
        let mut streams = self.streams.borrow_mut();
        streams.insert(uuid, StreamState::default());

        // Initialize the stream
        let stream = &streams[&uuid];
        stream.start(self, uuid, callback, frames).context("Failed to start stream")?;

        // The registry will now supply the id and ports
        Ok(())
    }

    /// Called when a stream is ended by the `PlatformManager`, or when a stream detects that its
    /// node has been destroyed.
    pub fn on_end(&self, uuid: Uuid) {
        if let Ok(mut streams) = self.streams.try_borrow_mut() {
            if streams.swap_remove(&uuid).is_some() {
                info!("stream successfully ended.");
            }
        }
    }

    /// Called when the registry populates the ID and ports of the stream.
    pub fn on_link(&self, uuid: Uuid, targets: &[AudioSource]) -> Result<()> {
        // Get the name of the link factory
        let link_factory = LINK_FACTORY.get().context("No link factory found")?;

        // Get the stream state
        let mut streams = self.streams.borrow_mut();
        let stream = streams.get_mut(&uuid)
            .with_context(|| format!("State not found for UUID: {uuid}"))?;

        // Process each target node
        let registry = self.registry.nodes.borrow();

        for target in targets {
            for id in target.id.iter().filter(|&&id| id != 0) {
                // Look up the target in the registry hash map
                let reg_node = registry.get(id)
                    .with_context(|| format!("Invalid node ID: {id}"))?;

                // Map the ports
                let ports = [
                    (&stream.in_left, &reg_node.out_left),
                    (&stream.in_right, &reg_node.out_right),
                ];

                // Link each port pair
                for (stream_input, target_output) in ports {
                    info!("linking: {}:{} -> {}:{}",id, target_output, stream.id, stream_input);

                    let link = self.core.create_object(
                        link_factory, &properties! {
                            *keys::LINK_OUTPUT_NODE => id.to_string(),
                            *keys::LINK_OUTPUT_PORT => target_output.as_ref(),
                            *keys::LINK_INPUT_NODE => stream.id.to_string(),
                            *keys::LINK_INPUT_PORT => stream_input.as_ref(),
                        },
                    ).context("Failed to create link")?;
                    stream.links.push(link);
                    stream.ids.push(*id);
                    stream.names.push(target.name.as_ref().into());
                }
            }
        }
        Ok(())
    }
}