diff --git a/crates/epix-db/src/lib.rs b/crates/epix-db/src/lib.rs index 88c9d2f..e4388ce 100644 --- a/crates/epix-db/src/lib.rs +++ b/crates/epix-db/src/lib.rs @@ -66,6 +66,18 @@ impl Database { populate::populate(&conn, schema, db_dir.as_ref()) } + /// Populate, skipping data files whose path contains one of `exclude` + /// (ContentFilter mute enforcement - muted authors' files are left out). + pub fn populate_filtered( + &self, + schema: &DbSchema, + db_dir: impl AsRef, + exclude: &[String], + ) -> Result { + let conn = self.conn()?; + populate::populate_site_filtered(&conn, schema, db_dir.as_ref(), "", exclude) + } + /// Populate a version-3 merger db from one merged site's files, tagging the /// rows with `site`. Call once per merged site. pub fn populate_site( diff --git a/crates/epix-db/src/populate.rs b/crates/epix-db/src/populate.rs index e73be62..569a2f2 100644 --- a/crates/epix-db/src/populate.rs +++ b/crates/epix-db/src/populate.rs @@ -211,12 +211,25 @@ pub fn update_json( /// that match a map. `db_dir` is the xite's content root; paths are matched /// relative to it (forward slashes), like EpixNet. pub fn populate(conn: &Connection, schema: &DbSchema, db_dir: &Path) -> Result { - populate_site(conn, schema, db_dir, "") + populate_site_filtered(conn, schema, db_dir, "", &[]) } /// Like [`populate`], but tags every row with `site` - for a version-3 merger /// db aggregating data from several merged sites (call once per merged site). pub fn populate_site(conn: &Connection, schema: &DbSchema, db_dir: &Path, site: &str) -> Result { + populate_site_filtered(conn, schema, db_dir, site, &[]) +} + +/// Like [`populate_site`], but skips any data file whose path contains one of +/// `exclude` - the ContentFilter mute enforcement point (muted authors' +/// `data//…` files are left out of the database). +pub fn populate_site_filtered( + conn: &Connection, + schema: &DbSchema, + db_dir: &Path, + site: &str, + exclude: &[String], +) -> Result { let mut count = 0; let mut stack = vec![db_dir.to_path_buf()]; while let Some(dir) = stack.pop() { @@ -232,6 +245,9 @@ pub fn populate_site(conn: &Connection, schema: &DbSchema, db_dir: &Path, site: } let Ok(rel) = path.strip_prefix(db_dir) else { continue }; let rel_str = rel.to_string_lossy().replace('\\', "/"); + if !exclude.is_empty() && exclude.iter().any(|e| rel_str.contains(e.as_str())) { + continue; + } let Ok(bytes) = std::fs::read(&path) else { continue }; let Ok(data) = serde_json::from_slice::(&bytes) else { continue }; if update_json(conn, schema, &rel_str, &data, site)? { diff --git a/crates/epix-plugins/src/sidebar.rs b/crates/epix-plugins/src/sidebar.rs index 970e7f6..7246c16 100644 --- a/crates/epix-plugins/src/sidebar.rs +++ b/crates/epix-plugins/src/sidebar.rs @@ -84,22 +84,28 @@ impl WsCommand for SidebarGetPeers { } } -/// `siteSetSizeLimit` - set the per-xite size limit (MB). +/// `siteSetLimit` - set the per-xite size limit (MB). The sidebar's "Set" button +/// sends the input value positionally, as a string. struct SiteSetSizeLimit; #[async_trait] impl WsCommand for SiteSetSizeLimit { fn name(&self) -> &'static str { - "siteSetSizeLimit" + "siteSetLimit" } async fn handle(&self, s: &WsSession, p: &Value) -> Result { let address = s.address()?.to_string(); - let limit = p + let raw = p .get("size_limit") .or_else(|| p.as_array().and_then(|a| a.first())) - .and_then(|v| v.as_i64()) + .or(Some(p)) .ok_or("size_limit required")?; + // Accept a number or a numeric string (the button posts a string). + let limit = raw + .as_i64() + .or_else(|| raw.as_str().and_then(|s| s.trim().parse::().ok())) + .ok_or("size_limit must be a number")?; s.state.set_size_limit(&address, limit).await; Ok(Value::String("ok".into())) } diff --git a/crates/epix-runtime/src/lib.rs b/crates/epix-runtime/src/lib.rs index e3da1f7..f0eef5d 100644 --- a/crates/epix-runtime/src/lib.rs +++ b/crates/epix-runtime/src/lib.rs @@ -203,13 +203,24 @@ async fn connection_loop(state: Arc, shutdown: Arc, period: Du /// data. Collects once immediately, then every `period`. async fn chart_loop(state: Arc, shutdown: Arc, period: Duration) { state.collect_chart().await; + // Enforce retention once at startup, then roughly once a day (the collector + // runs far more often, so archive only every Nth tick). + state.archive_chart().await; + let archive_every = (Duration::from_secs(24 * 60 * 60).as_secs() / period.as_secs().max(1)).max(1); + let mut ticks: u64 = 0; let mut tick = interval(period); tick.set_missed_tick_behavior(MissedTickBehavior::Delay); tick.tick().await; // consume the immediate first tick loop { tokio::select! { _ = shutdown.notified() => break, - _ = tick.tick() => state.collect_chart().await, + _ = tick.tick() => { + state.collect_chart().await; + ticks += 1; + if ticks % archive_every == 0 { + state.archive_chart().await; + } + } } } } @@ -226,6 +237,11 @@ async fn resync_loop(state: Arc, shutdown: Arc, period: Durati for address in state.xite_addresses().await { let _ = state.resync_xite(&address).await; } + // OptionalManager: keep optional files under the size cap. + let freed = state.enforce_optional_limit().await; + if freed > 0 { + state.log("INFO", format!("Optional-file cleanup freed {freed} bytes")).await; + } } } } diff --git a/crates/epix-ui/src/chart.rs b/crates/epix-ui/src/chart.rs index 20a75c3..9de8f21 100644 --- a/crates/epix-ui/src/chart.rs +++ b/crates/epix-ui/src/chart.rs @@ -119,6 +119,23 @@ impl ChartDb { } } + /// Enforce retention so the chart db does not grow without bound: drop + /// per-site datapoints older than one month and global datapoints older than + /// six months, then reclaim the space. Mirrors EpixNet's `ChartDb.archive` + /// retention (without its downsampling of old rows). + pub fn archive(&self, now: i64) { + const MONTH: i64 = 60 * 60 * 24 * 30; + let _ = self.db.execute( + "DELETE FROM data WHERE site_id IS NOT NULL AND date_added < ?", + &[Value::from(now - MONTH)], + ); + let _ = self.db.execute( + "DELETE FROM data WHERE site_id IS NULL AND date_added < ?", + &[Value::from(now - 6 * MONTH)], + ); + let _ = self.db.execute_batch("VACUUM"); + } + /// Run a read-only chart query (the `chartDbQuery` command). Only SELECT is /// allowed, matching the Python action. `params` is bound by name (a /// list-valued param expands `IN :key` into a placeholder list), so the @@ -144,3 +161,31 @@ fn load_ids(db: &Database, sql: &str) -> HashMap { } out } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn archive_enforces_retention_windows() { + let db = ChartDb::memory().unwrap(); + let now: i64 = 2_000_000_000; + let day = 60 * 60 * 24; + let site = db.site_id("1Site"); + // Old global (7 months) + old per-site (2 months) should be dropped; + // recent points of each kind should stay. + db.record(now - day * 210, None, &[Metric::now("a", 1.0)]); + db.record(now - day * 60, site, &[Metric::now("a", 2.0)]); + db.record(now - day * 2, None, &[Metric::now("a", 3.0)]); + db.record(now - day * 2, site, &[Metric::now("a", 4.0)]); + assert_eq!(count(&db), 4); + + db.archive(now); + assert_eq!(count(&db), 2); + } + + fn count(db: &ChartDb) -> i64 { + db.query("SELECT COUNT(*) AS n FROM data", &json!({})).unwrap()[0]["n"].as_i64().unwrap() + } +} diff --git a/crates/epix-ui/src/command.rs b/crates/epix-ui/src/command.rs index 1f31753..3859370 100644 --- a/crates/epix-ui/src/command.rs +++ b/crates/epix-ui/src/command.rs @@ -141,6 +141,19 @@ impl WsSession { Ok((bound, inner_path.to_string())) } + /// Resolve an `inner_path` for a file/optional command to its real target + /// `(address, inner_path)`, applying both cross-origin routings: a + /// `cors-
/…` prefix (Cors permission) and a + /// `merged-/
/…` prefix (MergerSite). So fileGet, fileRules, + /// fileNeed, and optionalFileInfo all reach a merged/cors site the same way. + pub async fn resolve_target(&self, inner_path: &str) -> Result<(String, String), String> { + let (address, inner) = self.cors_target(inner_path).await?; + match AppState::split_merged_path(&inner) { + Some((addr, inner)) => Ok((addr, inner)), + None => Ok((address, inner)), + } + } + /// Whether this connection has joined `channel`. pub fn in_channel(&self, channel: &str) -> bool { self.channels.lock().unwrap().contains(channel) @@ -307,12 +320,19 @@ fn default_commands() -> Vec> { Arc::new(SiteRecoverPrivatekey), Arc::new(UserSetSitePrivatekey), Arc::new(SiteUpdate), - Arc::new(simple("sitePause", json!("ok"))), - Arc::new(simple("siteResume", json!("ok"))), - Arc::new(simple("siteDelete", json!("ok"))), - Arc::new(simple("siteSetAutodownloadoptional", json!("ok"))), - Arc::new(simple("dbReload", json!("ok"))), - Arc::new(simple("dbRebuild", json!("ok"))), + Arc::new(SiteServing { cmd: "sitePause", serving: false }), + Arc::new(SiteServing { cmd: "siteResume", serving: true }), + Arc::new(SiteDelete), + Arc::new(SiteSetAutodownloadoptional), + Arc::new(DbRebuild { cmd: "dbReload" }), + Arc::new(DbRebuild { cmd: "dbRebuild" }), + Arc::new(SiteFavourite { cmd: "siteFavourite", favorite: true }), + Arc::new(SiteFavourite { cmd: "siteUnfavourite", favorite: false }), + Arc::new(PeerAdd), + Arc::new(ServerShowdirectory), + Arc::new(XidClearCache), + Arc::new(NotificationDismiss), + Arc::new(SiteblockIgnoreAddSite), // CryptMessage Arc::new(UserPublickey), Arc::new(EciesEncrypt), @@ -424,13 +444,8 @@ impl WsCommand for FileGet { .as_str() .or_else(|| p.get("inner_path").and_then(|v| v.as_str())) .ok_or("fileGet: missing inner_path")?; - // `cors-
/` routes to another site (Cors permission). - let (address, inner_path) = s.cors_target(inner_path).await?; - // A `merged-/
/` file reads from the merged site. - let (target, inner) = match AppState::split_merged_path(&inner_path) { - Some((addr, inner)) => (addr, inner), - None => (address, inner_path), - }; + // Route `cors-…` (Cors permission) and `merged-…` (MergerSite) paths. + let (target, inner) = s.resolve_target(inner_path).await?; match s.state.read_file(&target, &inner).await { Some(bytes) => Ok(Value::from(String::from_utf8_lossy(&bytes).into_owned())), None => Ok(Value::Null), @@ -694,6 +709,17 @@ impl WsCommand for ServerInfo { } let connections = s.state.connection_stats().await.total; let plugins = s.state.plugins().await; + // Multiuser: when the feature is built, report the active identity so the + // wrapper UI shows the identity switcher. The desktop operator is admin. + #[cfg(feature = "multiuser")] + let (multiuser, multiuser_admin, master_address) = ( + true, + true, + s.state.multiuser_list().await.first().cloned().unwrap_or_default(), + ); + #[cfg(not(feature = "multiuser"))] + let (multiuser, multiuser_admin, master_address): (bool, bool, String) = + (false, false, String::new()); let language = s .state .config_get("language") @@ -717,9 +743,9 @@ impl WsCommand for ServerInfo { "ui_port": 43110, "debug": false, "offline": false, - "multiuser": false, - "multiuser_admin": false, - "master_address": "", + "multiuser": multiuser, + "multiuser_admin": multiuser_admin, + "master_address": master_address, "connections": connections, "timecorrection": 0.0, "lib_verify_best": "sslcrypto", @@ -926,7 +952,7 @@ impl WsCommand for FileRules { .or_else(|| p.get("inner_path").and_then(|v| v.as_str())) .or_else(|| p.as_array().and_then(|a| a.first()).and_then(|v| v.as_str())) .unwrap_or("content.json"); - let (address, inner_path) = s.cors_target(inner_path).await?; + let (address, inner_path) = s.resolve_target(inner_path).await?; Ok(s.state.file_rules(&address, &inner_path).await) } } @@ -1427,9 +1453,9 @@ impl WsCommand for FileNeed { "fileNeed" } async fn handle(&self, s: &WsSession, p: &Value) -> Result { - let address = s.address()?.to_string(); let inner_path = arg_str(p, "inner_path", 0).ok_or("fileNeed: inner_path required")?; - s.state.file_need(&address, inner_path).await?; + let (address, inner_path) = s.resolve_target(inner_path).await?; + s.state.file_need(&address, &inner_path).await?; Ok(Value::from("ok")) } } @@ -1456,9 +1482,9 @@ impl WsCommand for OptionalFileInfo { "optionalFileInfo" } async fn handle(&self, s: &WsSession, p: &Value) -> Result { - let address = s.address()?.to_string(); let inner_path = arg_str(p, "inner_path", 0).ok_or("optionalFileInfo: inner_path required")?; - s.state.optional_file_info(&address, inner_path).await + let (address, inner_path) = s.resolve_target(inner_path).await?; + s.state.optional_file_info(&address, &inner_path).await } } @@ -1601,8 +1627,27 @@ impl WsCommand for MergerSiteList { } } -/// `mergerSiteAdd(addresses)` - accept sites into this merger. (Cloning the -/// merged sites into the node is a follow-up; this validates the merger role.) +/// Collect one or more addresses from a param (string or array). +fn arg_addresses(p: &Value) -> Vec { + match p { + Value::String(a) => vec![a.clone()], + Value::Array(a) => a + .iter() + .filter_map(|v| v.as_str().map(str::to_string)) + .collect(), + _ => p + .get("addresses") + .and_then(|v| v.as_array()) + .map(|a| a.iter().filter_map(|v| v.as_str().map(str::to_string)).collect()) + .or_else(|| p.get("address").and_then(|v| v.as_str()).map(|s| vec![s.to_string()])) + .unwrap_or_default(), + } +} + +/// `mergerSiteAdd(addresses)` - accept sites into this merger. Links any of the +/// requested sites already present on the node into the merger database. +/// (Cloning a not-yet-present site from the network awaits the on-demand clone +/// flow.) struct MergerSiteAdd; #[async_trait] impl WsCommand for MergerSiteAdd { @@ -1614,22 +1659,28 @@ impl WsCommand for MergerSiteAdd { if s.state.merger_types(&address).await.is_empty() { return Err("Not a merger site".into()); } + // Re-link present mergeable sites into every merger's database. + s.state.rebuild_merger_dbs().await; Ok(Value::from("ok")) } } -/// `mergerSiteDelete(address)` - remove a merged site from this merger. +/// `mergerSiteDelete(address)` - remove a merged site from the node. struct MergerSiteDelete; #[async_trait] impl WsCommand for MergerSiteDelete { fn name(&self) -> &'static str { "mergerSiteDelete" } - async fn handle(&self, s: &WsSession, _p: &Value) -> Result { + async fn handle(&self, s: &WsSession, p: &Value) -> Result { let address = s.address()?.to_string(); if s.state.merger_types(&address).await.is_empty() { return Err("Not a merger site".into()); } + for target in arg_addresses(p) { + s.state.remove_xite(&target).await; + } + s.state.rebuild_merger_dbs().await; Ok(Value::from("ok")) } } @@ -1711,6 +1762,211 @@ impl WsCommand for UserLogout { } } +// ---- Site management actions (sidebar controls) ---------------------------- + +/// The target xite for a site action: an explicit `address` param, else the +/// bound xite. +fn target_address(s: &WsSession, p: &Value) -> Result { + match arg_str(p, "address", 0) { + Some(a) => Ok(a.to_string()), + None => Ok(s.address()?.to_string()), + } +} + +/// `sitePause`/`siteResume` - stop or resume re-syncing a xite. +struct SiteServing { + cmd: &'static str, + serving: bool, +} +#[async_trait] +impl WsCommand for SiteServing { + fn name(&self) -> &'static str { + self.cmd + } + async fn handle(&self, s: &WsSession, p: &Value) -> Result { + let address = target_address(s, p)?; + if s.state.set_serving(&address, self.serving).await { + Ok(Value::from("ok")) + } else { + Err(format!("Unknown site: {address}")) + } + } +} + +/// `siteDelete` - remove a xite from the node and delete its files. +struct SiteDelete; +#[async_trait] +impl WsCommand for SiteDelete { + fn name(&self) -> &'static str { + "siteDelete" + } + async fn handle(&self, s: &WsSession, p: &Value) -> Result { + let address = target_address(s, p)?; + if s.state.remove_xite(&address).await { + Ok(Value::from("ok")) + } else { + Err(format!("Unknown site: {address}")) + } + } +} + +/// `siteSetAutodownloadoptional` - toggle auto-downloading optional files. +struct SiteSetAutodownloadoptional; +#[async_trait] +impl WsCommand for SiteSetAutodownloadoptional { + fn name(&self) -> &'static str { + "siteSetAutodownloadoptional" + } + async fn handle(&self, s: &WsSession, p: &Value) -> Result { + let address = s.address()?.to_string(); + let on = p + .get("owned") + .or_else(|| p.as_array().and_then(|a| a.first())) + .and_then(|v| v.as_bool()) + .unwrap_or(false); + s.state.set_autodownloadoptional(&address, on).await; + Ok(Value::from("ok")) + } +} + +/// `dbReload`/`dbRebuild` - rebuild the xite's database from its files. +struct DbRebuild { + cmd: &'static str, +} +#[async_trait] +impl WsCommand for DbRebuild { + fn name(&self) -> &'static str { + self.cmd + } + async fn handle(&self, s: &WsSession, p: &Value) -> Result { + let address = target_address(s, p)?; + if s.state.rebuild_xite_db(&address).await { + Ok(Value::from("ok")) + } else { + Err(format!("Unknown site: {address}")) + } + } +} + +/// `siteFavourite`/`siteUnfavourite` - toggle the sidebar favourite star. +struct SiteFavourite { + cmd: &'static str, + favorite: bool, +} +#[async_trait] +impl WsCommand for SiteFavourite { + fn name(&self) -> &'static str { + self.cmd + } + async fn handle(&self, s: &WsSession, p: &Value) -> Result { + let address = target_address(s, p)?; + if s.state.set_favorite(&address, self.favorite).await { + Ok(Value::from("ok")) + } else { + Err(format!("Unknown site: {address}")) + } + } +} + +/// `peerAdd(ip, port, site_address?)` - add a peer to a xite's known set. +struct PeerAdd; +#[async_trait] +impl WsCommand for PeerAdd { + fn name(&self) -> &'static str { + "peerAdd" + } + async fn handle(&self, s: &WsSession, p: &Value) -> Result { + let ip = arg_str(p, "ip", 0).ok_or("ip required")?; + let port = p + .get("port") + .or_else(|| p.as_array().and_then(|a| a.get(1))) + .and_then(|v| v.as_i64().or_else(|| v.as_str().and_then(|s| s.parse().ok()))) + .ok_or("port required")?; + let address = match arg_str(p, "site_address", 2) { + Some(a) => a.to_string(), + None => s.address()?.to_string(), + }; + s.state.add_peer_ipport(&address, ip, port as u16).await?; + Ok(Value::from("updated")) + } +} + +/// `serverShowdirectory(directory, address?)` - open a xite's folder (or the +/// data dir) in the OS file manager. Local, desktop-oriented. +struct ServerShowdirectory; +#[async_trait] +impl WsCommand for ServerShowdirectory { + fn name(&self) -> &'static str { + "serverShowdirectory" + } + async fn handle(&self, s: &WsSession, p: &Value) -> Result { + let directory = arg_str(p, "directory", 0).unwrap_or("backup"); + let path = if directory == "backup" { + s.state.data_dir().ok_or("no data directory")? + } else { + let address = match arg_str(p, "address", 1) { + Some(a) => a.to_string(), + None => s.address()?.to_string(), + }; + s.state.xite_root(&address).await.ok_or_else(|| format!("Unknown site: {address}"))? + }; + open_path(&path); + Ok(Value::from("ok")) + } +} + +/// Open a filesystem path in the OS file manager (best effort). +fn open_path(path: &std::path::Path) { + #[cfg(target_os = "macos")] + let program = "open"; + #[cfg(target_os = "windows")] + let program = "explorer"; + #[cfg(all(not(target_os = "macos"), not(target_os = "windows")))] + let program = "xdg-open"; + let _ = std::process::Command::new(program).arg(path).spawn(); +} + +/// `xidClearCache` - clear the xID resolver cache. The node's resolver cache is +/// per-process at the chain layer; clearing it here is a successful no-op. +struct XidClearCache; +#[async_trait] +impl WsCommand for XidClearCache { + fn name(&self) -> &'static str { + "xidClearCache" + } + async fn handle(&self, _s: &WsSession, _p: &Value) -> Result { + Ok(Value::from("ok")) + } +} + +/// `notificationDismiss(center)` - remember a dismissed notification banner. +struct NotificationDismiss; +#[async_trait] +impl WsCommand for NotificationDismiss { + fn name(&self) -> &'static str { + "notificationDismiss" + } + async fn handle(&self, s: &WsSession, p: &Value) -> Result { + let center = arg_str(p, "center", 0).ok_or("center required")?; + s.state.notification_dismiss(center).await; + Ok(Value::from("ok")) + } +} + +/// `siteblockIgnoreAddSite(site_address)` - unblock a site so it can be added. +struct SiteblockIgnoreAddSite; +#[async_trait] +impl WsCommand for SiteblockIgnoreAddSite { + fn name(&self) -> &'static str { + "siteblockIgnoreAddSite" + } + async fn handle(&self, s: &WsSession, p: &Value) -> Result { + let address = arg_str(p, "site_address", 0).ok_or("site_address required")?; + s.state.siteblock_remove(address).await; + Ok(Value::from("ok")) + } +} + // ---- ContentFilter: mute + siteblock lists --------------------------------- /// Read a positional-or-named string arg from the command params. diff --git a/crates/epix-ui/src/lib.rs b/crates/epix-ui/src/lib.rs index 2846917..e4cf591 100644 --- a/crates/epix-ui/src/lib.rs +++ b/crates/epix-ui/src/lib.rs @@ -166,10 +166,36 @@ async fn serve_uimedia(State(ctx): State, Path(path): Path) -> Resp } /// Serve the wrapper page for a xite (`GET /{address}/`). +/// The page shown in place of a blocked site (ContentFilter). +fn blocklisted_html(address: &str, reason: &str) -> String { + let esc = |s: &str| s.replace('&', "&").replace('<', "<").replace('>', ">"); + let reason_html = if reason.is_empty() { + String::new() + } else { + format!("

Reason: {}

", esc(reason)) + }; + format!( + "Site blocked\ + \ +

This site is blocked

{}

{}", + esc(address), + reason_html, + ) +} + async fn serve_wrapper(State(ctx): State, Path(address): Path) -> Response { if !ctx.state.has_xite(&address).await { return (StatusCode::NOT_FOUND, "unknown xite").into_response(); } + // ContentFilter: a blocked site is not served - show the block page instead. + if let Some(reason) = ctx.state.siteblock_reason(&address).await { + return ( + StatusCode::FORBIDDEN, + [(header::CONTENT_TYPE, "text/html; charset=utf-8")], + blocklisted_html(&address, &reason), + ) + .into_response(); + } let content = ctx.state.content(&address).await; let title = content .as_ref() diff --git a/crates/epix-ui/src/state.rs b/crates/epix-ui/src/state.rs index 6eb5dd0..f6c3714 100644 --- a/crates/epix-ui/src/state.rs +++ b/crates/epix-ui/src/state.rs @@ -11,7 +11,7 @@ use epix_xite::{content_stats, Xite, XiteSettings, XiteStorage}; use serde_json::{json, Value}; use sha2::{Digest, Sha256}; use std::collections::HashMap; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; @@ -425,7 +425,15 @@ impl AppState { _ => continue, }; if let Ok(rows) = self.db_query(address, query, ¶ms).await { - let count = rows.len() as i64; + // A notification query is usually `SELECT COUNT(*) AS count`, + // so the real total is a column on the first row - not the + // number of rows returned. Fall back to the row count only + // when neither `count` nor `COUNT(*)` is present. + let count = rows + .first() + .and_then(|r| r.get("count").or_else(|| r.get("COUNT(*)"))) + .and_then(|v| v.as_i64()) + .unwrap_or(rows.len() as i64); if count > 0 { num += count; any = true; @@ -495,7 +503,8 @@ impl AppState { if let Some(content) = &entry.content { settings.apply_content_stats(&content_stats(content)); } - let (db, db_schema) = match build_xite_db(&entry.storage) { + let muted = self.muted_authors().await; + let (db, db_schema) = match build_xite_db(&entry.storage, &muted) { Some((db, schema)) => (Some(db), Some(schema)), None => (None, None), }; @@ -664,12 +673,13 @@ impl AppState { /// Replace a xite's content.json, refreshing its stats and rebuilding its db. pub async fn update_content(&self, address: &str, content: Option) { + let muted = self.muted_authors().await; if let Some(x) = self.xites.write().await.get_mut(address) { if let Some(c) = &content { x.settings.apply_content_stats(&content_stats(c)); } x.content = content; - let (db, schema) = match build_xite_db(&x.storage) { + let (db, schema) = match build_xite_db(&x.storage, &muted) { Some((db, schema)) => (Some(db), Some(schema)), None => (None, None), }; @@ -682,6 +692,10 @@ impl AppState { /// verify it and download the changed files (updating live worker stats). /// Returns true if an update was applied. This is the node's re-sync step. pub async fn resync_xite(&self, address: &str) -> Result { + // A paused xite (sitePause) is not re-synced until resumed. + if !self.is_serving(address).await { + return Ok(false); + } let transport = self.transport.read().await.clone().ok_or("no transport")?; let peers = self.connectable_peers(address, 10).await; if peers.is_empty() { @@ -815,6 +829,8 @@ impl AppState { json!({ "cert_user_id": cert_user_id, "reason": reason, "date_added": now_secs() }); } self.save_filters().await; + // Rebuild dbs so the muted author's content drops out (ContentFilter). + self.rebuild_all_dbs().await; } pub async fn mute_remove(&self, auth_address: &str) { @@ -822,6 +838,16 @@ impl AppState { m.remove(auth_address); } self.save_filters().await; + // Rebuild dbs so the un-muted author's content comes back. + self.rebuild_all_dbs().await; + } + + /// Rebuild every served xite's database (e.g. after a mute change), so the + /// mute filter is re-applied across all sites. + async fn rebuild_all_dbs(&self) { + for address in self.xite_addresses().await { + self.rebuild_xite_db(&address).await; + } } /// The mute map (`auth_address -> info`). @@ -855,6 +881,35 @@ impl AppState { self.filters.read().await["siteblocks"].get(site_address).cloned().unwrap_or(Value::Bool(false)) } + /// The block reason for a site, if it is blocked (`ContentFilter` enforcement + /// point). Checks both the plain address and its `sha256` hash, matching + /// EpixNet's hashed-address blocklists. + pub async fn siteblock_reason(&self, site_address: &str) -> Option { + let hashed = { + use sha2::{Digest, Sha256}; + hex::encode(Sha256::digest(site_address.as_bytes())) + }; + let f = self.filters.read().await; + let blocks = &f["siteblocks"]; + for key in [site_address, hashed.as_str()] { + if let Some(info) = blocks.get(key) { + return Some( + info.get("reason").and_then(|r| r.as_str()).unwrap_or("").to_string(), + ); + } + } + None + } + + /// The muted authors' auth-addresses (`ContentFilter` enforcement point). + /// Content signed by these is excluded when (re)building a xite's database. + pub async fn muted_authors(&self) -> Vec { + self.filters.read().await["mutes"] + .as_object() + .map(|m| m.keys().cloned().collect()) + .unwrap_or_default() + } + // --- OptionalManager ----------------------------------------------------- /// Reconstruct a `Xite` view (address + storage + content) for file ops. @@ -1131,6 +1186,67 @@ impl AppState { } } + /// Enforce the optional-files cap (`OptionalManager`): if downloaded optional + /// files exceed the limit, delete the oldest un-pinned ones until back under. + /// Returns the bytes freed. Called periodically by the runtime. + pub async fn enforce_optional_limit(&self) -> i64 { + let limit = self.optional_limit_bytes().await; + if limit <= 0 { + return 0; + } + // One scan of all downloaded optional files. "Downloaded" is judged by + // the on-disk file matching the declared size (cheaper than re-hashing). + // `used` counts every downloaded optional file; `candidates` are the + // un-pinned ones we may delete, oldest first. + let mut used: i64 = 0; + let mut candidates: Vec<(String, String, i64, u64)> = Vec::new(); + { + let xites = self.xites.read().await; + for (addr, x) in xites.iter() { + let Some(files_opt) = + x.content.as_ref().and_then(|c| c.get("files_optional")).and_then(|f| f.as_object()) + else { + continue; + }; + for (inner, meta) in files_opt { + let size = meta.get("size").and_then(|v| v.as_i64()).unwrap_or(0); + let Ok(path) = x.storage.path(inner) else { continue }; + let Ok(md) = std::fs::metadata(&path) else { continue }; + if md.len() as i64 != size { + continue; // not fully downloaded + } + used += size; + if x.pinned.contains(inner) { + continue; // pinned files are never evicted + } + let mtime = md + .modified() + .ok() + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| d.as_secs()) + .unwrap_or(0); + candidates.push((addr.clone(), inner.clone(), size, mtime)); + } + } + } + if used <= limit { + return 0; + } + // Oldest first. + candidates.sort_by_key(|c| c.3); + let mut freed = 0i64; + for (addr, inner, size, _) in candidates { + if used <= limit { + break; + } + if self.optional_file_delete(&addr, &inner).await.is_ok() { + used -= size; + freed += size; + } + } + freed + } + // --- MergerSite ---------------------------------------------------------- /// Grant a permission to a xite (e.g. `ADMIN`, `Merger:ZeroMe`). Idempotent. @@ -1447,6 +1563,12 @@ impl AppState { self.push_event("notification", json!([kind, message, timeout_ms]), None, None); } + /// Enforce chart-db retention (drop old datapoints, reclaim space). Called + /// periodically by the runtime so `chart.db` does not grow without bound. + pub async fn archive_chart(&self) { + self.chart.archive(now_secs()); + } + /// Snapshot current node metrics into the chart db: one global datapoint /// set plus a per-xite set. Called at startup and periodically by the /// runtime so the dashboard's Stats page has data to draw. @@ -1860,6 +1982,113 @@ impl AppState { } } + /// Pause/resume a xite (`sitePause`/`siteResume`). A paused xite is skipped + /// by the re-sync loop. Returns false if the xite isn't served here. + pub async fn set_serving(&self, address: &str, serving: bool) -> bool { + let ok = if let Some(x) = self.xites.write().await.get_mut(address) { + x.settings.serving = serving; + true + } else { + false + }; + if ok { + self.push_site_info(address).await; + } + ok + } + + /// Whether a xite is currently serving (not paused). + pub async fn is_serving(&self, address: &str) -> bool { + self.xites.read().await.get(address).map(|x| x.settings.serving).unwrap_or(false) + } + + /// Toggle a xite's favourite flag (`siteFavourite`/`siteUnfavourite`). + pub async fn set_favorite(&self, address: &str, favorite: bool) -> bool { + let ok = if let Some(x) = self.xites.write().await.get_mut(address) { + x.settings.favorite = favorite; + true + } else { + false + }; + if ok { + self.push_site_info(address).await; + } + ok + } + + /// Set a xite's auto-download-optional flag (`siteSetAutodownloadoptional`). + pub async fn set_autodownloadoptional(&self, address: &str, on: bool) -> bool { + if let Some(x) = self.xites.write().await.get_mut(address) { + x.settings.autodownloadoptional = on; + true + } else { + false + } + } + + /// Rebuild a xite's database from its files on disk (`dbReload`/`dbRebuild`). + /// Returns false if the xite isn't served here. + pub async fn rebuild_xite_db(&self, address: &str) -> bool { + let muted = self.muted_authors().await; + let mut xites = self.xites.write().await; + let Some(x) = xites.get_mut(address) else { return false }; + let (db, schema) = match build_xite_db(&x.storage, &muted) { + Some((db, schema)) => (Some(db), Some(schema)), + None => (None, None), + }; + x.db = db; + x.db_schema = schema; + true + } + + /// Remove a xite from the node and delete its files on disk (`siteDelete`). + /// Returns false if the xite isn't served here. + pub async fn remove_xite(&self, address: &str) -> bool { + let removed = self.xites.write().await.remove(address); + match removed { + Some(x) => { + // Best-effort delete of the xite's storage directory. + let root = x.storage.root().to_path_buf(); + let _ = std::fs::remove_dir_all(&root); + self.persist_peers().await; + true + } + None => false, + } + } + + /// Add a peer (ip + port) to a xite's known-peer set (`peerAdd`). + pub async fn add_peer_ipport(&self, address: &str, ip: &str, port: u16) -> Result<(), String> { + let peer = PeerAddr::parse(&format!("{ip}:{port}")).map_err(|e| e.to_string())?; + self.add_peers(address, [peer]).await; + Ok(()) + } + + /// A xite's storage directory on disk (`serverShowdirectory` "site"). + pub async fn xite_root(&self, address: &str) -> Option { + self.xites.read().await.get(address).map(|x| x.storage.root().to_path_buf()) + } + + /// The node's data directory (parent of `users.json`), if this is a + /// persistent node (`serverShowdirectory` "backup"). + pub fn data_dir(&self) -> Option { + self.user_path.as_ref().and_then(|p| p.parent().map(Path::to_path_buf)) + } + + /// Remember a dismissed notification center id (`notificationDismiss`), so a + /// dismissed banner stays dismissed across reloads. + pub async fn notification_dismiss(&self, center: &str) { + let mut list = self + .config_get("notification_dismissed") + .await + .and_then(|v| v.as_array().cloned()) + .unwrap_or_default(); + if !list.iter().any(|v| v.as_str() == Some(center)) { + list.push(json!(center)); + self.config_set("notification_dismissed", json!(list)).await; + } + } + /// Build the `siteInfo` response for a xite - EpixNet's `formatSiteInfo`. /// Returns `Null` if the xite isn't served here. pub async fn site_info(&self, address: &str) -> Value { @@ -2047,15 +2276,16 @@ impl AppState { /// Build a xite's database from its `dbschema.json` (if present): open an /// in-memory db, create the tables, and populate from the xite's JSON data /// files. `None` if the xite has no schema or building fails. -fn build_xite_db(storage: &XiteStorage) -> Option<(Database, DbSchema)> { +fn build_xite_db(storage: &XiteStorage, muted: &[String]) -> Option<(Database, DbSchema)> { let bytes = storage.read("dbschema.json").ok()?; let schema = DbSchema::from_json(&String::from_utf8_lossy(&bytes)).ok()?; let db = Database::open_in_memory().ok()?; db.apply_schema(&schema).ok()?; // A version-3 merger db is filled from its merged sites (rebuild_merger_dbs), - // not from its own files; everything else populates from its own tree. + // not from its own files; everything else populates from its own tree - + // skipping muted authors' data files (ContentFilter enforcement). if schema.version != 3 { - let _ = db.populate(&schema, storage.root()); + let _ = db.populate_filtered(&schema, storage.root(), muted); } Some((db, schema)) } @@ -2401,6 +2631,161 @@ mod tests { assert_eq!(one[0]["title"], "World"); } + #[tokio::test] + async fn optional_limit_evicts_oldest_unpinned_and_keeps_pinned() { + let dir = tempdir().unwrap(); + let storage = XiteStorage::new(dir.path()); + // Three 1000-byte optional files. + for name in ["a.bin", "b.bin", "c.bin"] { + storage.write(name, &vec![0u8; 1000]).unwrap(); + } + let addr = "epix1dashuu6pvsut7aw9dx44f543mv7xt9zlydsj9t"; + let content = json!({ + "address": addr, + "files": {}, + "files_optional": { + "a.bin": { "size": 1000, "sha512": "a" }, + "b.bin": { "size": 1000, "sha512": "b" }, + "c.bin": { "size": 1000, "sha512": "c" }, + } + }); + let state = AppState::new("test"); + state.add_xite(addr, XiteEntry { storage, content: Some(content) }).await; + state.set_pin(addr, "a.bin", true).await; + + // ~2791-byte cap: 3000 downloaded > cap, so eviction runs. + state.set_optional_limit("0.0000026").await; + let limit = state.optional_limit_bytes().await; + assert!(limit > 1000 && limit < 3000, "limit was {limit}"); + + let freed = state.enforce_optional_limit().await; + assert!(freed > 0, "expected some bytes freed"); + // The pinned file is never evicted. + assert!(dir.path().join("a.bin").exists()); + // Usage is back under the cap. + let remaining: i64 = ["a.bin", "b.bin", "c.bin"] + .iter() + .filter(|n| dir.path().join(n).exists()) + .count() as i64 + * 1000; + assert!(remaining <= limit, "remaining {remaining} > limit {limit}"); + } + + #[tokio::test] + async fn muting_an_author_drops_their_rows_from_the_db() { + let dir = tempdir().unwrap(); + let storage = XiteStorage::new(dir.path()); + storage + .write( + "dbschema.json", + br#"{ "db_name":"Blog","db_file":"db/db.db","version":2, + "maps": { "data/.*/data.json": { "to_table": [{"node":"posts","table":"post"}] } }, + "tables": { "post": { "cols": [["post_id","INTEGER"],["title","TEXT"],["json_id","INTEGER"]] } } }"#, + ) + .unwrap(); + storage.write("data/alice/data.json", br#"{ "posts": [ {"post_id":1,"title":"a"} ] }"#).unwrap(); + storage.write("data/mallory/data.json", br#"{ "posts": [ {"post_id":2,"title":"spam"} ] }"#).unwrap(); + + let addr = "1MuteBlog"; + let state = AppState::new("test"); + state.add_xite(addr, XiteEntry { storage, content: None }).await; + // Both authors' posts present initially. + assert_eq!(state.db_query(addr, "SELECT COUNT(*) AS n FROM post", &Value::Null).await.unwrap()[0]["n"], 2); + + // Muting mallory rebuilds the db without their data file. + state.mute_add("mallory", "mallory@cert", "spam").await; + let rows = state.db_query(addr, "SELECT title FROM post", &Value::Null).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0]["title"], "a"); + + // Un-muting brings it back. + state.mute_remove("mallory").await; + assert_eq!(state.db_query(addr, "SELECT COUNT(*) AS n FROM post", &Value::Null).await.unwrap()[0]["n"], 2); + } + + #[tokio::test] + async fn siteblock_reason_matches_plain_and_hashed_address() { + let state = AppState::new("test"); + assert!(state.siteblock_reason("1BadSite").await.is_none()); + state.siteblock_add("1BadSite", "malware").await; + assert_eq!(state.siteblock_reason("1BadSite").await.as_deref(), Some("malware")); + + // A hashed-address block also matches. + let hashed = { + use sha2::{Digest, Sha256}; + hex::encode(Sha256::digest("1HashBlocked".as_bytes())) + }; + state.siteblock_add(&hashed, "hashed").await; + assert_eq!(state.siteblock_reason("1HashBlocked").await.as_deref(), Some("hashed")); + } + + #[tokio::test] + async fn notification_count_reads_the_count_column_not_row_count() { + let dir = tempdir().unwrap(); + let storage = XiteStorage::new(dir.path()); + storage + .write( + "dbschema.json", + br#"{ "db_name":"Blog","db_file":"db/db.db","version":2, + "maps": { "data/.*/data.json": { "to_table": [{"node":"posts","table":"post"}] } }, + "tables": { "post": { "cols": [["post_id","INTEGER"],["title","TEXT"],["json_id","INTEGER"]] } } }"#, + ) + .unwrap(); + storage + .write( + "data/alice/data.json", + br#"{ "posts": [ {"post_id":1,"title":"a"}, {"post_id":2,"title":"b"}, {"post_id":3,"title":"c"} ] }"#, + ) + .unwrap(); + let addr = "1BlogAddr"; + let state = AppState::new("test"); + state.add_xite(addr, XiteEntry { storage, content: None }).await; + // A COUNT(*) subscription must report the column value (3), not 1 row. + state + .notification_subscribe(addr, json!({ "unread": ["SELECT COUNT(*) AS count FROM post", null] })) + .await; + let q = state.notification_query().await; + assert_eq!(q["num"], 3); + assert_eq!(q["results"][0]["count"], 3); + } + + #[tokio::test] + async fn pause_stops_resync_and_reflects_in_site_info() { + let content = json!({ "address": "1PauseMe", "files": {} }); + let state = AppState::new("test"); + let dir = tempdir().unwrap(); + state + .add_xite("1PauseMe", XiteEntry { storage: XiteStorage::new(dir.path()), content: Some(content) }) + .await; + assert!(state.is_serving("1PauseMe").await); + + assert!(state.set_serving("1PauseMe", false).await); + assert!(!state.is_serving("1PauseMe").await); + // A paused xite is not re-synced (returns Ok(false), even with no transport). + assert_eq!(state.resync_xite("1PauseMe").await, Ok(false)); + assert_eq!(state.site_info("1PauseMe").await["settings"]["serving"], false); + + assert!(state.set_serving("1PauseMe", true).await); + assert_eq!(state.site_info("1PauseMe").await["settings"]["serving"], true); + } + + #[tokio::test] + async fn favourite_and_delete_take_effect() { + let content = json!({ "address": "1FavMe", "files": {} }); + let state = AppState::new("test"); + let dir = tempdir().unwrap(); + state + .add_xite("1FavMe", XiteEntry { storage: XiteStorage::new(dir.path()), content: Some(content) }) + .await; + assert!(state.set_favorite("1FavMe", true).await); + assert_eq!(state.site_info("1FavMe").await["settings"]["favorite"], true); + + // Delete removes the xite; a second delete reports not-found. + assert!(state.remove_xite("1FavMe").await); + assert!(!state.has_xite("1FavMe").await); + assert!(!state.remove_xite("1FavMe").await); + } + #[tokio::test] async fn file_write_then_site_sign_produces_owned_signed_content() { // A key that owns the xite (address == the xite address). diff --git a/crates/epix-xite/src/settings.rs b/crates/epix-xite/src/settings.rs index 53a9e8b..957589d 100644 --- a/crates/epix-xite/src/settings.rs +++ b/crates/epix-xite/src/settings.rs @@ -49,6 +49,9 @@ pub struct XiteSettings { pub size_limit: Option, #[serde(default)] pub autodownloadoptional: bool, + /// Whether the user favourited this xite (sidebar star). + #[serde(default)] + pub favorite: bool, /// Random key authorizing this xite's WebSocket (part of the wrapper URL). pub wrapper_key: String, /// Random key authorizing AJAX/media requests. @@ -74,6 +77,7 @@ impl XiteSettings { peers: 0, size_limit: None, autodownloadoptional: false, + favorite: false, wrapper_key: epix_crypt::new_seed(), ajax_key: epix_crypt::new_seed(), cache: Cache::default(),