Something went wrong. Try again.
Hangar is your Bluesky home base—built natively for Linux. ✈️ hangar.blue
libadwaita gtk4 bluesky flatpak gnome atprotocol linux atproto appimage rust
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360// SPDX-License-Identifier: MPL-2.0
use crate::cache::CacheError;use crate::cache::schema::{MIGRATION_3, SCHEMA, SCHEMA_VERSION};use rusqlite::Connection;use std::path::{Path, PathBuf};use std::sync::{Arc, Mutex};
/// Nothing in here is authoritative, it all refetches, so paying an fsync per/// commit buys nothing and a home feed refresh commits six times on the main/// thread. WAL plus NORMAL keeps that off the sync path. The busy timeout/// covers the connection being shared across threads.const PRAGMAS: &str = "PRAGMA journal_mode = WAL;PRAGMA synchronous = NORMAL;PRAGMA busy_timeout = 5000;";
/// Handle to the cache database for a specific user#[derive(Clone)]pub struct CacheDb { conn: Arc<Mutex<Connection>>, #[allow(dead_code)] user_did: String,}
impl CacheDb { /// Open or create cache database for user /// Path: ~/.local/share/hangar/{user_did}/cache.db pub fn open(user_did: &str) -> Result<Self, CacheError> { let path = Self::cache_path(user_did)?;
// Ensure parent directory exists if let Some(parent) = path.parent() { std::fs::create_dir_all(parent) .map_err(|e| CacheError::Path(format!("failed to create cache dir: {}", e)))?; }
Ok(Self { conn: Arc::new(Mutex::new(Self::open_resilient(&path)?)), user_did: user_did.to_string(), }) }
/// Open the file, and if it cannot be opened or migrated, throw it away and /// build a fresh one. A cache we refuse to touch costs the user every read /// for the whole session; a rebuilt one costs a single refresh. fn open_resilient(path: &Path) -> Result<Connection, CacheError> { match Self::prepare(path) { Ok(conn) => Ok(conn), Err(e) => { eprintln!("Cache at {} unusable, rebuilding: {}", path.display(), e); Self::discard(path)?; Self::prepare(path) } } }
/// One connection, pragmas applied, schema current. fn prepare(path: &Path) -> Result<Connection, CacheError> { let conn = Connection::open(path)?; conn.execute_batch(PRAGMAS)?; Self::check_integrity(&conn)?; Self::migrate(&conn)?; Ok(conn) }
/// `Connection::open` accepts any file at all, so a truncated or scribbled /// cache only fails later, one swallowed read at a time. fn check_integrity(conn: &Connection) -> Result<(), CacheError> { let result: String = conn.query_row("PRAGMA quick_check(1)", [], |row| row.get(0))?; if result == "ok" { Ok(()) } else { Err(corrupt(&result)) } }
/// Remove the database and its WAL sidecars so a retry starts clean. fn discard(path: &Path) -> Result<(), CacheError> { for suffix in ["", "-wal", "-shm"] { let mut name = path.as_os_str().to_os_string(); name.push(suffix); let file = PathBuf::from(name); if let Err(e) = std::fs::remove_file(&file) && e.kind() != std::io::ErrorKind::NotFound { return Err(CacheError::Path(format!( "failed to remove {}: {}", file.display(), e ))); } } Ok(()) }
/// Bring the schema up to `SCHEMA_VERSION`. A fresh file is at version 0 /// and takes the whole schema; older files walk the steps they missed. fn migrate(conn: &Connection) -> Result<(), CacheError> { let version: i64 = conn.query_row("PRAGMA user_version", [], |row| row.get(0))?;
if version == SCHEMA_VERSION { return Ok(()); } if version > SCHEMA_VERSION { // Written by a newer build. Rebuilding costs one refresh; reading a // schema we do not know costs correctness. return Err(corrupt(&format!( "schema is version {}, this build understands {}", version, SCHEMA_VERSION ))); }
let tx = conn.unchecked_transaction()?; if version < 1 { tx.execute_batch(SCHEMA)?; } // Versions 1 and 2 differ only in tables version 3 drops, so both land // here with nothing left to add. if version < 3 { tx.execute_batch(MIGRATION_3)?; } tx.execute_batch(&format!("PRAGMA user_version = {}", SCHEMA_VERSION))?; tx.commit()?; Ok(()) }
/// Get XDG data directory for cache /// Where this account's cache lives. Public so Clear Cache can delete /// it after the connection is dropped. pub fn cache_path(user_did: &str) -> Result<PathBuf, CacheError> { let data_dir = dirs::data_dir() .ok_or_else(|| CacheError::Path("could not find data directory".to_string()))?;
// Sanitize DID for filesystem (replace : with _) let safe_did = user_did.replace(':', "_");
Ok(data_dir.join("hangar").join(safe_did).join("cache.db")) }
/// Access connection for operations pub fn conn(&self) -> std::sync::MutexGuard<'_, Connection> { self.conn.lock().expect("cache lock poisoned") }
/// Get current unix timestamp pub fn now() -> i64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_secs() as i64) .unwrap_or(0) }
/// Cleanup old entries with sensible defaults: /// - Feed items: 24 hours (feed order changes constantly) /// - Orphan posts: 7 days (posts not in any feed) /// - Profiles: kept if any post references them pub fn cleanup_stale(&self) -> Result<(), CacheError> { let conn = self.conn(); let now = Self::now();
// Feed items expire after 24 hours; the ordering is stale by then let feed_cutoff = now - (24 * 60 * 60); conn.execute("DELETE FROM feed_items WHERE fetched_at < ?", [feed_cutoff])?;
// Feed state older than 24 hours conn.execute( "DELETE FROM feed_state WHERE last_refresh_at < ?", [feed_cutoff], )?;
// Orphan posts (not in any feed) older than 7 days let post_cutoff = now - (7 * 24 * 60 * 60); conn.execute( r#" DELETE FROM posts WHERE fetched_at < ? AND uri NOT IN (SELECT post_uri FROM feed_items) "#, [post_cutoff], )?;
// Orphan profiles (no posts reference them) older than 7 days conn.execute( r#" DELETE FROM profiles WHERE fetched_at < ? AND did NOT IN (SELECT author_did FROM posts) "#, [post_cutoff], )?;
Ok(()) }}
/// A cache we cannot make sense of is corrupt as far as callers care, and it/// takes the same rebuild path as a truncated file.fn corrupt(detail: &str) -> CacheError { CacheError::Database(rusqlite::Error::SqliteFailure( rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_CORRUPT), Some(detail.to_string()), ))}
#[cfg(test)]mod tests { use super::*; use std::sync::atomic::{AtomicU32, Ordering};
static COUNTER: AtomicU32 = AtomicU32::new(0);
/// A unique cache path under the system temp dir, removed on drop. struct TempDb(PathBuf);
impl TempDb { fn new() -> Self { let n = COUNTER.fetch_add(1, Ordering::Relaxed); let dir = std::env::temp_dir().join(format!("hangar-cache-{}-{}", std::process::id(), n)); std::fs::create_dir_all(&dir).expect("temp dir"); Self(dir.join("cache.db")) }
fn path(&self) -> &Path { &self.0 } }
impl Drop for TempDb { fn drop(&mut self) { if let Some(dir) = self.0.parent() { let _ = std::fs::remove_dir_all(dir); } } }
fn version(conn: &Connection) -> i64 { conn.query_row("PRAGMA user_version", [], |row| row.get(0)) .expect("user_version") }
fn table_exists(conn: &Connection, name: &str) -> bool { conn.query_row( "SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = ?", [name], |row| row.get::<_, i64>(0), ) .expect("sqlite_master") > 0 }
#[test] fn fresh_cache_is_wal_and_unsynced() { let db = TempDb::new(); let conn = CacheDb::open_resilient(db.path()).expect("open");
let journal: String = conn .query_row("PRAGMA journal_mode", [], |row| row.get(0)) .expect("journal_mode"); assert_eq!(journal, "wal");
let synchronous: i64 = conn .query_row("PRAGMA synchronous", [], |row| row.get(0)) .expect("synchronous"); assert_eq!(synchronous, 1, "NORMAL");
let busy: i64 = conn .query_row("PRAGMA busy_timeout", [], |row| row.get(0)) .expect("busy_timeout"); assert_eq!(busy, 5000); }
#[test] fn fresh_cache_is_stamped_and_reopens_clean() { let db = TempDb::new(); let conn = CacheDb::open_resilient(db.path()).expect("open"); assert_eq!(version(&conn), SCHEMA_VERSION); conn.execute( "INSERT INTO feed_state (feed_key, has_more) VALUES ('home', 1)", [], ) .expect("insert"); drop(conn);
// Reopening is a no-op migration, so the row survives. let conn = CacheDb::open_resilient(db.path()).expect("reopen"); assert_eq!(version(&conn), SCHEMA_VERSION); let rows: i64 = conn .query_row("SELECT count(*) FROM feed_state", [], |row| row.get(0)) .expect("count"); assert_eq!(rows, 1); }
#[test] fn version_two_cache_upgrades_in_place() { let db = TempDb::new(); { let conn = Connection::open(db.path()).expect("open"); conn.execute_batch(SCHEMA).expect("schema"); conn.execute_batch( "CREATE TABLE IF NOT EXISTS notifications (uri TEXT PRIMARY KEY); CREATE TABLE IF NOT EXISTS images (url TEXT PRIMARY KEY); PRAGMA user_version = 2;", ) .expect("legacy tables"); conn.execute( "INSERT INTO feed_state (feed_key, has_more) VALUES ('home', 1)", [], ) .expect("insert"); }
let conn = CacheDb::open_resilient(db.path()).expect("open"); assert_eq!(version(&conn), SCHEMA_VERSION); assert!(!table_exists(&conn, "notifications")); assert!(!table_exists(&conn, "images"));
// The upgrade is in place, so cached rows are not thrown away. let rows: i64 = conn .query_row("SELECT count(*) FROM feed_state", [], |row| row.get(0)) .expect("count"); assert_eq!(rows, 1); }
#[test] fn corrupt_file_is_rebuilt() { let db = TempDb::new(); std::fs::write(db.path(), b"SQLite format 3\0this is not a database") .expect("write garbage");
let conn = CacheDb::open_resilient(db.path()).expect("rebuild"); assert_eq!(version(&conn), SCHEMA_VERSION); assert!(table_exists(&conn, "posts")); }
#[test] fn cache_from_a_newer_build_is_rebuilt() { let db = TempDb::new(); { let conn = CacheDb::open_resilient(db.path()).expect("open"); conn.execute_batch("PRAGMA user_version = 99") .expect("bump"); conn.execute( "INSERT INTO feed_state (feed_key, has_more) VALUES ('home', 1)", [], ) .expect("insert"); }
let conn = CacheDb::open_resilient(db.path()).expect("rebuild"); assert_eq!(version(&conn), SCHEMA_VERSION); let rows: i64 = conn .query_row("SELECT count(*) FROM feed_state", [], |row| row.get(0)) .expect("count"); assert_eq!(rows, 0, "the unreadable cache was thrown away"); }}