koan-core 0.66.2

//! Finding renderers: SSDP search, then the network's own announcements.
//!
//! One `M-SEARCH` when something first asks, and again whenever a picker
//! opens. After that the list follows `ssdp:alive` and `ssdp:byebye` as
//! renderers send them, and an entry whose `max-age` runs out without being
//! renewed is left out of the list when it is next read. Nothing here runs on
//! a timer, but for the window after a search on a platform that keeps
//! multicast out unless asked (`Network`). Answers to a search are unicast and
//! arrive regardless; there, announcements are heard only in that window, and
//! the next search catches what was missed.

use std::collections::{HashMap, HashSet};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4, UdpSocket};
use std::sync::OnceLock;
use std::thread;
use std::time::{Duration, Instant};

use parking_lot::Mutex;
use url::Url;

use super::description::{self, Renderer};

const GROUP: Ipv4Addr = Ipv4Addr::new(239, 255, 255, 250);
const PORT: u16 = 1900;
const SEARCH_TARGET: &str = "urn:schemas-upnp-org:device:MediaRenderer:1";

/// What an announcement without a `max-age` is assumed to last.
const DEFAULT_MAX_AGE: Duration = Duration::from_secs(1800);

struct Known {
    renderer: Renderer,
    expires: Instant,
    /// Playing or paused when last asked, whoever is driving it.
    busy: bool,
}

#[derive(Default)]
struct Store {
    renderers: HashMap<String, Known>,
    describing: HashSet<String>,
}

static STORE: Mutex<Option<Store>> = Mutex::new(None);
static SEARCH: OnceLock<Option<UdpSocket>> = OnceLock::new();
/// When `search` last ran.
static SEARCHED: Mutex<Option<Instant>> = Mutex::new(None);

/// How long after a search multicast is let in where the platform keeps it
/// out: the `MX` renderers spread their answers over, and as long again, so
/// a renderer starting up alongside them is heard announcing itself.
const SEARCH_WINDOW: Duration = Duration::from_secs(5);

/// Where the platform decides what reaches the sockets. Android drops
/// multicast unless the app holds a lock for it, and a phone with mobile data
/// may send a search out of the wrong interface and join the group on it.
pub trait Network: Send + Sync {
    /// The Wi-Fi interface's IPv4 address, to send searches and join the group
    /// on; `None` leaves it to the routing table.
    fn wifi_address(&self) -> Option<Ipv4Addr>;
    /// Let multicast announcements in, or stop letting them in.
    fn hold_multicast(&self, on: bool);
}

static NETWORK: Mutex<Option<std::sync::Arc<dyn Network>>> = Mutex::new(None);
/// Moves with each search, so only the latest one's window lets go.
static WINDOW: Mutex<u64> = Mutex::new(0);
/// The announcement listener, and the interface it joined the group on, if
/// it could.
static NOTIFY: Mutex<Option<(UdpSocket, Option<Ipv4Addr>)>> = Mutex::new(None);

/// Install the platform's network host.
pub fn set_network(network: std::sync::Arc<dyn Network>) {
    *NETWORK.lock() = Some(network);
}

/// Let multicast in for one search's window, from now.
fn open_window(network: &std::sync::Arc<dyn Network>) {
    open_window_for(network, SEARCH_WINDOW);
}

fn open_window_for(network: &std::sync::Arc<dyn Network>, length: Duration) {
    let window = {
        let mut latest = WINDOW.lock();
        *latest += 1;
        network.hold_multicast(true);
        *latest
    };
    let network = network.clone();
    let closing = thread::Builder::new()
        .name("koan-ssdp-window".into())
        .spawn(move || {
            thread::sleep(length);
            log::info!(
                "upnp: {} renderers known {}s after a search",
                renderers().len(),
                length.as_secs()
            );
            let latest = WINDOW.lock();
            if *latest == window {
                network.hold_multicast(false);
            }
        });
    if closing.is_err() {
        // Held until the next search's window closes, rather than dropped
        // before this one's answers arrive.
        log::warn!("upnp: could not time the multicast window");
    }
}

fn with<R>(f: impl FnOnce(&mut Store) -> R) -> R {
    f(STORE.lock().get_or_insert_with(Store::default))
}

