Kordophone/kpcli/src/daemon/xpc.rs

use super::{ConfigCommands, DaemonInterface};
use anyhow::Result;
use async_trait::async_trait;
use futures_util::StreamExt;
use std::collections::HashMap;
use std::ffi::{CStr, CString};
use std::ops::Deref;
use std::{pin::Pin, task::Poll};

use xpc_connection::Message;

use futures::{
    channel::mpsc::{unbounded as unbounded_channel, UnboundedReceiver, UnboundedSender},
    Stream,
};


const SERVICE_NAME: &str = "net.buzzert.kordophonecd\0";
const GET_VERSION_METHOD: &str = "GetVersion";

// We can't use XPCClient from xpc-connection because of some strange decisions with which flags
// are passed to xpc_connection_create_mach_service.
struct XPCClient {
    connection: xpc_connection_sys::xpc_connection_t,
    receiver: UnboundedReceiver<Message>,
    sender: UnboundedSender<Message>,
    event_handler_is_running: bool,
}

impl XPCClient {
    pub fn connect(name: impl AsRef<CStr>) -> Self {
        use block::ConcreteBlock;
        use xpc_connection::xpc_object_to_message;
        use xpc_connection_sys::xpc_connection_set_event_handler;
        use xpc_connection_sys::xpc_connection_resume;

        let name = name.as_ref();
        let connection = unsafe {
            xpc_connection_sys::xpc_connection_create_mach_service(name.as_ptr(), std::ptr::null_mut(), 0)
        };

        let (sender, receiver) = unbounded_channel();
        let sender_clone = sender.clone();

        let block = ConcreteBlock::new(move |event| {
            let message = xpc_object_to_message(event);
            sender_clone.unbounded_send(message).ok()
        });

        let block = block.copy();

        unsafe {
            xpc_connection_set_event_handler(connection, block.deref() as *const _ as *mut _);
            xpc_connection_resume(connection);
        }

        Self {
            connection,
            receiver,
            sender,
            event_handler_is_running: true,
        }
    }

    pub fn send_message(&self, message: Message) {
        use xpc_connection::message_to_xpc_object;
        use xpc_connection_sys::xpc_connection_send_message;
        use xpc_connection_sys::xpc_release;

        let xpc_object = message_to_xpc_object(message);
        unsafe {
            xpc_connection_send_message(self.connection, xpc_object);
            xpc_release(xpc_object);
        }
    }
}

impl Drop for XPCClient {
    fn drop(&mut self) {
        use xpc_connection_sys::xpc_release;
        use xpc_connection_sys::xpc_object_t;

        unsafe { xpc_release(self.connection as xpc_object_t) };
    }
}

impl Stream for XPCClient {
    type Item = Message;

    fn poll_next(
        mut self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        match Stream::poll_next(Pin::new(&mut self.receiver), cx) {
            Poll::Ready(Some(Message::Error(xpc_connection::MessageError::ConnectionInvalid))) => {
                self.event_handler_is_running = false;
                Poll::Ready(None)
            }
            v => v,
        }
    }
}

unsafe impl Send for XPCClient {}

/// XPC-based implementation of DaemonInterface that sends method calls to the daemon over libxpc.
pub struct XpcDaemonInterface;

impl XpcDaemonInterface {
    /// Create a new XpcDaemonInterface. No state is held.
    pub fn new() -> Result<Self> {
        Ok(Self)
    }

    fn build_service_name() -> Result<CString> {
        let service_name = SERVICE_NAME.trim_end_matches('\0');
        Ok(CString::new(service_name)?)
    }

    fn build_request(method: &str, args: Option<HashMap<CString, Message>>) -> HashMap<CString, Message> {
        let mut request = HashMap::new();
        request.insert(CString::new("method").unwrap(), Message::String(CString::new(method).unwrap()));
        if let Some(arguments) = args {
            request.insert(CString::new("arguments").unwrap(), Message::Dictionary(arguments));
        }
        request
    }

    async fn call_method(&self, client: &mut XPCClient, method: &str, args: Option<HashMap<CString, Message>>) -> anyhow::Result<HashMap<CString, Message>> {
        let request = Self::build_request(method, args);
        client.send_message(Message::Dictionary(request));

        match client.next().await {
            Some(Message::Dictionary(map)) => Ok(map),
            Some(other) => Err(anyhow::anyhow!("Unexpected XPC reply: {:?}", other)),
            None => Err(anyhow::anyhow!("No reply received from XPC daemon")),
        }
    }

    fn get_string<'a>(map: &'a HashMap<CString, Message>, key: &str) -> Option<&'a CStr> {
        map.get(&CString::new(key).ok()?).and_then(|v| match v { Message::String(s) => Some(s.as_c_str()), _ => None })
    }
}

#[async_trait(?Send)]
impl DaemonInterface for XpcDaemonInterface {
    async fn print_version(&mut self) -> Result<()> {
        // Build service name and connect
        let mach_port_name = Self::build_service_name()?;
        let mut client = XPCClient::connect(&mach_port_name);

        // Call generic method and parse reply
        let map = self.call_method(&mut client, GET_VERSION_METHOD, None).await?;
        if let Some(ver) = Self::get_string(&map, "version") {
            println!("Server version: {}", ver.to_string_lossy());
            Ok(())
        } else if let Some(ty) = Self::get_string(&map, "type") {
            println!("XPC replied with type: {}", ty.to_string_lossy());
            Ok(())
        } else {
            Err(anyhow::anyhow!("Unexpected XPC reply payload for GetVersion"))
        }
    }

    // Remaining methods unimplemented on macOS
    async fn print_conversations(&mut self) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
    async fn sync_conversations(&mut self, _conversation_id: Option<String>) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
    async fn sync_conversations_list(&mut self) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
    async fn print_messages(
        &mut self,
        _conversation_id: String,
        _last_message_id: Option<String>,
    ) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
    async fn enqueue_outgoing_message(
        &mut self,
        _conversation_id: String,
        _text: String,
    ) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
    async fn wait_for_signals(&mut self) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
    async fn config(&mut self, _cmd: ConfigCommands) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
    async fn delete_all_conversations(&mut self) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
    async fn download_attachment(&mut self, _attachment_id: String) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
    async fn upload_attachment(&mut self, _path: String) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
    async fn mark_conversation_as_read(&mut self, _conversation_id: String) -> Result<()> {
        Err(anyhow::anyhow!("Feature not implemented for XPC"))
    }
}