use crate::LOGGER;
use crate::backend::audio::ControllerMsg::Request;
use crate::backend::{BackendThread, Registrator};
use crossbeam_channel::{Receiver, Sender};
use eventfd::{EfdFlags, EventFD};
use jmb_logger_rs::*;
use mainloop::standard::Mainloop;
use pulse::callbacks::ListResult;
use pulse::context::subscribe::{Facility, InterestMaskSet};
use pulse::context::{Context, FlagSet, State};
use pulse::mainloop::api;
use pulse::mainloop::events::io;
use pulse::proplist::Proplist;
use pulse::volume::{ChannelVolumes, Volume};
use pulse::{mainloop, operation};
use std::boxed::Box;
use std::cell::Cell;
use std::collections::HashMap;
use std::os::fd::{AsRawFd, RawFd};
use std::rc::Rc;
use std::sync::{Arc, Mutex};
use std::thread;
use zbus::export::ordered_stream::OrderedStreamExt;
use zbus::fdo::{NameOwnerChanged, PropertiesProxy};
use zbus::message::Type;
use zbus::names::InterfaceName;
use zbus::zvariant::OwnedValue;
use zbus::{MatchRule, MessageStream};

pub struct PulseController {
    pub mainloop: Mainloop,
    context: Context,
    device_index: Option<u32>,
    device_bt: Option<bool>,
    current_vol: u8,
    current_mute: bool,
}

#[derive(Debug)]
pub enum ControllerMsg {
    NotifyServer,
    NotifySink(u32),
    Request,
    Set(u8, bool),
}

impl PulseController {
    pub fn new(tx: Sender<ControllerMsg>) -> Self {
        let mut mainloop = Mainloop::new().unwrap();
        let mut prop_list = Proplist::new().unwrap();

        prop_list
            .set_str(pulse::proplist::properties::APPLICATION_NAME, "jmb-bar")
            .unwrap();

        let mut context = Context::new_with_proplist(&mainloop, "jmb-bar", &prop_list).unwrap();

        context.connect(None, FlagSet::NOFLAGS, None).unwrap();

        loop {
            match context.get_state() {
                State::Ready => break,
                State::Failed | State::Terminated => panic!("PulseAudio connection failed"),
                _ => mainloop.iterate(false),
            };
        }

        let device_list = Arc::new(Mutex::new(Vec::new()));
        let clone = Arc::clone(&device_list);
        let op = context
            .introspect()
            .get_sink_info_list(move |info| match info {
                ListResult::Item(device) => {
                    if let Some(description) = device.description.as_ref()
                        && let Some(name) = device.name.as_ref()
                    {
                        if let Ok(mut lock) = clone.lock() {
                            lock.push((name.to_string(), description.to_string()));
                        } else {
                            error!(&LOGGER, "Failed to unlock device list on device {}", name);
                        }
                    } else {
                        error!(&LOGGER, "Failed to access device description");
                    }
                }
                ListResult::End => {}
                ListResult::Error => {}
            });

        while op.get_state() == operation::State::Running {
            mainloop.iterate(false);
        }

        context.set_subscribe_callback(Some(Box::new(move |facility, _, index| {
            if facility.is_some_and(|f| f == Facility::Sink) {
                let _ = tx.send(ControllerMsg::NotifySink(index));
            } else if facility.is_some_and(|f| f == Facility::Server) {
                let _ = tx.send(ControllerMsg::NotifyServer);
            }
        })));
        let _op = context.subscribe(InterestMaskSet::ALL, |success| {
            trace!(&LOGGER, "Subscribed to pulseaudio events: {success}");
        });

        let mut controller = PulseController {
            context,
            mainloop,
            device_index: None,
            device_bt: None,
            current_vol: 0,
            current_mute: true,
        };

        controller.query_default_device_blocking();
        controller
    }