/// The renderers on the network, by name.
pub fn renderers() -> Vec<Renderer> {
    let now = Instant::now();
    let mut list: Vec<Renderer> = with(|s| {
        s.renderers
            .values()
            .filter(|k| k.expires > now)
            .map(|k| k.renderer.clone())
            .collect()
    });
    list.sort_by_key(|r| r.name.to_lowercase());
    list
}

/// Whether the renderer was playing or paused when last asked: picking one
/// that is takes it over from whatever is driving it.
pub fn busy(udn: &str) -> bool {
    with(|s| s.renderers.get(udn).is_some_and(|k| k.busy))
}

/// Whether `renderer` is playing or paused for something, asked now. `None`
/// when it does not answer.
pub fn in_use(renderer: &Renderer) -> Option<bool> {
    let args = super::soap::call(
        &super::soap::client(),
        &renderer.av_transport,
        "GetTransportInfo",
        &[("InstanceID", "0")],
    )
    .ok()?;
    Some(matches!(
        super::soap::arg(&args, "CurrentTransportState"),
        Some("PLAYING" | "PAUSED_PLAYBACK" | "TRANSITIONING")
    ))
}

/// The URI `renderer` is on, asked now. `None` when it does not answer.
pub fn playing_uri(renderer: &Renderer) -> Option<String> {
    let args = super::soap::call(
        &super::soap::client(),
        &renderer.av_transport,
        "GetPositionInfo",
        &[("InstanceID", "0")],
    )
    .ok()?;
    super::soap::arg(&args, "TrackURI").map(str::to_string)
}

/// Ask every known renderer whether it is in use, each on its own thread.
fn check_busy() {
    for renderer in renderers() {
        let _ = thread::Builder::new()
            .name("koan-upnp-status".into())
            .spawn(move || {
                let Some(busy) = in_use(&renderer) else {
                    return;
                };
                let changed = with(|s| {
                    s.renderers
                        .get_mut(&renderer.udn)
                        .is_some_and(|k| std::mem::replace(&mut k.busy, busy) != busy)
                });
                if changed {
                    crate::signal::engine_changed().bump();
                }
            });
    }
}

pub fn find(udn: &str) -> Option<Renderer> {
    with(|s| s.renderers.get(udn).map(|k| k.renderer.clone()))
}

/// Add or refresh a renderer, as if it had announced itself.
pub fn remember(renderer: Renderer, max_age: Duration) {
    let udn = renderer.udn.clone();
    let changed = with(|s| {
        let expires = Instant::now() + max_age;
        let changed = s.renderers.get(&udn).is_none_or(|k| k.renderer != renderer);
        let busy = s.renderers.get(&udn).is_some_and(|k| k.busy);
        s.renderers.insert(
            udn,
            Known {
                renderer,
                expires,
                busy,
            },
        );
        changed
    });
    if changed {
        crate::signal::engine_changed().bump();
    }
}

type GoneHook = (u64, String, Box<dyn Fn() + Send>);

static GONE: Mutex<Vec<GoneHook>> = Mutex::new(Vec::new());
static GONE_NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// Run `f` when the renderer `udn` says goodbye. Returns a handle for
/// `forget_gone`.
pub(crate) fn on_gone(udn: &str, f: impl Fn() + Send + 'static) -> u64 {
    let id = GONE_NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    GONE.lock().push((id, udn.to_string(), Box::new(f)));
    id
}

pub(crate) fn forget_gone(id: u64) {
    GONE.lock().retain(|(i, ..)| *i != id);
}

pub(crate) fn forget(udn: &str) {
    if with(|s| s.renderers.remove(udn).is_some()) {
        log::info!("upnp: {udn} left");
        crate::signal::engine_changed().bump();
    }
    for (_, hooked, f) in GONE.lock().iter() {
        if hooked == udn {
            f();
        }
    }
}

