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"; const GET_CONVERSATIONS_METHOD: &str = "GetConversations"; // 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, sender: UnboundedSender, event_handler_is_running: bool, } impl XPCClient { pub fn connect(name: impl AsRef) -> 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> { 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 { Ok(Self) } fn build_service_name() -> Result { let service_name = SERVICE_NAME.trim_end_matches('\0'); Ok(CString::new(service_name)?) } fn build_request(method: &str, args: Option>) -> HashMap { 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>) -> anyhow::Result> { 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 key(k: &str) -> CString { CString::new(k).unwrap() } fn get_string<'a>(map: &'a HashMap, key: &str) -> Option<&'a CStr> { map.get(&Self::key(key)).and_then(|v| match v { Message::String(s) => Some(s.as_c_str()), _ => None }) } fn get_i64_from_str(map: &HashMap, key: &str) -> Option { Self::get_string(map, key).and_then(|s| s.to_string_lossy().parse().ok()) } } #[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<()> { // Connect let mach_port_name = Self::build_service_name()?; let mut client = XPCClient::connect(&mach_port_name); // Build arguments: limit=100, offset=0 (string-encoded for portability) let mut args = HashMap::new(); args.insert(CString::new("limit").unwrap(), Message::String(CString::new("100").unwrap())); args.insert(CString::new("offset").unwrap(), Message::String(CString::new("0").unwrap())); // Call let reply = self .call_method(&mut client, GET_CONVERSATIONS_METHOD, Some(args)) .await?; // Expect an array under "conversations" match reply.get(&Self::key("conversations")) { Some(Message::Array(items)) => { println!("Number of conversations: {}", items.len()); for item in items { if let Message::Dictionary(map) = item { // Convert to PrintableConversation let guid = Self::get_string(map, "guid").map(|s| s.to_string_lossy().into_owned()).unwrap_or_default(); let display_name = Self::get_string(map, "display_name").map(|s| s.to_string_lossy().into_owned()); let last_preview = Self::get_string(map, "last_message_preview").map(|s| s.to_string_lossy().into_owned()); let unread_count = Self::get_i64_from_str(map, "unread_count").unwrap_or(0) as i32; let date_ts: i64 = Self::get_i64_from_str(map, "date").unwrap_or(0); let participants: Vec = match map.get(&Self::key("participants")) { Some(Message::Array(arr)) => arr.iter().filter_map(|m| match m { Message::String(s) => Some(s.to_string_lossy().into_owned()), _ => None }).collect(), _ => Vec::new(), }; // Build PrintableConversation directly let conv = crate::printers::PrintableConversation { guid, display_name, last_message_preview: last_preview, unread_count, date: time::OffsetDateTime::from_unix_timestamp(date_ts).unwrap_or_else(|_| time::OffsetDateTime::UNIX_EPOCH), participants, }; println!("{}", crate::printers::ConversationPrinter::new(&conv)); } } Ok(()) } Some(other) => Err(anyhow::anyhow!("Unexpected conversations payload: {:?}", other)), None => Err(anyhow::anyhow!("Missing conversations in reply")), } } async fn sync_conversations(&mut self, _conversation_id: Option) -> 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, ) -> 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")) } }