    pub fn query_default_device_blocking(&mut self) {
        let default_device = Arc::new(Mutex::new(None));
        let dev_clone = default_device.clone();
        let op = self.context.introspect().get_server_info(move |info| {
            if let Some(name) = info.default_sink_name.as_ref() {
                let _ = dev_clone
                    .lock()
                    .expect("Couldn't get lock on default dev")
                    .insert(name.to_string());
            }
        });

        while op.get_state() == operation::State::Running {
            self.mainloop.iterate(false);
        }

        let name = default_device
            .lock()
            .expect("Couldn't get lock on device name")
            .take()
            .unwrap_or(String::new());

        let index = Arc::new(Mutex::new(None));
        let index_clone = index.clone();
        let op = self
            .context
            .introspect()
            .get_sink_info_by_name(&name, move |res| match res {
                ListResult::Item(item) => {
                    let _ = index_clone
                        .lock()
                        .expect("Couldn't get lock on device index")
                        .insert(item.index);
                }
                ListResult::End => {}
                ListResult::Error => {}
            });

        while op.get_state() == operation::State::Running {
            self.mainloop.iterate(false);
        }

        let val = index
            .lock()
            .expect("Couldn't get lock on device index")
            .take();

        self.device_index = val;
        self.device_bt = Some(name.starts_with("bluez"));
        debug!(
            &LOGGER,
            "Found default Audio Sink Idx: {:?} Bluetooth: {}",
            self.device_index,
            self.device_bt.unwrap()
        );
    }

    pub fn query_state_blocking(&mut self) -> bool {
        if let Some(idx) = self.device_index {
            let volume = Rc::new(Cell::new(self.current_vol));
            let vol_clone = volume.clone();
            let mute = Rc::new(Cell::new(self.current_mute));
            let mute_clone = mute.clone();
            let op =
                self.context
                    .introspect()
                    .get_sink_info_by_index(idx, move |info| match info {
                        ListResult::Item(info) => {
                            let percent =
                                info.volume.avg().0 as f64 * 100.0 / Volume::NORMAL.0 as f64;
                            trace!(
                                &LOGGER,
                                "State of default Sink changed: {percent} {}", info.mute
                            );
                            vol_clone.set(percent as u8);
                            mute_clone.set(info.mute);
                        }
                        ListResult::End => {}
                        ListResult::Error => {}
                    });

            while op.get_state() == operation::State::Running {
                self.mainloop.iterate(false);
            }

            let mut dirty = false;
            let volume = volume.take();
            if self.current_vol != volume {
                self.current_vol = volume;
                dirty = true;
            }
            let mute = mute.take();
            if self.current_mute != mute {
                self.current_mute = mute;
                dirty = true;
            }
            return dirty;
        }
        false
    }

    pub fn set_state(&mut self, volume: u8, mute: bool) {
        if let Some(idx) = self.device_index {
            if self.current_vol != volume {
                self.current_vol = volume;
                let volume = Volume(
                    ((volume.min(150) as f64 / 100.0) * Volume::NORMAL.0 as f64).round() as u32,
                );

                let mut channel_volumes = ChannelVolumes::default();
                channel_volumes.set(2, volume);

                self.context.introspect().set_sink_volume_by_index(
                    idx,
                    &channel_volumes,
                    Some(Box::new(|success| {
                        if !success {
                            warn!(&LOGGER, "Failed setting sink volume");
                        }
                    })),
                );
            }

            if self.current_mute != mute {
                self.current_mute = mute;
                let a = self.context.introspect().set_sink_mute_by_index(
                    idx,
                    mute,
                    Some(Box::new(|success| {
                        if !success {
                            warn!(&LOGGER, "Failed setting sink mute");
                        }
                    })),
                );
                while a.get_state() == operation::State::Running {
                    self.mainloop.iterate(false);
                }
            }
        }
    }
}

#[derive(Clone, Debug)]
pub enum AudioMessage {
    Init(Sender<AudioMessage>),
    Request,
    Cmd(u8, bool, bool),
}

pub struct AudioBackend {
    subscribers: Vec<Sender<AudioMessage>>,
    command_tx: Option<Sender<AudioMessage>>,
}

impl AudioBackend {
    pub fn new() -> Self {
        Self {
            subscribers: vec![],
            command_tx: None,
        }
    }
}

impl BackendThread<AudioMessage, Sender<AudioMessage>> for AudioBackend {
    fn subscribe(
        registrator: Registrator<AudioMessage>,
    ) -> (Receiver<AudioMessage>, Sender<AudioMessage>) {
        let (result_tx, result_rx) = crossbeam_channel::unbounded::<AudioMessage>();
        let _ = registrator.send(result_tx);
        while let Ok(state) = result_rx.recv() {
            return match state {
                AudioMessage::Init(sender) => (result_rx, sender),
                _ => continue,
            };
        }
        panic!("Channel closed unexpectedly, without returning anything")
    }

    fn subscribers(&self) -> &Vec<Sender<AudioMessage>> {
        &self.subscribers
    }