/// Search the network for renderers, starting discovery if it has not
/// started, and ask the ones already known whether they are in use. Answers
/// arrive over the next couple of seconds. Nothing, with `playback.renderers`
/// off.
pub fn search() {
    if !crate::config::Config::cached().playback.renderers {
        return;
    }
    let Some(socket) = SEARCH.get_or_init(start) else {
        return;
    };
    *SEARCHED.lock() = Some(Instant::now());
    let network = NETWORK.lock().clone();
    let mut interface = Ipv4Addr::UNSPECIFIED;
    if let Some(network) = &network {
        open_window(network);
        interface = network.wifi_address().unwrap_or(Ipv4Addr::UNSPECIFIED);
        if let Err(e) = socket2::SockRef::from(socket).set_multicast_if_v4(&interface) {
            log::warn!("upnp: cannot search from {interface}: {e}");
        }
    }
    rejoin(interface);
    log::info!(
        "upnp: searching from {interface}, {} renderers known",
        renderers().len()
    );
    check_busy();
    let msg = format!(
        "M-SEARCH * HTTP/1.1\r\nHOST: {GROUP}:{PORT}\r\nMAN: \"ssdp:discover\"\r\nMX: 2\r\nST: {SEARCH_TARGET}\r\nUSER-AGENT: koan/{} UPnP/1.0\r\n\r\n",
        env!("CARGO_PKG_VERSION")
    );
    // UDP may drop one; renderers ignore a repeat they have already answered.
    for _ in 0..2 {
        if let Err(e) = socket.send_to(msg.as_bytes(), (GROUP, PORT)) {
            log::warn!("upnp: search failed: {e}");
            return;
        }
    }
}

/// `search`, unless what is known is still good: renderers announce
/// themselves and their departures while the listener runs, so the cache
/// stands until one of its entries has outlived its `max-age`, or nothing has
/// been searched for in `DEFAULT_MAX_AGE`.
pub fn search_if_stale() {
    let now = Instant::now();
    let fresh = SEARCHED
        .lock()
        .is_some_and(|t| now.duration_since(t) < DEFAULT_MAX_AGE)
        && with(|s| s.renderers.values().all(|k| k.expires > now));
    if !fresh {
        search();
    }
}

/// Open the search socket, and a listener for announcements. The search
/// socket alone is enough to find renderers, so failing to share port 1900
/// with whatever else holds it costs only the live updates.
fn start() -> Option<UdpSocket> {
    let search = match UdpSocket::bind(("0.0.0.0", 0)) {
        Ok(s) => s,
        Err(e) => {
            log::warn!("upnp: cannot open a socket for discovery: {e}");
            return None;
        }
    };
    let _ = search.set_multicast_ttl_v4(2);
    if let Ok(reader) = search.try_clone() {
        spawn_reader("koan-ssdp-search", reader);
    }
    match notify_socket() {
        Ok(socket) => {
            if let Ok(kept) = socket.try_clone() {
                *NOTIFY.lock() = Some((kept, None));
            }
            spawn_reader("koan-ssdp-notify", socket);
            let interface = NETWORK
                .lock()
                .as_ref()
                .and_then(|n| n.wifi_address())
                .unwrap_or(Ipv4Addr::UNSPECIFIED);
            rejoin(interface);
        }
        Err(e) => log::info!("upnp: not listening for announcements: {e}"),
    }
    Some(search)
}

/// Join the announcement listener to the group on `interface`, moving it when
/// Wi-Fi has come, gone or changed address since, or when no join has worked
/// yet: with no network up, none can.
fn rejoin(interface: Ipv4Addr) {
    let mut notify = NOTIFY.lock();
    let Some((socket, joined)) = notify.as_mut() else {
        return;
    };
    if *joined == Some(interface) {
        return;
    }
    if let Some(old) = joined.take() {
        let _ = socket.leave_multicast_v4(&GROUP, &old);
    }
    match socket.join_multicast_v4(&GROUP, &interface) {
        Ok(()) => *joined = Some(interface),
        Err(e) => log::info!("upnp: cannot listen for announcements on {interface}: {e}"),
    }
}

fn notify_socket() -> std::io::Result<UdpSocket> {
    use socket2::{Domain, Protocol, Socket, Type};
    let socket = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP))?;
    socket.set_reuse_address(true)?;
    #[cfg(unix)]
    socket.set_reuse_port(true)?;
    socket.bind(&SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, PORT)).into())?;
    Ok(socket.into())
}

