Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502use std::{collections::BTreeSet, path::Path};use time::{Duration, OffsetDateTime};
use crate::{ config::{AppaConfig, ConfiguredFolder}, domain::{ AuditEvent, AuditEventKind, DeviceId, EntryKind, FolderRoster, MemberRole, RosterMember, is_conflict_artifact, manifest_root_hash, }, iroh::NodeHost, protocol::{INVITATION_PROTOCOL_VERSION, Invite, decode_invite, encode_invite}, storage::{AppPaths, AuditVerification, FolderConfig, ManifestRevision, PeerInfo, StateStore},};
pub type AppResult<T> = anyhow::Result<T>;
mod manifest;mod materialize;pub(crate) mod run;mod sync;
pub(super) const MAX_CONCURRENT_BLOB_TRANSFERS: usize = 4;pub(super) const MAX_CONCURRENT_PEER_REQUESTS: usize = 4;const INVITE_TTL: Duration = Duration::hours(24);const INVITE_ONLINE_WAIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);const OWNER_DISPLAY_NAME: &str = "owner";const JOINED_DEVICE_DISPLAY_NAME: &str = "joined device";
use materialize::restore_manifest;
pub(super) struct SyncContext<'a> { pub(super) folder: &'a FolderConfig, pub(super) node: &'a NodeHost, pub(super) state_store: &'a StateStore, pub(super) now: OffsetDateTime,}
pub struct AppaService { paths: AppPaths, state_store: StateStore,}
#[derive(Clone, Debug)]pub struct FolderStatus { pub folder: FolderConfig, pub peers: Vec<PeerInfo>, pub member_count: usize, pub file_count: usize, pub directory_count: usize, pub deleted_count: usize, pub conflict_count: usize, pub history_revision_count: u64, pub last_successful_sync: Option<OffsetDateTime>, pub last_sync_error: Option<String>,}
#[derive(Clone, Debug)]pub struct DoctorReport { pub folder_count: u64, pub issues: Vec<String>,}
#[derive(Clone, Debug)]pub struct ConflictInfo { pub path: String, pub author_device_id: DeviceId, pub modified_at: OffsetDateTime,}
#[derive(Clone, Debug)]pub struct MemberInfo { pub device_id: String, pub label: Option<String>,}
impl AppaService { pub fn open() -> AppResult<Self> { Self::open_at(AppPaths::discover()?) } pub fn open_at(paths: AppPaths) -> AppResult<Self> { Ok(Self { state_store: StateStore::open(&paths)?, paths, }) } pub fn register_folder(&self, folder_path: &Path) -> AppResult<FolderConfig> { let folder = self.state_store.register_folder(folder_path)?; self.ensure_owner_roster(&folder)?; Ok(folder) }
fn ensure_owner_roster(&self, folder: &FolderConfig) -> AppResult<()> { if self.state_store.load_roster(folder.id)?.is_some() { return Ok(()); } let owner_identity = self.paths.load_identity()?; let owner = RosterMember { device_id: owner_identity.public().to_string(), display_name: Some(OWNER_DISPLAY_NAME.to_owned()), role: MemberRole::Owner, }; let roster = FolderRoster::create(folder.id, folder.capability.clone(), owner)? .sign(&owner_identity)?; let audit_event = self.roster_audit_event(&roster, OffsetDateTime::now_utc())?; self.state_store .save_roster_with_audit(&roster, &audit_event) }
pub fn export_folder_inventory(&self, config_path: &Path) -> AppResult<()> { AppaConfig::write_template(config_path, &self.state_store.folders()?) }
pub fn validate_folder_inventory(&self, config_path: &Path) -> AppResult<()> { let config = AppaConfig::read(config_path)?; for folder in &config.folders { if !folder.path.is_dir() { anyhow::bail!("{} is not a directory", folder.path.display()); } } Ok(()) }
pub fn import_folder_inventory(&self, config_path: &Path) -> AppResult<Vec<FolderConfig>> { let config = AppaConfig::read(config_path)?; let folders = config .folders .iter() .map(|folder| self.import_folder(folder)) .collect::<AppResult<Vec<_>>>()?; Ok(folders) }
pub async fn create_invite(&self, folder_path: &Path) -> AppResult<String> { let node = self.load_node().await?; let result = self.create_invite_with_node(folder_path, &node).await; node.shutdown().await?; result }
pub(crate) async fn create_invite_with_node( &self, folder_path: &Path, node: &NodeHost, ) -> AppResult<String> { self.create_invite_with_node_at(folder_path, node, OffsetDateTime::now_utc()) .await }
async fn create_invite_with_node_at( &self, folder_path: &Path, node: &NodeHost, now: OffsetDateTime, ) -> AppResult<String> { let folder = self.require_folder(folder_path)?; let roster = self .state_store .load_roster(folder.id)? .ok_or_else(|| anyhow::anyhow!("folder has no signed roster"))?; node.wait_until_online_for(INVITE_ONLINE_WAIT_TIMEOUT) .await?; let invite = Invite { protocol_version: INVITATION_PROTOCOL_VERSION, folder_id: folder.id, folder_name: folder.name.clone(), inviter_device_id: node.endpoint_address().id.to_string(), inviter_endpoint: node.endpoint_address(), capability: folder.capability.clone(), roster, expires_at: now + INVITE_TTL, signature: None, }; let invite = invite.sign(&self.paths.load_identity()?)?; encode_invite(&invite) }
pub fn join_folder(&self, folder_path: &Path, ticket: &str) -> AppResult<FolderConfig> { let invite = decode_invite(ticket)?; let folder = self.state_store.import_invite(folder_path, &invite)?; Ok(folder) }
pub fn forget_folder(&self, folder_path: &Path) -> AppResult<()> { let folder = self.require_folder(folder_path)?; self.state_store.forget_folder(&folder) }
pub fn status(&self, folder_path: &Path) -> AppResult<FolderStatus> { let folder = self.require_folder(folder_path)?; self.status_for_folder(folder) }
pub fn statuses(&self) -> AppResult<Vec<FolderStatus>> { self.state_store .folders()? .into_iter() .map(|folder| self.status_for_folder(folder)) .collect() }
pub fn peers(&self, folder_path: &Path) -> AppResult<Vec<PeerInfo>> { let folder = self.require_folder(folder_path)?; self.state_store.peer_info(folder.id) }
pub fn history(&self, folder_path: &Path) -> AppResult<Vec<ManifestRevision>> { let folder = self.require_folder(folder_path)?; self.state_store.manifest_history(folder.id) }
pub fn audit_log(&self, folder_path: &Path) -> AppResult<Vec<crate::domain::AuditEvent>> { let folder = self.require_folder(folder_path)?; self.state_store.audit_events(folder.id) }
pub fn verify_audit(&self, folder_path: &Path) -> AppResult<AuditVerification> { let folder = self.require_folder(folder_path)?; self.state_store.verify_audit(folder.id) }
pub fn conflicts(&self, folder_path: &Path) -> AppResult<Vec<ConflictInfo>> { let folder = self.require_folder(folder_path)?; let manifest = self.state_store.load_manifest(folder.id)?; Ok(manifest .entries .values() .filter(|entry| is_conflict_artifact(&entry.path)) .map(|entry| ConflictInfo { path: entry.path.clone(), author_device_id: entry.author_device_id.clone(), modified_at: entry.modified_at, }) .collect()) }
pub async fn restore(&self, folder_path: &Path, revision: i64) -> AppResult<usize> { let folder = self.require_folder(folder_path)?; let historical_manifest = self.state_store.manifest_revision(folder.id, revision)?; let current_manifest = self.state_store.load_manifest(folder.id)?; let node = self.load_node().await?; let context = SyncContext { folder: &folder, node: &node, state_store: &self.state_store, now: OffsetDateTime::now_utc(), }; let result = restore_manifest(&context, ¤t_manifest, &historical_manifest).await; node.shutdown().await?; result }
pub fn device_identity(&self) -> AppResult<String> { self.device_id() }
pub fn export_identity(&self, destination: &Path) -> AppResult<()> { self.paths.export_identity(destination) }
pub fn import_identity(&self, source: &Path, replace_existing: bool) -> AppResult<()> { self.paths.import_identity(source, replace_existing) }
pub fn members(&self, folder_path: &Path) -> AppResult<Vec<MemberInfo>> { let folder = self.require_folder(folder_path)?; let roster = self.load_roster(folder.id)?; Ok(roster .members .values() .map(|member| MemberInfo { device_id: member.device_id.clone(), label: member.display_name.clone(), }) .collect()) }
pub fn can_revoke_members(&self, folder_path: &Path) -> AppResult<bool> { let folder = self.require_folder(folder_path)?; Ok(self.load_roster(folder.id)?.owner_device_id == self.device_id()?) }
pub fn revoke_member(&self, folder_path: &Path, device_id: &str) -> AppResult<()> { let folder = self.require_folder(folder_path)?; let roster = self.load_roster(folder.id)?; if roster.owner_device_id != self.device_id()? { anyhow::bail!("only the folder owner can revoke a member"); } let roster = roster .remove_member(device_id)? .rotate_capability(self.state_store.new_capability()) .sign(&self.paths.load_identity()?)?; let audit_event = self.roster_audit_event(&roster, OffsetDateTime::now_utc())?; self.state_store.replace_capability_and_roster_with_audit( folder.id, &roster, &audit_event, )?; Ok(()) }
fn status_for_folder(&self, folder: FolderConfig) -> AppResult<FolderStatus> { let manifest = self.state_store.load_manifest(folder.id)?; let peers = self.state_store.peer_info(folder.id)?; Ok(FolderStatus { member_count: self.active_member_ids(folder.id)?.len(), peers, file_count: count_entries(&manifest, EntryKind::File), directory_count: count_entries(&manifest, EntryKind::Directory), deleted_count: count_entries(&manifest, EntryKind::Deleted), conflict_count: manifest .entries .keys() .filter(|path| is_conflict_artifact(path)) .count(), history_revision_count: self.state_store.manifest_history_count(folder.id)?, last_successful_sync: self.state_store.last_successful_sync(folder.id)?, last_sync_error: self.state_store.last_sync_error(folder.id)?, folder, }) }
pub fn doctor(&self) -> AppResult<DoctorReport> { self.paths.load_identity()?; let folders = self.state_store.folders()?; let mut issues = Vec::new(); for folder in &folders { if !folder.path.is_dir() { issues.push(format!( "{} is no longer a directory", folder.path.display() )); continue; } if let Some(roster) = self.state_store.load_roster(folder.id)? { if let Err(error) = roster.validate() { issues.push(format!( "{} has an invalid roster: {error}", folder.path.display() )); } } else { issues.push(format!("{} has no signed roster", folder.path.display())); } if self.state_store.load_manifest(folder.id).is_err() { issues.push(format!("{} has no local manifest", folder.path.display())); } let audit_verification = self.state_store.verify_audit(folder.id)?; if !audit_verification.is_valid() { issues.push(format!( "{} has {} audit integrity fault(s)", folder.path.display(), audit_verification.faults.len() )); } if let Some(error) = self.state_store.last_sync_error(folder.id)? { issues.push(format!( "{} last sync failed: {error}", folder.path.display() )); } if self.active_member_ids(folder.id)?.is_empty() { issues.push(format!("{} has no active members", folder.path.display())); } } Ok(DoctorReport { folder_count: u64::try_from(folders.len())?, issues, }) }
#[cfg(test)] pub async fn sync_once(&self, folder_path: &Path) -> AppResult<usize> { let folder = self.require_folder(folder_path)?; let node = self.load_node().await?; let result = self.sync_with_node(&folder, &node).await; if let Err(error) = &result { self.state_store.record_sync_failure(folder.id, error)?; } node.shutdown().await?; result }
async fn load_node(&self) -> AppResult<NodeHost> { NodeHost::load(&self.paths.data_directory, self.paths.load_identity()?).await }
fn require_folder(&self, folder_path: &Path) -> AppResult<FolderConfig> { self.state_store.find_folder(folder_path)?.ok_or_else(|| { anyhow::anyhow!( "Appa has not been initialized for {}", folder_path.display() ) }) }
fn import_folder(&self, folder: &ConfiguredFolder) -> AppResult<FolderConfig> { let configured_folder = self.state_store.register_configured_folder( &folder.path, folder.name.clone(), folder.mode, )?; self.ensure_owner_roster(&configured_folder)?; Ok(configured_folder) }
fn load_roster(&self, folder_id: crate::domain::FolderId) -> AppResult<FolderRoster> { self.state_store .load_roster(folder_id)? .ok_or_else(|| anyhow::anyhow!("folder has no signed roster")) }
fn active_member_ids( &self, folder_id: crate::domain::FolderId, ) -> AppResult<BTreeSet<DeviceId>> { Ok(self.load_roster(folder_id)?.active_device_ids()) }
fn enroll_discovered_member( &self, folder_id: crate::domain::FolderId, device_id: &str, ) -> AppResult<()> { let roster = self.load_roster(folder_id)?; if roster.contains_member(device_id) { return Ok(()); } if roster.owner_device_id != self.device_id()? { tracing::debug!(folder = %folder_id, device = %device_id, "Only the folder owner can enroll a discovered device"); return Ok(()); } let roster = roster .add_member(RosterMember { device_id: device_id.to_owned(), display_name: Some(JOINED_DEVICE_DISPLAY_NAME.to_owned()), role: MemberRole::Member, })? .sign(&self.paths.load_identity()?)?; let audit_event = self.roster_audit_event(&roster, OffsetDateTime::now_utc())?; self.state_store .save_roster_with_audit(&roster, &audit_event) }
fn roster_audit_event( &self, roster: &FolderRoster, now: OffsetDateTime, ) -> AppResult<AuditEvent> { let identity = self.paths.load_identity()?; let author_device_id = identity.public().to_string(); if roster.owner_device_id != author_device_id { anyhow::bail!("only the folder owner can sign a roster audit event"); } let (sequence, parent_hash) = self .state_store .next_audit_sequence_and_parent(roster.folder_id, &author_device_id)?; let manifest = self.state_store.load_manifest(roster.folder_id)?; let event = AuditEvent::create( roster.folder_id, author_device_id, sequence, parent_hash, AuditEventKind::RosterUpdated, Some(manifest_root_hash(&manifest)?), roster.hash()?, roster.clone(), now, ); event.sign(&identity) }
fn device_id(&self) -> AppResult<String> { Ok(self.paths.load_identity()?.public().to_string()) }
fn folders_to_run(&self, folder_path: Option<&Path>) -> AppResult<Vec<FolderConfig>> { match folder_path { Some(path) => Ok(vec![self.require_folder(path)?]), None => self.state_store.folders(), } }
fn folders_to_watch(&self, folder_path: Option<&Path>) -> AppResult<Vec<FolderConfig>> { match folder_path { Some(path) => Ok(self.state_store.find_folder(path)?.into_iter().collect()), None => self.state_store.folders(), } }}
fn count_entries(manifest: &crate::domain::Manifest, kind: EntryKind) -> usize { manifest .entries .values() .filter(|entry| entry.kind == kind) .count()}
#[cfg(test)]#[path = "app/tests.rs"]mod tests;