    fn init(mut self, registration_rx: Receiver<Sender<AudioMessage>>) {
        let (command_tx, command_rx) = crossbeam_channel::unbounded::<AudioMessage>();
        self.command_tx = Some(command_tx);
        let instance = Arc::new(Mutex::new(self));
        let (controller_tx, controller_rx) = crossbeam_channel::unbounded::<ControllerMsg>();
        let notify = Arc::new(EventFD::new(0, EfdFlags::EFD_CLOEXEC).unwrap());
        thread::spawn({
            let instance = instance.clone();
            let controller_tx = controller_tx.clone();
            let notify = notify.clone();
            move || {
                let mut controller = PulseController::new(controller_tx);
                let _event = api::Mainloop::new_io_event(
                    &mut controller.mainloop,
                    notify.as_raw_fd(),
                    io::FlagSet::INPUT,
                    Box::new(move |_, fd, _flags| match drain_eventfd(fd) {
                        Ok(_) => trace!(&LOGGER, "Woke pulseaudio loop using EventFD"),
                        Err(e) => warn!(&LOGGER, "Error draining EventFD {}", e.to_string()),
                    }),
                )
                .unwrap();
                loop {
                    while let Ok(msg) = controller_rx.try_recv() {
                        match msg {
                            Request => {
                                instance.lock().expect("Couldn't get lock").publish(
                                    AudioMessage::Cmd(
                                        controller.current_vol,
                                        controller.current_mute,
                                        controller.device_bt.unwrap_or(false),
                                    ),
                                );
                            }
                            ControllerMsg::NotifyServer => {
                                controller.query_default_device_blocking();
                                let _ = controller.query_state_blocking();
                                instance.lock().expect("Couldn't get lock").publish(
                                    AudioMessage::Cmd(
                                        controller.current_vol,
                                        controller.current_mute,
                                        controller.device_bt.unwrap_or(false),
                                    ),
                                );
                            }
                            ControllerMsg::NotifySink(idx) => {
                                if controller.device_index.is_some_and(|i| i == idx) {
                                    let changed = controller.query_state_blocking();
                                    if changed {
                                        instance.lock().expect("Couldn't get lock").publish(
                                            AudioMessage::Cmd(
                                                controller.current_vol,
                                                controller.current_mute,
                                                controller.device_bt.unwrap_or(false),
                                            ),
                                        );
                                        trace!(
                                            &LOGGER,
                                            "Change: {} muted: {}",
                                            controller.current_vol,
                                            controller.current_mute
                                        );
                                    }
                                }
                            }
                            ControllerMsg::Set(volume, mute) => {
                                controller.set_state(volume, mute);
                                instance.lock().expect("Couldn't get lock").publish(
                                    AudioMessage::Cmd(
                                        controller.current_vol,
                                        controller.current_mute,
                                        controller.device_bt.unwrap_or(false),
                                    ),
                                );
                            }
                        }
                    }
                    controller.mainloop.iterate(true);
                }
            }
        });
        thread::spawn({
            let instance = instance.clone();
            move || {
                while let Ok(sender) = registration_rx.recv() {
                    if sender
                        .send(AudioMessage::Init(
                            instance
                                .lock()
                                .expect("Couldn't get lock")
                                .command_tx
                                .clone()
                                .unwrap(),
                        ))
                        .is_ok()
                    {
                        let _ = sender.send(AudioMessage::Request);
                        instance
                            .lock()
                            .expect("Couldn't get subscribers")
                            .subscribers
                            .push(sender);
                    }
                }
            }
        });
        thread::spawn(move || {
            while let Ok(command) = command_rx.recv() {
                match command {
                    AudioMessage::Cmd(vol, mute, _) => {
                        let _ = controller_tx.send(ControllerMsg::Set(vol, mute));
                        notify.write(1).expect("Error writing notify");
                    }
                    AudioMessage::Request => {
                        let _ = controller_tx.send(Request);
                        notify.write(1).expect("Error writing notify");
                    }
                    _ => {
                        warn!(&LOGGER, "Frontend sent invalid Command")
                    }
                }
            }
        });
    }
}

const MPRIS_PREFIX: &str = "org.mpris.MediaPlayer2";
const MPRIS_PATH: &str = "/org/mpris/MediaPlayer2";
const MPRIS_PLAYER_INTERFACE: &str = "org.mpris.MediaPlayer2.Player";

#[derive(Clone, Debug)]
pub enum PlayerStatus {
    Playing(String),
    Paused(String),
    Stopped,
}

