Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387//! SQLite-backed Appa identity, folders, peers, and manifests.
use std::{ env, path::{Path, PathBuf},};
use anyhow::Context;use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};use iroh::SecretKey;use rusqlite::{Connection, OptionalExtension, params};use uuid::Uuid;
use crate::{ domain::{Entry, FolderId, FolderMode, Manifest}, protocol::Invite,};
mod audit;mod fingerprints;mod live_blobs;mod manifests;mod materializations;mod paths;mod peers;mod rosters;mod schema;mod sync_state;
const FOLDER_SCOPED_TABLES: &[&str] = &[ "known_peers", "join_invites", "folder_rosters", "manifests", "manifest_history_compact", "counters", "sync_state", "file_fingerprints", "pending_materializations", "audit_events", "audit_heads", "audit_integrity_faults",];
pub use audit::AuditVerification;pub use fingerprints::FileFingerprint;pub use manifests::ManifestRevision;pub use paths::AppPaths;pub use peers::PeerInfo;
const DEFAULT_HISTORY_REVISIONS: usize = 10;
pub struct StateStore { connection: Connection, history_limit: usize,}
#[derive(Clone, Debug)]pub struct FolderConfig { pub id: FolderId, pub name: String, pub path: PathBuf, pub capability: String, pub mode: FolderMode,}
#[derive(Clone, Debug)]pub struct PendingMaterialization { pub entries: Vec<Entry>, pub resulting_manifest: Manifest, pub is_legacy_single_entry: bool,}
impl StateStore { pub fn open(paths: &AppPaths) -> anyhow::Result<Self> { const DATABASE_BUSY_TIMEOUT_MILLIS: u64 = 5_000; let mut store = Self { connection: Connection::open(&paths.database_path).with_context(|| { format!( "could not open Appa state at {}", paths.database_path.display() ) })?, history_limit: history_revision_limit()?, }; store .connection .busy_timeout(std::time::Duration::from_millis( DATABASE_BUSY_TIMEOUT_MILLIS, ))?; store .connection .execute_batch("PRAGMA journal_mode = WAL; PRAGMA foreign_keys = ON;")?; store.initialize_schema()?; Ok(store) }
/// Opens a short-lived connection for the garbage collector's `add_protected` /// callback. This runs on the blob store's own runtime and needs independent /// access to the SQLite database to compute the set of live blob hashes. pub(crate) fn open_for_gc(database_path: &Path) -> anyhow::Result<Self> { const DATABASE_BUSY_TIMEOUT_MILLIS: u64 = 5_000; let store = Self { connection: Connection::open(database_path).with_context(|| { format!( "could not open Appa state for GC at {}", database_path.display() ) })?, // The history limit is irrelevant for a read-only live-set query; use // the default so validation never rejects the connection. history_limit: DEFAULT_HISTORY_REVISIONS, }; store .connection .busy_timeout(std::time::Duration::from_millis( DATABASE_BUSY_TIMEOUT_MILLIS, ))?; store .connection .execute_batch("PRAGMA journal_mode = WAL; PRAGMA query_only = ON;")?; Ok(store) }
pub fn register_folder(&self, path: &Path) -> anyhow::Result<FolderConfig> { if !path.is_dir() { anyhow::bail!("{} is not a directory", path.display()); } if let Some(folder) = self.find_folder(path)? { return Ok(folder); } let folder = FolderConfig { id: Uuid::new_v4(), name: folder_name(path)?, path: path.canonicalize()?, capability: create_capability(), mode: FolderMode::default(), }; self.connection.execute( "INSERT INTO folders (id, name, path, capability, mode) VALUES (?1, ?2, ?3, ?4, ?5)", params![ folder.id.to_string(), folder.name, folder.path.display().to_string(), folder.capability, serde_json::to_string(&folder.mode)? ], )?; self.save_manifest(&Manifest::empty(folder.id))?; Ok(folder) }
pub fn register_configured_folder( &self, path: &Path, name: Option<String>, mode: FolderMode, ) -> anyhow::Result<FolderConfig> { if !path.is_dir() { anyhow::bail!("{} is not a directory", path.display()); } if let Some(folder) = self.find_folder(path)? { return self.update_folder_mode(folder, mode); } let folder = FolderConfig { id: Uuid::new_v4(), name: name.unwrap_or(folder_name(path)?), path: path.canonicalize()?, capability: create_capability(), mode, }; self.connection.execute( "INSERT INTO folders (id, name, path, capability, mode) VALUES (?1, ?2, ?3, ?4, ?5)", params![ folder.id.to_string(), folder.name, folder.path.display().to_string(), folder.capability, serde_json::to_string(&folder.mode)? ], )?; self.save_manifest(&Manifest::empty(folder.id))?; Ok(folder) }
fn update_folder_mode( &self, mut folder: FolderConfig, mode: FolderMode, ) -> anyhow::Result<FolderConfig> { if folder.mode == mode { return Ok(folder); } self.connection.execute( "UPDATE folders SET mode = ?1 WHERE id = ?2", params![serde_json::to_string(&mode)?, folder.id.to_string()], )?; folder.mode = mode; Ok(folder) }
pub fn import_invite(&self, path: &Path, invite: &Invite) -> anyhow::Result<FolderConfig> { if !path.is_dir() { anyhow::bail!("{} is not a directory", path.display()); } if invite.roster.folder_id != invite.folder_id { anyhow::bail!("invitation roster belongs to a different folder"); } self.save_roster_if_newer(&invite.roster)?; if let Some(existing_folder) = self.find_folder(path)? && existing_folder.id != invite.folder_id { self.forget_folder(&existing_folder)?; } let folder = FolderConfig { id: invite.folder_id, name: invite.folder_name.clone(), path: path.canonicalize()?, capability: invite.capability.clone(), mode: FolderMode::default(), }; self.connection.execute("INSERT INTO folders (id, name, path, capability, mode) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(path) DO UPDATE SET id = excluded.id, name = excluded.name, capability = excluded.capability", params![folder.id.to_string(), folder.name, folder.path.display().to_string(), folder.capability, serde_json::to_string(&folder.mode)?])?; self.connection.execute( "INSERT INTO join_invites (folder_id, invite) VALUES (?1, ?2) ON CONFLICT(folder_id) DO UPDATE SET invite = excluded.invite", params![folder.id.to_string(), serde_json::to_string(invite)?], )?; self.save_manifest(&Manifest::empty(folder.id))?; self.save_peer(folder.id, &invite.inviter_endpoint)?; Ok(folder) }
pub fn join_invite(&self, folder_id: FolderId) -> anyhow::Result<Option<Invite>> { let invite: Option<String> = self .connection .query_row( "SELECT invite FROM join_invites WHERE folder_id = ?1", params![folder_id.to_string()], |row| row.get(0), ) .optional()?; invite .map(|value| serde_json::from_str(&value).map_err(Into::into)) .transpose() }
pub fn new_capability(&self) -> String { create_capability() }
pub fn update_capability(&self, folder_id: FolderId, capability: &str) -> anyhow::Result<()> { self.connection.execute( "UPDATE folders SET capability = ?1 WHERE id = ?2", params![capability, folder_id.to_string()], )?; Ok(()) }
pub fn find_folder(&self, path: &Path) -> anyhow::Result<Option<FolderConfig>> { let normalized_path = path.canonicalize()?; self.connection .query_row( "SELECT id, name, path, capability, mode FROM folders WHERE path = ?1", params![normalized_path.display().to_string()], folder_from_row, ) .optional() .map_err(Into::into) }
pub fn forget_folder(&self, folder: &FolderConfig) -> anyhow::Result<()> { let transaction = self.connection.unchecked_transaction()?; let folder_id = folder.id.to_string(); for table in FOLDER_SCOPED_TABLES { transaction.execute( &format!("DELETE FROM {table} WHERE folder_id = ?1"), params![folder_id], )?; } transaction.execute("DELETE FROM folders WHERE id = ?1", params![folder_id])?; transaction.commit()?; Ok(()) }
pub fn folders(&self) -> anyhow::Result<Vec<FolderConfig>> { let mut statement = self .connection .prepare("SELECT id, name, path, capability, mode FROM folders ORDER BY name, path")?; statement .query_map([], folder_from_row)? .collect::<Result<Vec<_>, _>>() .map_err(Into::into) }
fn initialize_schema(&mut self) -> anyhow::Result<()> { schema::initialize(&mut self.connection)?; Ok(()) }}
fn folder_from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<FolderConfig> { let id: String = row.get(0)?; let id = Uuid::parse_str(&id).map_err(|error| { rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(error)) })?; Ok(FolderConfig { id, name: row.get(1)?, path: PathBuf::from(row.get::<_, String>(2)?), capability: row.get(3)?, mode: serde_json::from_str(&row.get::<_, String>(4)?).map_err(|error| { rusqlite::Error::FromSqlConversionFailure( 4, rusqlite::types::Type::Text, Box::new(error), ) })?, })}
fn folder_name(path: &Path) -> anyhow::Result<String> { path.file_name() .and_then(|name| name.to_str()) .map(str::to_owned) .ok_or_else(|| anyhow::anyhow!("folder name must be valid UTF-8"))}
fn create_capability() -> String { URL_SAFE_NO_PAD.encode(SecretKey::generate().to_bytes())}
fn history_revision_limit() -> anyhow::Result<usize> { match env::var("APPA_HISTORY_REVISIONS") { Ok(value) => value .parse() .map_err(|_| anyhow::anyhow!("APPA_HISTORY_REVISIONS must be a positive integer")) .and_then(validate_history_revision_limit), Err(env::VarError::NotPresent) => Ok(DEFAULT_HISTORY_REVISIONS), Err(error) => Err(error.into()), }}
fn validate_history_revision_limit(limit: usize) -> anyhow::Result<usize> { if limit == 0 { anyhow::bail!("APPA_HISTORY_REVISIONS must be a positive integer"); } Ok(limit)}
const DEFAULT_GC_INTERVAL_SECS: u64 = 10 * 60;const MIN_GC_INTERVAL_SECS: u64 = 60;
pub fn gc_interval() -> anyhow::Result<std::time::Duration> { match env::var("APPA_GC_INTERVAL_SECS") { Ok(value) => value .parse() .map_err(|_| anyhow::anyhow!("APPA_GC_INTERVAL_SECS must be a positive integer")) .and_then(validate_gc_interval), Err(env::VarError::NotPresent) => { Ok(std::time::Duration::from_secs(DEFAULT_GC_INTERVAL_SECS)) } Err(error) => Err(error.into()), }}
fn validate_gc_interval(secs: u64) -> anyhow::Result<std::time::Duration> { if secs < MIN_GC_INTERVAL_SECS { anyhow::bail!("APPA_GC_INTERVAL_SECS must be at least {MIN_GC_INTERVAL_SECS} seconds"); } Ok(std::time::Duration::from_secs(secs))}
fn prune_manifest_history( transaction: &rusqlite::Transaction<'_>, folder_id: FolderId, history_limit: usize,) -> anyhow::Result<()> { transaction.execute( "DELETE FROM manifest_history_compact WHERE folder_id = ?1 AND revision NOT IN (SELECT revision FROM manifest_history_compact WHERE folder_id = ?1 ORDER BY revision DESC LIMIT ?2)", params![folder_id.to_string(), i64::try_from(history_limit)?], )?; Ok(())}
#[cfg(test)]#[path = "storage/tests.rs"]mod tests;