fn spawn_reader(name: &str, socket: UdpSocket) {
    let spawned = thread::Builder::new().name(name.into()).spawn(move || {
        let mut buf = [0u8; 4096];
        loop {
            match socket.recv_from(&mut buf) {
                Ok((n, _)) => {
                    if let Some(msg) = Message::parse(&String::from_utf8_lossy(&buf[..n])) {
                        heard(msg);
                    }
                }
                Err(e) => {
                    log::warn!("upnp: discovery socket closed: {e}");
                    return;
                }
            }
        }
    });
    if let Err(e) = spawned {
        log::warn!("upnp: could not start {name}: {e}");
    }
}

#[derive(Debug, PartialEq)]
pub(crate) struct Message {
    udn: String,
    kind: Kind,
}

#[derive(Debug, PartialEq)]
enum Kind {
    Alive { location: Url, max_age: Duration },
    ByeBye,
}

impl Message {
    /// A search response or announcement about a renderer. Anything else on
    /// the multicast group, and there is plenty, is `None`.
    pub(crate) fn parse(text: &str) -> Option<Self> {
        let mut lines = text.split("\r\n");
        let first = lines.next()?;
        let headers: Vec<(String, &str)> = lines
            .filter_map(|l| l.split_once(':'))
            .map(|(k, v)| (k.trim().to_ascii_uppercase(), v.trim()))
            .collect();
        let header = |name: &str| headers.iter().find(|(k, _)| k == name).map(|(_, v)| *v);
        let udn = header("USN")?.split("::").next()?.to_string();
        if !udn.starts_with("uuid:") {
            return None;
        }

        if first.starts_with("NOTIFY") {
            if header("NTS") == Some("ssdp:byebye") {
                return Some(Self {
                    udn,
                    kind: Kind::ByeBye,
                });
            }
            if !is_renderer(header("NT")?) {
                return None;
            }
        } else if first.starts_with("HTTP/") {
            if !is_renderer(header("ST")?) {
                return None;
            }
        } else {
            return None;
        }

        let max_age = header("CACHE-CONTROL")
            .and_then(|v| {
                v.split(',')
                    .filter_map(|d| d.trim().split_once('='))
                    .find(|(k, _)| k.trim().eq_ignore_ascii_case("max-age"))
                    .and_then(|(_, v)| v.trim().parse().ok())
            })
            .map(Duration::from_secs)
            .unwrap_or(DEFAULT_MAX_AGE);
        Some(Self {
            udn,
            kind: Kind::Alive {
                location: Url::parse(header("LOCATION")?).ok()?,
                max_age,
            },
        })
    }
}

fn is_renderer(target: &str) -> bool {
    target.contains(":device:MediaRenderer:") || target.contains(":service:AVTransport:")
}

fn heard(msg: Message) {
    match msg.kind {
        Kind::ByeBye => forget(&msg.udn),
        Kind::Alive { location, max_age } => {
            // Known at this address: the announcement only renews it.
            let fresh = with(|s| {
                if let Some(known) = s.renderers.get_mut(&msg.udn)
                    && known.renderer.location == location
                {
                    known.expires = Instant::now() + max_age;
                    return false;
                }
                s.describing.insert(msg.udn.clone())
            });
            if fresh {
                describe(msg.udn, location, max_age);
            }
        }
    }
}

fn describe(udn: String, location: Url, max_age: Duration) {
    let spawned = thread::Builder::new()
        .name("koan-upnp-describe".into())
        .spawn({
            let udn = udn.clone();
            move || {
                match fetch(&location) {
                    Ok(Some(renderer)) => {
                        log::info!(
                            "upnp: found {} ({} {}) at {location}{}",
                            renderer.name,
                            renderer.manufacturer,
                            renderer.model,
                            if renderer.gapless { ", gapless" } else { "" }
                        );
                        remember(renderer, max_age);
                    }
                    Ok(None) => log::debug!("upnp: {location} is not a renderer"),
                    Err(e) => log::info!("upnp: cannot describe {location}: {e}"),
                }
                with(|s| s.describing.remove(&udn));
            }
        });
    if spawned.is_err() {
        with(|s| s.describing.remove(&udn));
    }
}