#[derive(Clone, Debug)]
pub enum MediaMessage {
    Init(Sender<MediaMessage>),
    PlayerStatus(PlayerStatus),
    PlayerPlay,
    PlayerPause,
    PlayerStop,
}

pub struct MediaBackend {
    subscribers: Vec<Sender<MediaMessage>>,
    players: HashMap<String, PlayerStatus>,
    active_player: Option<String>,
}

impl BackendThread<MediaMessage, Sender<MediaMessage>> for MediaBackend {
    fn subscribe(
        registrator: Registrator<MediaMessage>,
    ) -> (Receiver<MediaMessage>, Sender<MediaMessage>) {
        let (result_tx, result_rx) = crossbeam_channel::unbounded::<MediaMessage>();
        let _ = registrator.send(result_tx);
        while let Ok(state) = result_rx.recv() {
            return match state {
                MediaMessage::Init(sender) => (result_rx, sender),
                _ => continue,
            };
        }
        panic!("Channel closed unexpectedly, without returning anything")
    }

    fn subscribers(&self) -> &Vec<Sender<MediaMessage>> {
        &self.subscribers
    }

    fn init(self, registration_rx: Receiver<Sender<MediaMessage>>) {
        let (tx, rx) = crossbeam_channel::unbounded::<MediaMessage>();
        let instance = Arc::new(Mutex::new(self));
        thread::spawn({
            let instance = instance.clone();
            move || {
                while let Ok(sender) = registration_rx.recv() {
                    if let Ok(_) = sender.send(MediaMessage::Init(tx.clone())) {
                        let mut instance_guard = instance.lock().unwrap();
                        if let Some(name) = &instance_guard.active_player
                            && let Some(status) = instance_guard.players.get(name)
                        {
                            let _ = sender.send(MediaMessage::PlayerStatus(status.clone()));
                        }
                        instance_guard.subscribers.push(sender);
                    }
                }
            }
        });
        thread::spawn({
            let instance = instance.clone();
            move || {
                let runtime = tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .expect("Failed to create Tokio runtime");

                runtime.block_on(async move {
                    if let Err(e) = Self::wait_for_players(instance).await {
                        error!(&LOGGER, "Error waiting for players: {e}");
                    }
                });
            }
        });
        thread::spawn(move || {
            let runtime = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("Failed to create Tokio runtime");

            runtime.block_on({
                let instance = instance.clone();
                async move {
                    let Ok(connection) = zbus::Connection::session().await else {
                        error!(&LOGGER, "Couldn't create D-Bus connection");
                        return;
                    };
                    while let Ok(msg) = rx.recv() {
                        let mut instance_guard = instance.lock().unwrap();
                        if let Some(name) = &instance_guard.active_player {
                            if let Some(status) = instance_guard.players.get(name) {
                                match msg {
                                    MediaMessage::PlayerPlay => {
                                        if let PlayerStatus::Paused(_) = status {
                                            instance_guard.active_player_play(&connection).await;
                                        }
                                    }
                                    MediaMessage::PlayerPause => {
                                        if let PlayerStatus::Playing(_) = status {
                                            instance_guard.active_player_pause(&connection).await;
                                        }
                                    }
                                    MediaMessage::PlayerStop => {
                                        if let PlayerStatus::Playing(_) | PlayerStatus::Paused(_) =
                                            status
                                        {
                                            instance_guard.active_player_stop(&connection).await;
                                        }
                                    }
                                    _ => {}
                                }
                            }
                        }
                    }
                }
            });
        });
    }
}

impl MediaBackend {
    pub fn new() -> Self {
        Self {
            subscribers: vec![],
            players: HashMap::new(),
            active_player: None,
        }
    }
    async fn find_existing_players(
        connection: &zbus::Connection,
        instance: Arc<Mutex<Self>>,
    ) -> zbus::fdo::Result<()> {
        let dbus = zbus::Proxy::new(
            connection,
            "org.freedesktop.DBus",
            "/org/freedesktop/DBus",
            "org.freedesktop.DBus",
        )
        .await?;

        let names: Vec<String> = dbus.call("ListNames", &()).await?;
        let players: Vec<String> = names
            .into_iter()
            .filter(|name| name.starts_with(&format!("{MPRIS_PREFIX}.")))
            .collect();

        for player in players {
            Self::spawn_new_player_watcher(&connection, &instance, &player);
        }

        Ok(())
    }

