Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310//! Local IPC protocol and client for the Appa daemon.
use std::{io, path::PathBuf, time::Duration};
use anyhow::Context;use serde::{Deserialize, Serialize};use tokio::{ io::{AsyncReadExt, AsyncWriteExt}, net::UnixStream,};
use crate::{ core::{Dashboard, FolderDetails}, storage::AppPaths,};
pub const IPC_PROTOCOL_VERSION: u16 = 1;const MAX_MESSAGE_BYTES: usize = 1024 * 1024;const DAEMON_REQUEST_TIMEOUT: Duration = Duration::from_secs(8);
#[derive(Debug, Deserialize, Serialize)]pub(crate) struct Request { pub protocol_version: u16, pub command: Command,}
#[derive(Debug, Deserialize, Serialize)]pub(crate) enum Command { Dashboard, FolderDetails { folder_path: PathBuf, }, RegisterFolder { folder_path: PathBuf, }, ImportFolderInventory { inventory_path: PathBuf, }, JoinFolder { folder_path: PathBuf, invitation: String, }, CreateInvite { folder_path: PathBuf, }, RevokeMember { folder_path: PathBuf, device_id: String, }, ForgetFolder { folder_path: PathBuf, }, RestoreRevision { folder_path: PathBuf, revision: i64, }, Status { folder_path: Option<PathBuf>, json: bool, }, Peers { folder_path: PathBuf, json: bool, }, History { folder_path: PathBuf, }, Audit { folder_path: PathBuf, verify: bool, }, Conflicts { folder_path: PathBuf, }, Members { folder_path: PathBuf, }, Doctor { json: bool, }, /// Drop persistent legacy blob tags; periodic GC reclaims orphaned bytes. DropLegacyTags { json: bool, },}
#[derive(Debug, Deserialize, Serialize)]pub(crate) enum Response { Dashboard(Dashboard), FolderDetails(FolderDetails), Invitation(String), Text(String), Count(usize), Success, Error(String),}
/// A short-lived connection to the user-owned Appa daemon.#[derive(Clone, Debug)]pub struct DaemonClient { socket_path: PathBuf,}
impl DaemonClient { pub fn connect() -> anyhow::Result<Self> { Ok(Self { socket_path: AppPaths::discover()?.daemon_socket_path(), }) }
pub async fn dashboard(&self) -> anyhow::Result<Dashboard> { match self.request(Command::Dashboard).await? { Response::Dashboard(dashboard) => Ok(dashboard), response => unexpected_response(response), } }
pub async fn folder_details(&self, folder_path: PathBuf) -> anyhow::Result<FolderDetails> { match self.request(Command::FolderDetails { folder_path }).await? { Response::FolderDetails(details) => Ok(details), response => unexpected_response(response), } }
pub async fn register_folder(&self, folder_path: PathBuf) -> anyhow::Result<()> { self.expect_success(Command::RegisterFolder { folder_path }) .await }
pub async fn import_folder_inventory(&self, inventory_path: PathBuf) -> anyhow::Result<usize> { match self .request(Command::ImportFolderInventory { inventory_path }) .await? { Response::Count(count) => Ok(count), response => unexpected_response(response), } }
pub async fn join_folder( &self, folder_path: PathBuf, invitation: String, ) -> anyhow::Result<()> { self.expect_success(Command::JoinFolder { folder_path, invitation, }) .await }
pub async fn create_invite(&self, folder_path: PathBuf) -> anyhow::Result<String> { match self.request(Command::CreateInvite { folder_path }).await? { Response::Invitation(invitation) => Ok(invitation), response => unexpected_response(response), } }
pub async fn revoke_member( &self, folder_path: PathBuf, device_id: String, ) -> anyhow::Result<()> { self.expect_success(Command::RevokeMember { folder_path, device_id, }) .await }
pub async fn forget_folder(&self, folder_path: PathBuf) -> anyhow::Result<()> { self.expect_success(Command::ForgetFolder { folder_path }) .await }
pub async fn restore_revision( &self, folder_path: PathBuf, revision: i64, ) -> anyhow::Result<usize> { match self .request(Command::RestoreRevision { folder_path, revision, }) .await? { Response::Count(count) => Ok(count), response => unexpected_response(response), } }
pub async fn status(&self, folder_path: Option<PathBuf>, json: bool) -> anyhow::Result<String> { self.request_text(Command::Status { folder_path, json }) .await }
pub async fn peers(&self, folder_path: PathBuf, json: bool) -> anyhow::Result<String> { self.request_text(Command::Peers { folder_path, json }) .await }
pub async fn history(&self, folder_path: PathBuf) -> anyhow::Result<String> { self.request_text(Command::History { folder_path }).await }
pub async fn audit(&self, folder_path: PathBuf, verify: bool) -> anyhow::Result<String> { self.request_text(Command::Audit { folder_path, verify, }) .await }
pub async fn conflicts(&self, folder_path: PathBuf) -> anyhow::Result<String> { self.request_text(Command::Conflicts { folder_path }).await }
pub async fn members(&self, folder_path: PathBuf) -> anyhow::Result<String> { self.request_text(Command::Members { folder_path }).await }
pub async fn doctor(&self, json: bool) -> anyhow::Result<String> { self.request_text(Command::Doctor { json }).await }
pub async fn drop_legacy_tags(&self, json: bool) -> anyhow::Result<String> { self.request_text(Command::DropLegacyTags { json }).await }
async fn expect_success(&self, command: Command) -> anyhow::Result<()> { match self.request(command).await? { Response::Success => Ok(()), response => unexpected_response(response), } }
async fn request_text(&self, command: Command) -> anyhow::Result<String> { match self.request(command).await? { Response::Text(text) => Ok(text), response => unexpected_response(response), } }
async fn request(&self, command: Command) -> anyhow::Result<Response> { tokio::time::timeout(DAEMON_REQUEST_TIMEOUT, self.request_inner(command)) .await .map_err(|_| { anyhow::anyhow!( "Appa daemon did not respond within {} seconds", DAEMON_REQUEST_TIMEOUT.as_secs() ) })? }
async fn request_inner(&self, command: Command) -> anyhow::Result<Response> { let mut stream = UnixStream::connect(&self.socket_path) .await .with_context(|| { format!( "could not connect to Appa daemon at {}", self.socket_path.display() ) })?; write_message( &mut stream, &Request { protocol_version: IPC_PROTOCOL_VERSION, command, }, ) .await?; read_message(&mut stream).await }}
fn unexpected_response<T>(response: Response) -> anyhow::Result<T> { match response { Response::Error(error) => Err(anyhow::anyhow!(error)), response => Err(anyhow::anyhow!( "Appa daemon returned an unexpected response: {response:?}" )), }}
pub(crate) async fn read_message<T: for<'de> Deserialize<'de>>( stream: &mut UnixStream,) -> anyhow::Result<T> { let length = stream.read_u32().await? as usize; if length > MAX_MESSAGE_BYTES { anyhow::bail!("Appa daemon message exceeds {MAX_MESSAGE_BYTES} bytes"); } let mut bytes = vec![0; length]; stream.read_exact(&mut bytes).await?; Ok(serde_json::from_slice(&bytes)?)}
pub(crate) async fn write_message<T: Serialize>( stream: &mut UnixStream, message: &T,) -> anyhow::Result<()> { let bytes = serde_json::to_vec(message)?; let length = u32::try_from(bytes.len()) .map_err(|_| io::Error::other("Appa daemon message is too large"))?; stream.write_u32(length).await?; stream.write_all(&bytes).await?; stream.flush().await?; Ok(())}