/// Fetch and read a renderer's description, and its AVTransport SCPD for
/// whether it takes a next track.
pub fn fetch(location: &Url) -> Result<Option<Renderer>, String> {
    let http = super::soap::client();
    let get = |url: &Url| -> Result<String, String> {
        http.get(url.clone())
            .send()
            .and_then(|r| r.error_for_status())
            .and_then(|r| r.text())
            .map_err(|e| e.to_string())
    };
    let Some(mut renderer) = description::parse_device(&get(location)?, location)? else {
        return Ok(None);
    };
    renderer.gapless = get(&renderer.av_transport.scpd)
        .and_then(|doc| description::parse_actions(&doc))
        .map(|actions| actions.iter().any(|a| a == "SetNextAVTransportURI"))
        .unwrap_or(false);
    Ok(Some(renderer))
}

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

    struct Counting {
        holds: Mutex<Vec<bool>>,
    }

    impl Network for Counting {
        fn wifi_address(&self) -> Option<Ipv4Addr> {
            None
        }
        fn hold_multicast(&self, on: bool) {
            self.holds.lock().push(on);
        }
    }

    #[test]
    fn multicast_is_let_in_for_the_latest_searchs_window_only() {
        let counting = std::sync::Arc::new(Counting {
            holds: Mutex::new(Vec::new()),
        });
        let network: std::sync::Arc<dyn Network> = counting.clone();
        open_window_for(&network, Duration::from_millis(100));
        // A second search before the first window closes keeps it open.
        open_window_for(&network, Duration::from_millis(1000));
        thread::sleep(Duration::from_millis(500));
        assert_eq!(
            *counting.holds.lock(),
            [true, true],
            "the first window's end is not the second's"
        );
        thread::sleep(Duration::from_millis(1500));
        assert_eq!(*counting.holds.lock(), [true, true, false]);
    }

    #[test]
    fn a_search_response_names_the_renderer_and_its_description() {
        let msg = Message::parse(
            "HTTP/1.1 200 OK\r\nCACHE-CONTROL: max-age=100\r\nEXT:\r\nLOCATION: http://192.168.1.20:1597/\r\nSERVER: UPnP/1.0 Kodi\r\nST: urn:schemas-upnp-org:device:MediaRenderer:1\r\nUSN: uuid:bc9d::urn:schemas-upnp-org:device:MediaRenderer:1\r\n\r\n",
        )
        .unwrap();
        assert_eq!(msg.udn, "uuid:bc9d");
        assert_eq!(
            msg.kind,
            Kind::Alive {
                location: Url::parse("http://192.168.1.20:1597/").unwrap(),
                max_age: Duration::from_secs(100),
            }
        );
    }

    #[test]
    fn announcements_are_read_and_others_ignored() {
        let alive = "NOTIFY * HTTP/1.1\r\nHost: 239.255.255.250:1900\r\nNT: urn:schemas-upnp-org:service:AVTransport:1\r\nNTS: ssdp:alive\r\nLocation: http://10.0.0.5/d.xml\r\nUSN: uuid:amp::urn:schemas-upnp-org:service:AVTransport:1\r\n\r\n";
        assert!(matches!(
            Message::parse(alive).unwrap().kind,
            Kind::Alive { max_age, .. } if max_age == DEFAULT_MAX_AGE
        ));

        let bye = "NOTIFY * HTTP/1.1\r\nNT: upnp:rootdevice\r\nNTS: ssdp:byebye\r\nUSN: uuid:amp::upnp:rootdevice\r\n\r\n";
        assert_eq!(Message::parse(bye).unwrap().kind, Kind::ByeBye);

        let server = "NOTIFY * HTTP/1.1\r\nNT: urn:schemas-upnp-org:device:MediaServer:1\r\nNTS: ssdp:alive\r\nLocation: http://10.0.0.6/\r\nUSN: uuid:nas::urn:schemas-upnp-org:device:MediaServer:1\r\n\r\n";
        assert!(Message::parse(server).is_none());

        let search = "M-SEARCH * HTTP/1.1\r\nST: ssdp:all\r\n\r\n";
        assert!(Message::parse(search).is_none());
    }
}