    async fn wait_for_players(instance: Arc<Mutex<Self>>) -> zbus::fdo::Result<()> {
        let connection = zbus::Connection::session().await?;

        if let Err(e) = Self::find_existing_players(&connection, instance.clone()).await {
            warn!(&LOGGER, "Error finding existing Media Players: {e}");
        }

        let rule = MatchRule::builder()
            .msg_type(Type::Signal)
            .sender("org.freedesktop.DBus")?
            .interface("org.freedesktop.DBus")?
            .member("NameOwnerChanged")?
            .build();

        let mut stream = MessageStream::for_match_rule(rule, &connection, Some(32)).await?;

        while let Some(Ok(message)) = stream.next().await {
            let Some(signal) = NameOwnerChanged::from_message(message) else {
                continue;
            };

            let Ok(args) = signal.args() else {
                continue;
            };
            let name = args.name.as_str();

            if !name.starts_with(&format!("{MPRIS_PREFIX}.")) {
                continue;
            }

            let old_owner = args.old_owner();
            let new_owner = args.new_owner();

            let map = &mut instance.lock().unwrap().players;
            if old_owner.is_none() && new_owner.is_some() {
                Self::spawn_new_player_watcher(&connection, &instance, name);
            } else if old_owner.is_some() && new_owner.is_none() {
                map.remove(name);
                debug!(&LOGGER, "(1) Player removed: {name}");
            } else {
                debug!(&LOGGER, "Player owner changed: {name}");
            }
        }
        warn!(&LOGGER, "Player Finder ended prematurely");
        Ok(())
    }

    fn spawn_new_player_watcher(
        connection: &zbus::Connection,
        instance: &Arc<Mutex<Self>>,
        name: &str,
    ) {
        tokio::spawn({
            let instance = instance.clone();
            let connection = connection.clone();
            let name = name.to_string();
            async move {
                let _ = Self::watch_player_status(&connection, instance, &name).await;
            }
        });
    }

    async fn watch_player_status(
        connection: &zbus::Connection,
        instance: Arc<Mutex<Self>>,
        name: &str,
    ) -> zbus::fdo::Result<()> {
        let props = PropertiesProxy::builder(connection)
            .destination(name)?
            .path(MPRIS_PATH)?
            .build()
            .await?;

        let mut changes = props.receive_properties_changed().await?;

        let try_string: Result<String, _> = props
            .get(
                InterfaceName::from_static_str_unchecked(MPRIS_PLAYER_INTERFACE),
                "PlaybackStatus",
            )
            .await?
            .try_into();

        let initial: String = if let Ok(s) = try_string {
            s
        } else {
            String::from("Stopped")
        };

        debug!(&LOGGER, "Player added: {name} {initial}");
        let _ = Self::insert_or_update_player(&instance, &props, name.into(), initial).await;

        while let Some(change) = changes.next().await {
            let args = change.args()?;

            if !args.interface_name().eq(MPRIS_PLAYER_INTERFACE) {
                continue;
            }

            if let Some(value) = args.changed_properties().get("PlaybackStatus") {
                if let Ok(status) = <String>::try_from(value.clone()) {
                    debug!(&LOGGER, "Player ({name}) status has changed to: '{status}'");
                    Self::insert_or_update_player(&instance, &props, name.into(), status).await;
                }
            }
        }
        debug!(&LOGGER, "(2) Removing player: {name}");
        instance.lock().unwrap().players.remove(name);

        Ok(())
    }

    async fn insert_or_update_player(
        instance: &Arc<Mutex<Self>>,
        props: &PropertiesProxy<'_>,
        player_name: String,
        player_status: String,
    ) {
        match Self::get_player_status_from_string(&props, player_status).await {
            Ok(status) => {
                let mut instance_guard = instance.lock().unwrap();
                instance_guard.players.insert(player_name.clone(), status);
                instance_guard.select_active_player(Some(player_name));
            }
            Err(e) => {
                warn!(&LOGGER, "Error parsing player status: {e}");
            }
        }
    }

    async fn get_title(props: &PropertiesProxy<'_>) -> zbus::fdo::Result<String> {
        let metadata = props
            .get(
                InterfaceName::from_static_str_unchecked(MPRIS_PLAYER_INTERFACE),
                "Metadata",
            )
            .await?;
        let metadata = <HashMap<String, OwnedValue>>::try_from(metadata);
        match metadata {
            Ok(metadata) => {
                if let Some(title) = metadata.get("xesam:title") {
                    <String>::try_from(title.clone())
                        .map_err(|e| zbus::fdo::Error::Failed(e.to_string()))
                } else {
                    Err(zbus::fdo::Error::Failed("Couldn't parse title".into()))
                }
            }
            Err(err) => {
                warn!(&LOGGER, "Error parsing metadata {}", err);
                Err(zbus::fdo::Error::Failed("Couldn't parse metadata".into()))
            }
        }
    }

    async fn get_player_status_from_string(
        props: &PropertiesProxy<'_>,
        status: String,
    ) -> zbus::fdo::Result<PlayerStatus> {
        match status.as_str() {
            "Playing" => match Self::get_title(props).await {
                Ok(title) => Ok(PlayerStatus::Playing(title)),
                Err(e) => Err(e),
            },
            "Paused" => match Self::get_title(props).await {
                Ok(title) => Ok(PlayerStatus::Paused(title)),
                Err(e) => Err(e),
            },
            "Stopped" => Ok(PlayerStatus::Stopped),
            other => Err(zbus::fdo::Error::Failed(format!(
                "Invalid status string: '{other}' {:?}",
                other.as_bytes()
            ))),
        }
    }

    async fn active_player_play(&self, connection: &zbus::Connection) {
        if let Some(player_name) = &self.active_player {
            if let Ok(proxy) = zbus::Proxy::new(
                &connection,
                player_name.as_str(),
                MPRIS_PATH,
                MPRIS_PLAYER_INTERFACE,
            )
            .await
            {
                if let Err(e) = proxy.call_method("Play", &()).await {
                    warn!(
                        &LOGGER,
                        "Error starting play for player: {player_name}, {e}"
                    );
                }
            }
        }
    }

    async fn active_player_pause(&mut self, connection: &zbus::Connection) {
        if let Some(player_name) = &self.active_player {
            if let Ok(proxy) = zbus::Proxy::new(
                &connection,
                player_name.as_str(),
                MPRIS_PATH,
                MPRIS_PLAYER_INTERFACE,
            )
            .await
            {
                if let Err(e) = proxy.call_method("Pause", &()).await {
                    warn!(&LOGGER, "Error pausing player: {player_name}, {e}");
                }
            }
            self.select_active_player(self.active_player.clone());
        }
    }

    async fn active_player_stop(&mut self, connection: &zbus::Connection) {
        if let Some(player_name) = &self.active_player {
            if let Ok(proxy) = zbus::Proxy::new(
                &connection,
                player_name.as_str(),
                MPRIS_PATH,
                MPRIS_PLAYER_INTERFACE,
            )
            .await
            {
                if let Err(e) = proxy.call_method("Stop", &()).await {
                    warn!(&LOGGER, "Error stopping player: {player_name}, {e}");
                }
            }
        }
        self.select_active_player(None);
    }

    fn select_active_player(&mut self, changed_player: Option<String>) {
        let mut changed_also_playing = false;
        let mut new_active: Option<String> = None;
        let mut send_status = PlayerStatus::Stopped;
        if let Some(player) = changed_player
            && let Some(status) = self.players.get(&player)
        {
            match status {
                PlayerStatus::Playing(_) => {
                    new_active = Some(player);
                    send_status = status.clone();
                    changed_also_playing = true;
                }
                PlayerStatus::Paused(_) => {
                    new_active = Some(player);
                    send_status = status.clone();
                }
                PlayerStatus::Stopped => {} // Stopped players shouldn't be active
            }
        }

        if !changed_also_playing {
            for (name, status) in &self.players {
                match status {
                    PlayerStatus::Playing(_) => {
                        // Found a player that is playing media -> prioritize this one
                        new_active = Some(name.clone());
                        send_status = status.clone();
                        break;
                    }
                    PlayerStatus::Paused(_) => {
                        if new_active.is_none() {
                            new_active = Some(name.clone());
                            send_status = status.clone();
                        }
                    }
                    PlayerStatus::Stopped => {} // Stopped players shouldn't be active
                }
            }
        }

        self.active_player = new_active;
        self.publish(MediaMessage::PlayerStatus(send_status));
    }
}

fn drain_eventfd(fd: RawFd) -> std::io::Result<u64> {
    let mut value: u64 = 0;

    let result = unsafe { libc::read(fd, (&mut value as *mut u64).cast(), size_of::<u64>()) };

    if result == size_of::<u64>() as isize {
        Ok(value)
    } else if result == -1 {
        Err(std::io::Error::last_os_error())
    } else {
        Err(std::io::Error::new(
            std::io::ErrorKind::UnexpectedEof,
            "short eventfd read",
        ))
    }
}