You are a constitutional council ranking individual git commits for ownership allocation. Compare these two commits. Decide which contributed more lasting value to the project. Judge substance, not spectacle: - Prefer correct, lasting design and real bugfixes over churn, formatting, renames, or generated noise. - Prefer clarity and necessity over sheer line count. A small precise change can beat a large diffuse one. - Do not favor a side merely because its patch is longer or noisier. - Weight what the change does for the project, not the contributor's name. Return ONLY a JSON object: {"winner": "A" or "B", "ratio": "N:M", "explanation": "..."} The explanation must cite concrete differences in the patches (1-3 sentences). Side A — contributor: tommy-mor Side A — commit message: [239c074b] url schema stuff Side A — unified diff (full patch): diff --git a/AGENTS.md b/AGENTS.md index 426a88e7c1da54fe0a28c5c76fa4e1f1bc117fcf..e60b9ba6012593361ef10e8fdd9439cd9932e09b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -58,3 +58,4 @@ Use **tmux** for `cargo run --package sorter2-server` (dev server). Rebuild afte - First `cargo test` / `cargo build --release` is slow; Clojure smoke test always does a release build. - `legacy/` and `ideas/` are not part of the workspace build. +- **ItemId** for web URLs is a canonical full URL (`https://reddit.com/r/rust`). Rules live in [`server/src/url_rules/`](server/src/url_rules/) (composable Rust, not a config DSL). After changing canonicalization rules, rebuild the projection: `cargo run --package sorter2-server -- replay-index`. diff --git a/Cargo.lock b/Cargo.lock index 0dd4fce5fb6400ae153cca4e3dbf5a5158e6d8b4..49a908ef935c430dbe63c6a28d8a24e38b489486 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1951,6 +1951,7 @@ dependencies = [ "tower-http 0.5.2", "tracing", "tracing-subscriber", + "url", "urlencoding", ] diff --git a/REPLAY.sh b/REPLAY.sh new file mode 100755 index 0000000000000000000000000000000000000000..f2dbd8aea60c02d2feef74805f7ef5c2b7022537 --- /dev/null +++ b/REPLAY.sh @@ -0,0 +1,2 @@ +cargo run --package sorter2-server -- replay-index + diff --git a/server/Cargo.toml b/server/Cargo.toml index 27f552c20b97ef28cdde4cb6b1a4980375135111..ad4912791aff59fb1d3293f66ad381ae618cd60b 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -24,6 +24,7 @@ async-stream = "0.3" futures-util = { version = "0.3", default-features = false, features = ["std"] } rand = "0.8" urlencoding = "2" +url = "2" durable = { path = "../durable" } [dev-dependencies] diff --git a/server/src/entity_store.rs b/server/src/entity_store.rs index d5d17c3676e4a8ddec998e9f5a9dbafe9c2d9d0e..d29f39aecca6f12cdcf263cf77c3654eb4ee6cfa 100644 --- a/server/src/entity_store.rs +++ b/server/src/entity_store.rs @@ -124,7 +124,7 @@ mod tests { fn round_trip_payload() { let tmp = tempfile::tempdir().unwrap(); let store = EntityStore::open(tmp.path()).unwrap(); - let id = ItemId::parse("reddit.com/r/rust").unwrap(); + let id = ItemId::from_url("https://reddit.com/r/rust").unwrap(); let payload = json!({"kind": "t5", "data": {"display_name": "rust"}}); store.put(&id, &payload).unwrap(); diff --git a/server/src/event_log.rs b/server/src/event_log.rs index 36f5b406084065b608735987cdb483c236e03081..2c9290b6fdbf2c2ad1c0f1ffd7374b2d9cc97f36 100644 --- a/server/src/event_log.rs +++ b/server/src/event_log.rs @@ -199,7 +199,7 @@ mod tests { log.append(&sample_record( 1, Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }, )) .await @@ -237,7 +237,7 @@ mod tests { let path = tmp.path().join("events.jsonl"); let log = EventLog::new(&path); let event = Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }; log.append(&sample_record(1, event)).await.unwrap(); @@ -255,7 +255,7 @@ mod tests { let path = tmp.path().join("events.jsonl"); std::fs::write( &path, - r#"{"type":"node_ensured","id":"reddit.com/r/rust"} + r#"{"type":"node_ensured","id":"https://reddit.com/r/rust"} {"schema":1,"seq":1,"ts":1,"event":{"type":"vote_recorded","ts":1,"a":"a","b":"b","ratio_left":2,"ratio_right":1,"scope":""}} "#, ) @@ -295,7 +295,7 @@ mod tests { log.append(&sample_record( 1, Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }, )) .await @@ -303,7 +303,7 @@ mod tests { log.append(&sample_record( 3, Event::NodeEnsured { - id: "reddit.com/r/python".into(), + id: "https://reddit.com/r/python".into(), }, )) .await diff --git a/server/src/journal.rs b/server/src/journal.rs index 521a108019de1ea870d14c4fafbfe572c20ce0de..50bc89f976edb82b7b0e49e954a8eccbbe82bf87 100644 --- a/server/src/journal.rs +++ b/server/src/journal.rs @@ -141,10 +141,10 @@ mod tests { let j2 = journal.clone(); let (r1, r2) = tokio::join!( j1.append(Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }), j2.append(Event::NodeEnsured { - id: "reddit.com/r/python".into(), + id: "https://reddit.com/r/python".into(), }), ); r1.unwrap(); @@ -153,10 +153,10 @@ mod tests { assert_eq!(projection_store.last_applied_event_count().unwrap(), 2); let tree = projection_store.load_tree().unwrap(); assert!(tree - .get(&ItemId::parse("reddit.com/r/rust").unwrap()) + .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) .is_some()); assert!(tree - .get(&ItemId::parse("reddit.com/r/python").unwrap()) + .get(&ItemId::parse("https://reddit.com/r/python").unwrap()) .is_some()); } @@ -170,7 +170,7 @@ mod tests { 1, 1, Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }, )) .await @@ -186,7 +186,7 @@ mod tests { 1, 1, Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }, )], ) @@ -202,7 +202,7 @@ mod tests { ); journal .append(Event::NodeEnsured { - id: "reddit.com/r/python".into(), + id: "https://reddit.com/r/python".into(), }) .await .unwrap(); @@ -227,13 +227,13 @@ mod tests { journal .append_many(vec![ Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }, Event::NodeEnsured { - id: "reddit.com/r/python".into(), + id: "https://reddit.com/r/python".into(), }, Event::NodeEnsured { - id: "reddit.com/r/clojure".into(), + id: "https://reddit.com/r/clojure".into(), }, ]) .await @@ -245,7 +245,7 @@ mod tests { assert_eq!(projection_store.last_applied_event_count().unwrap(), 3); let tree = projection_store.load_tree().unwrap(); assert!(tree - .get(&ItemId::parse("reddit.com/r/clojure").unwrap()) + .get(&ItemId::parse("https://reddit.com/r/clojure").unwrap()) .is_some()); } } diff --git a/server/src/lib.rs b/server/src/lib.rs index 9bd5f76fd1406b9b1be4c272f4ba8647edde2678..5c02c8e704e4664453bad75d819df8a067668176 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -9,6 +9,7 @@ pub mod journal; pub mod pair; pub mod parser; pub mod path_types; +pub mod url_rules; pub mod projection_apply; pub mod projection_store; pub mod ranking; diff --git a/server/src/pair.rs b/server/src/pair.rs index 43f780ba6ea6ce1cdc2e1f4cbb252ba8a10684b9..815a97b80e3e9f348e0937a4f147f2862018edb0 100644 --- a/server/src/pair.rs +++ b/server/src/pair.rs @@ -381,42 +381,42 @@ mod tests { #[test] fn suggest_prefers_unvoted_pair() { - let parent = ItemId::parse("reddit.com/r/rust").unwrap(); + let parent = ItemId::parse("https://reddit.com/r/rust").unwrap(); let mut tree = seed_children( &parent, &[ - "reddit.com/r/rust/a", - "reddit.com/r/rust/b", - "reddit.com/r/rust/c", + "https://reddit.com/r/rust/a", + "https://reddit.com/r/rust/b", + "https://reddit.com/r/rust/c", ], ); let vote = - VoteData::from_recorded(1, "reddit.com/r/rust/a", "reddit.com/r/rust/b", 2, 1).unwrap(); + VoteData::from_recorded(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1).unwrap(); tree.apply_vote(&parent, vote); let group = tree.get(&parent).unwrap().local_ranking.clone(); let pool = children_of(&tree, &parent); let (l, r) = suggest_next_pair_in_pool(&group, &pool, None).unwrap(); - let voted_ab = (l.as_str() == "reddit.com/r/rust/a" && r.as_str() == "reddit.com/r/rust/b") - || (l.as_str() == "reddit.com/r/rust/b" && r.as_str() == "reddit.com/r/rust/a"); + let voted_ab = (l.as_str() == "https://reddit.com/r/rust/a" && r.as_str() == "https://reddit.com/r/rust/b") + || (l.as_str() == "https://reddit.com/r/rust/b" && r.as_str() == "https://reddit.com/r/rust/a"); assert!(!voted_ab); } #[test] fn suggest_bridges_separate_components() { - let parent = ItemId::parse("reddit.com/r/rust").unwrap(); + let parent = ItemId::parse("https://reddit.com/r/rust").unwrap(); let mut tree = seed_children( &parent, &[ - "reddit.com/r/rust/a", - "reddit.com/r/rust/b", - "reddit.com/r/rust/c", - "reddit.com/r/rust/d", + "https://reddit.com/r/rust/a", + "https://reddit.com/r/rust/b", + "https://reddit.com/r/rust/c", + "https://reddit.com/r/rust/d", ], ); let ab = - VoteData::from_recorded(1, "reddit.com/r/rust/a", "reddit.com/r/rust/b", 2, 1).unwrap(); + VoteData::from_recorded(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1).unwrap(); let cd = - VoteData::from_recorded(2, "reddit.com/r/rust/c", "reddit.com/r/rust/d", 2, 1).unwrap(); + VoteData::from_recorded(2, "https://reddit.com/r/rust/c", "https://reddit.com/r/rust/d", 2, 1).unwrap(); tree.apply_vote(&parent, ab); tree.apply_vote(&parent, cd); let group = tree.get(&parent).unwrap().local_ranking.clone(); @@ -424,37 +424,37 @@ mod tests { let pair = suggest_next_pair_in_pool(&group, &pool, None).unwrap(); let chosen = pair_set(&pair); let from_ab = - chosen.contains("reddit.com/r/rust/a") || chosen.contains("reddit.com/r/rust/b"); + chosen.contains("https://reddit.com/r/rust/a") || chosen.contains("https://reddit.com/r/rust/b"); let from_cd = - chosen.contains("reddit.com/r/rust/c") || chosen.contains("reddit.com/r/rust/d"); + chosen.contains("https://reddit.com/r/rust/c") || chosen.contains("https://reddit.com/r/rust/d"); assert!(from_ab && from_cd, "expected bridge pair, got {:?}", chosen); } #[test] fn suggest_prefers_attach_over_isolate_pair_among_many_unranked() { - let parent = ItemId::parse("reddit.com/r/rust").unwrap(); + let parent = ItemId::parse("https://reddit.com/r/rust").unwrap(); let mut tree = seed_children( &parent, &[ - "reddit.com/r/rust/a", - "reddit.com/r/rust/b", - "reddit.com/r/rust/c", - "reddit.com/r/rust/d", - "reddit.com/r/rust/e", + "https://reddit.com/r/rust/a", + "https://reddit.com/r/rust/b", + "https://reddit.com/r/rust/c", + "https://reddit.com/r/rust/d", + "https://reddit.com/r/rust/e", ], ); let ab = - VoteData::from_recorded(1, "reddit.com/r/rust/a", "reddit.com/r/rust/b", 2, 1).unwrap(); + VoteData::from_recorded(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1).unwrap(); tree.apply_vote(&parent, ab); let group = tree.get(&parent).unwrap().local_ranking.clone(); let pool = children_of(&tree, &parent); let pair = suggest_next_pair_in_pool(&group, &pool, None).unwrap(); let chosen = pair_set(&pair); let from_ab = - chosen.contains("reddit.com/r/rust/a") || chosen.contains("reddit.com/r/rust/b"); - let from_cde = chosen.contains("reddit.com/r/rust/c") - || chosen.contains("reddit.com/r/rust/d") - || chosen.contains("reddit.com/r/rust/e"); + chosen.contains("https://reddit.com/r/rust/a") || chosen.contains("https://reddit.com/r/rust/b"); + let from_cde = chosen.contains("https://reddit.com/r/rust/c") + || chosen.contains("https://reddit.com/r/rust/d") + || chosen.contains("https://reddit.com/r/rust/e"); assert!( from_ab && from_cde, "expected ranked+unranked attach, got {:?}", @@ -464,40 +464,40 @@ mod tests { #[test] fn suggest_connects_isolate_to_existing_component() { - let parent = ItemId::parse("reddit.com/r/rust").unwrap(); + let parent = ItemId::parse("https://reddit.com/r/rust").unwrap(); let mut tree = seed_children( &parent, &[ - "reddit.com/r/rust/a", - "reddit.com/r/rust/b", - "reddit.com/r/rust/c", + "https://reddit.com/r/rust/a", + "https://reddit.com/r/rust/b", + "https://reddit.com/r/rust/c", ], ); let ab = - VoteData::from_recorded(1, "reddit.com/r/rust/a", "reddit.com/r/rust/b", 2, 1).unwrap(); + VoteData::from_recorded(1, "https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 2, 1).unwrap(); tree.apply_vote(&parent, ab); let group = tree.get(&parent).unwrap().local_ranking.clone(); let pool = children_of(&tree, &parent); let pair = suggest_next_pair_in_pool(&group, &pool, None).unwrap(); let chosen = pair_set(&pair); - assert!(chosen.contains("reddit.com/r/rust/c")); - assert!(chosen.contains("reddit.com/r/rust/a") || chosen.contains("reddit.com/r/rust/b")); + assert!(chosen.contains("https://reddit.com/r/rust/c")); + assert!(chosen.contains("https://reddit.com/r/rust/a") || chosen.contains("https://reddit.com/r/rust/b")); } #[test] fn suggest_zips_adjacent_ranks_when_tree_complete() { - let parent = ItemId::parse("reddit.com/r/rust").unwrap(); + let parent = ItemId::parse("https://reddit.com/r/rust").unwrap(); let mut tree = seed_children( &parent, &[ - "reddit.com/r/rust/a", - "reddit.com/r/rust/b", - "reddit.com/r/rust/c", + "https://reddit.com/r/rust/a", + "https://reddit.com/r/rust/b", + "https://reddit.com/r/rust/c", ], ); for (a, b, l, r) in [ - ("reddit.com/r/rust/a", "reddit.com/r/rust/b", 3, 1), - ("reddit.com/r/rust/a", "reddit.com/r/rust/c", 2, 1), + ("https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b", 3, 1), + ("https://reddit.com/r/rust/a", "https://reddit.com/r/rust/c", 2, 1), ] { let v = VoteData::from_recorded(1, a, b, l, r).unwrap(); tree.apply_vote(&parent, v); @@ -506,26 +506,26 @@ mod tests { let pool = children_of(&tree, &parent); let pair = suggest_next_pair_in_pool(&group, &pool, None).unwrap(); let chosen = pair_set(&pair); - assert!(chosen.contains("reddit.com/r/rust/b")); - assert!(chosen.contains("reddit.com/r/rust/c")); + assert!(chosen.contains("https://reddit.com/r/rust/b")); + assert!(chosen.contains("https://reddit.com/r/rust/c")); } #[test] fn suggest_zip_prefers_1v2_before_2v3_when_both_unvoted() { - let parent = ItemId::parse("reddit.com/r/rust").unwrap(); + let parent = ItemId::parse("https://reddit.com/r/rust").unwrap(); let mut tree = seed_children( &parent, &[ - "reddit.com/r/rust/a", - "reddit.com/r/rust/b", - "reddit.com/r/rust/c", - "reddit.com/r/rust/d", + "https://reddit.com/r/rust/a", + "https://reddit.com/r/rust/b", + "https://reddit.com/r/rust/c", + "https://reddit.com/r/rust/d", ], ); for (a, b, l, r) in [ - ("reddit.com/r/rust/c", "reddit.com/r/rust/d", 3, 1), - ("reddit.com/r/rust/b", "reddit.com/r/rust/c", 2, 1), - ("reddit.com/r/rust/a", "reddit.com/r/rust/c", 2, 1), + ("https://reddit.com/r/rust/c", "https://reddit.com/r/rust/d", 3, 1), + ("https://reddit.com/r/rust/b", "https://reddit.com/r/rust/c", 2, 1), + ("https://reddit.com/r/rust/a", "https://reddit.com/r/rust/c", 2, 1), ] { let v = VoteData::from_recorded(1, a, b, l, r).unwrap(); tree.apply_vote(&parent, v); @@ -534,16 +534,16 @@ mod tests { let pool = children_of(&tree, &parent); let pair = suggest_next_pair_in_pool(&group, &pool, None).unwrap(); let chosen = pair_set(&pair); - assert!(chosen.contains("reddit.com/r/rust/a")); - assert!(chosen.contains("reddit.com/r/rust/b")); + assert!(chosen.contains("https://reddit.com/r/rust/a")); + assert!(chosen.contains("https://reddit.com/r/rust/b")); } #[test] fn resolve_pair_picks_from_pool() { - let parent = ItemId::parse("reddit.com/r/rust").unwrap(); - let tree = seed_children(&parent, &["reddit.com/r/rust/a", "reddit.com/r/rust/b"]); + let parent = ItemId::parse("https://reddit.com/r/rust").unwrap(); + let tree = seed_children(&parent, &["https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b"]); let pair = resolve_pair(&tree, &parent, None, None).unwrap(); - let pool: HashSet<_> = ["reddit.com/r/rust/a", "reddit.com/r/rust/b"] + let pool: HashSet<_> = ["https://reddit.com/r/rust/a", "https://reddit.com/r/rust/b"] .into_iter() .collect(); assert!(pool.contains(pair.0.as_str())); diff --git a/server/src/parser.rs b/server/src/parser.rs index 9df2dcc9313f7fe250ce3b6aa167f6ec5d57951f..b2a963dd6cab415576c8d3a9a241966758565bb8 100644 --- a/server/src/parser.rs +++ b/server/src/parser.rs @@ -23,7 +23,7 @@ mod tests { fn parses_short_path() { assert_eq!( parse_reddit_url("r/rust").unwrap().as_str(), - "reddit.com/r/rust" + "https://reddit.com/r/rust" ); } @@ -33,7 +33,7 @@ mod tests { parse_reddit_url("https://www.reddit.com/r/programming/hot") .unwrap() .as_str(), - "reddit.com/r/programming" + "https://reddit.com/r/programming" ); } @@ -43,7 +43,10 @@ mod tests { "https://old.reddit.com/r/AmItheAsshole/comments/1trnvdl/aita_for_cancelling/", ) .unwrap(); - assert_eq!(id.as_str(), "reddit.com/r/amitheasshole/comments/1trnvdl"); + assert_eq!( + id.as_str(), + "https://reddit.com/r/amitheasshole/comments/1trnvdl" + ); } #[test] diff --git a/server/src/path_types.rs b/server/src/path_types.rs index fafd924452f6fd85e7a5b27ed2653a19581e6e56..71ffc01f35c5589686ff05dae3b5610fa01f30ce 100644 --- a/server/src/path_types.rs +++ b/server/src/path_types.rs @@ -1,13 +1,14 @@ use serde::{Deserialize, Serialize}; use std::fmt; -/// Canonical hierarchical identity for any URL/path in the fractal tree. +use crate::url_rules::{looks_like_url, navigable_breadcrumbs, parent_url, resolve_canonical}; + +/// Canonical identity: a real URL (with scheme) or an opaque non-URL key. #[derive(Debug, Clone, Hash, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, Default)] pub struct ItemId(String); impl ItemId { - /// Parse an already-canonical path (no URL normalization). Empty string is invalid here; - /// use [`Self::root`] for the tree root. + /// Parse an already-canonical id (no normalization). Empty string is invalid; use [`Self::root`]. pub fn parse(s: &str) -> Option { let t = s.trim(); if t.is_empty() { @@ -16,12 +17,12 @@ impl ItemId { Some(Self(t.to_string())) } - /// Build an opaque item key (legacy demo votes, non-URL items). + /// Build an opaque item key (demo votes, non-URL items). pub fn opaque(s: impl Into) -> Self { Self(s.into()) } - /// Root of the internet tree (empty path). + /// Root of the internet tree. pub fn root() -> Self { Self(String::new()) } @@ -34,23 +35,18 @@ impl ItemId { &self.0 } - /// Creates a canonical ID from a raw URL or path. Normalizes domains and - /// trims tracking query params. + /// Canonical URL from a raw pasted or fetched URL. pub fn from_url(raw_url: &str) -> Option { - Self::canonicalize(raw_url).map(Self) + resolve_canonical(raw_url).map(Self) } - /// Normalize strings from forms, events, and Reddit imports into the same - /// stored id shape (e.g. drop post title slug after comment id). + /// Normalize strings from forms, events, and imports into canonical identity. pub fn from_storage(s: &str) -> Option { let t = s.trim(); if t.is_empty() { return None; } - if t.contains("://") || t.starts_with("r/") { - return Self::from_url(t).or_else(|| Self::parse(t)); - } - if t.starts_with("reddit.com/") && t.contains("/comments/") { + if looks_like_url(t) { return Self::from_url(t).or_else(|| Self::parse(t)); } Self::parse(t).or_else(|| Self::from_url(t)) @@ -62,35 +58,52 @@ impl ItemId { if s.is_empty() { return Self::root(); } - Self(format!("reddit.com/r/{s}")) + if looks_like_url(s) || s.contains('/') { + Self::from_storage(s).unwrap_or_else(|| Self::opaque(s)) + } else { + Self(format!("https://reddit.com/r/{s}")) + } } - /// Extract the parent, e.g. `reddit.com/r/aww/comments/1trnvdl` → - /// `reddit.com/r/aww`. + /// Immediate parent scope in the tree. pub fn parent(&self) -> Option { - if self.0.is_empty() { + if self.is_root() { return None; } - + if looks_like_url(self.0.as_str()) { + return parent_url(self.0.as_str()).map(Self); + } let parts: Vec<&str> = self.0.trim_end_matches('/').split('/').collect(); if parts.len() <= 1 { return None; } - - if self.0.contains("/comments/") { - return Some(Self(parts[..parts.len().saturating_sub(2)].join("/"))); - } - Some(Self(parts[..parts.len() - 1].join("/"))) } pub fn segments(&self) -> Vec<&str> { + if self.is_root() { + return vec![]; + } + if let Some(rest) = self.0.strip_prefix("https://") { + return rest.split('/').filter(|s| !s.is_empty()).collect(); + } + if let Some(rest) = self.0.strip_prefix("http://") { + return rest.split('/').filter(|s| !s.is_empty()).collect(); + } self.0.split('/').filter(|s| !s.is_empty()).collect() } - /// Cumulative paths for breadcrumb rendering, e.g. - /// `reddit.com/r/movies` → `["reddit.com", "reddit.com/r", "reddit.com/r/movies"]`. + /// Cumulative navigable paths for breadcrumbs and tree wiring (includes self). pub fn breadcrumb_paths(&self) -> Vec { + if self.is_root() { + return vec![]; + } + if looks_like_url(self.0.as_str()) { + return navigable_breadcrumbs(self.0.as_str()) + .into_iter() + .map(ItemId) + .collect(); + } let segs = self.segments(); let mut paths = Vec::with_capacity(segs.len()); let mut current = String::new(); @@ -111,13 +124,13 @@ impl ItemId { if self.is_root() { return String::new(); } - if self.as_str().contains("://") { - return self.as_str().to_string(); + if self.0.contains("://") { + return self.0.clone(); } if self.segments().first().is_some_and(|s| s.contains('.')) { - format!("https://{}", self.as_str()) + format!("https://{}", self.0) } else { - self.as_str().to_string() + self.0.clone() } } @@ -144,70 +157,6 @@ impl ItemId { pub fn from_browse_uri(path: &str) -> Option { path.strip_prefix("/~/").map(ItemId::from_browse_tail) } - - fn canonicalize(raw: &str) -> Option { - let s = raw.trim(); - if s.is_empty() { - return None; - } - - let owned = if let Some(rest) = s.strip_prefix("r/") { - format!("reddit.com/r/{rest}") - } else if let Some(rest) = s.strip_prefix("/r/") { - format!("reddit.com/r/{rest}") - } else { - s.to_string() - }; - - let (host_path, _query) = split_query(&owned); - let host_path = host_path.trim_end_matches('/'); - - let path = if host_path.contains("://") { - parse_url_host_path(host_path)? - } else if host_path.starts_with("reddit.com") || host_path.starts_with("www.reddit.com") { - normalize_reddit_host_path(host_path) - } else if host_path.contains('/') { - host_path.to_string() - } else { - return None; - }; - - Some(normalize_reddit_path(&path)) - } -} - -fn split_query(s: &str) -> (&str, Option<&str>) { - if let Some((path, q)) = s.split_once('?') { - (path, Some(q)) - } else { - (s, None) - } -} - -fn parse_url_host_path(url: &str) -> Option { - let rest = url - .strip_prefix("https://") - .or_else(|| url.strip_prefix("http://")) - .unwrap_or(url); - let (host, path) = rest.split_once('/').unwrap_or((rest, "")); - let host = normalize_host(host); - if path.is_empty() { - Some(host) - } else { - Some(format!("{host}/{path}")) - } -} - -fn normalize_host(host: &str) -> String { - let h = host - .strip_prefix("www.") - .unwrap_or(host) - .to_ascii_lowercase(); - if h == "old.reddit.com" || h == "new.reddit.com" || h == "reddit.com" { - "reddit.com".to_string() - } else { - h - } } fn normalize_browse_tail(tail: &str) -> String { @@ -215,7 +164,6 @@ fn normalize_browse_tail(tail: &str) -> String { if t.is_empty() { return String::new(); } - // Some HTTP stacks collapse `https://` → `https:/` inside a path segment. if t.starts_with("https:/") && !t.starts_with("https://") { return format!("https://{}", &t[7..]); } @@ -225,33 +173,6 @@ fn normalize_browse_tail(tail: &str) -> String { t.to_string() } -fn normalize_reddit_host_path(s: &str) -> String { - let (host, path) = s.split_once('/').unwrap_or((s, "")); - let host = normalize_host(host); - if path.is_empty() { - host - } else { - format!("{host}/{path}") - } -} - -/// Lowercase subreddit segment, drop listing suffixes, drop title slug after post id. -fn normalize_reddit_path(path: &str) -> String { - let mut parts: Vec = path.split('/').map(str::to_string).collect(); - if parts.len() >= 3 && parts[1] == "r" { - parts[2] = parts[2].to_ascii_lowercase(); - } - if let Some(i) = parts.iter().position(|p| p == "comments") { - if parts.len() > i + 2 { - parts.truncate(i + 2); - } - } else if parts.len() > 3 && parts.get(1).map(|s| s.as_str()) == Some("r") { - // reddit.com/r/{sub}/hot → reddit.com/r/{sub} - parts.truncate(3); - } - parts.join("/") -} - impl fmt::Display for ItemId { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.write_str(&self.0) @@ -268,43 +189,59 @@ mod tests { "https://old.reddit.com/r/AmItheAsshole/comments/1trnvdl/aita_for_cancelling/", ) .unwrap(); - assert_eq!(id.as_str(), "reddit.com/r/amitheasshole/comments/1trnvdl"); + assert_eq!( + id.as_str(), + "https://reddit.com/r/amitheasshole/comments/1trnvdl" + ); } #[test] fn from_url_strips_query() { let id = ItemId::from_url("https://www.reddit.com/r/rust/?sort=top").unwrap(); - assert_eq!(id.as_str(), "reddit.com/r/rust"); + assert_eq!(id.as_str(), "https://reddit.com/r/rust"); } #[test] fn from_url_short_path() { assert_eq!( ItemId::from_url("r/rust").unwrap().as_str(), - "reddit.com/r/rust" + "https://reddit.com/r/rust" ); } #[test] fn parent_of_post_is_subreddit() { - let id = ItemId::parse("reddit.com/r/aww/comments/1trnvdl").unwrap(); - assert_eq!(id.parent().unwrap().as_str(), "reddit.com/r/aww"); + let id = ItemId::from_url("https://reddit.com/r/aww/comments/1trnvdl").unwrap(); + assert_eq!(id.parent().unwrap().as_str(), "https://reddit.com/r/aww"); } #[test] fn parent_of_subreddit_is_r_segment() { - let id = ItemId::parse("reddit.com/r/movies").unwrap(); - assert_eq!(id.parent().unwrap().as_str(), "reddit.com/r"); + let id = ItemId::from_url("https://reddit.com/r/movies").unwrap(); + assert_eq!(id.parent().unwrap().as_str(), "https://reddit.com/r"); + } + + #[test] + fn breadcrumb_paths_skip_phantom_comments() { + let id = ItemId::from_url("https://reddit.com/r/aww/comments/1trnvdl").unwrap(); + let crumbs = id.breadcrumb_paths(); + let paths: Vec<_> = crumbs.iter().map(|p| p.as_str()).collect(); + assert!(!paths.iter().any(|p| p.ends_with("/comments"))); + assert!(paths.contains(&"https://reddit.com/r/aww")); } #[test] - fn breadcrumb_paths() { - let id = ItemId::parse("reddit.com/r/movies").unwrap(); + fn breadcrumb_paths_subreddit() { + let id = ItemId::from_url("https://reddit.com/r/movies").unwrap(); let crumbs = id.breadcrumb_paths(); let paths: Vec<_> = crumbs.iter().map(|p| p.as_str()).collect(); assert_eq!( paths, - vec!["reddit.com", "reddit.com/r", "reddit.com/r/movies"] + vec![ + "https://reddit.com", + "https://reddit.com/r", + "https://reddit.com/r/movies" + ] ); } @@ -312,33 +249,33 @@ mod tests { fn legacy_scope_maps_to_reddit_sub() { assert_eq!( ItemId::from_legacy_scope("rust").as_str(), - "reddit.com/r/rust" + "https://reddit.com/r/rust" ); assert!(ItemId::from_legacy_scope("").is_root()); } #[test] - fn browse_href_wraps_canonical_path() { - let id = ItemId::parse("reddit.com/r/rust").unwrap(); + fn browse_href_wraps_canonical_url() { + let id = ItemId::from_url("https://reddit.com/r/rust").unwrap(); assert_eq!(id.browse_href(), "/~/https://reddit.com/r/rust"); } #[test] fn from_browse_tail_parses_full_url() { let id = ItemId::from_browse_tail("https://reddit.com/r/AmITheAsshole"); - assert_eq!(id.as_str(), "reddit.com/r/amitheasshole"); + assert_eq!(id.as_str(), "https://reddit.com/r/amitheasshole"); } #[test] fn from_storage_strips_post_title_slug() { let id = ItemId::from_storage("reddit.com/r/rust/comments/aaa/announcing_rust_199").unwrap(); - assert_eq!(id.as_str(), "reddit.com/r/rust/comments/aaa"); + assert_eq!(id.as_str(), "https://reddit.com/r/rust/comments/aaa"); } #[test] fn from_browse_uri_strips_prefix() { let id = ItemId::from_browse_uri("/~/https://reddit.com/r/rust").unwrap(); - assert_eq!(id.as_str(), "reddit.com/r/rust"); + assert_eq!(id.as_str(), "https://reddit.com/r/rust"); } } diff --git a/server/src/projection_apply.rs b/server/src/projection_apply.rs index ebe122e417bda1d9369a53443de93d28213c5a0d..5644557a41b3e9497c7421b444155ae629fa79f1 100644 --- a/server/src/projection_apply.rs +++ b/server/src/projection_apply.rs @@ -19,13 +19,21 @@ use crate::{ storage_schema::{ensure_path_writes, entity_view_writes, vote_writes}, }; -/// Legacy-compatible scope parsing for persisted vote events. +fn parse_event_id(id: &str) -> Result { + ItemId::from_storage(id) + .or_else(|| ItemId::parse(id)) + .ok_or_else(|| EventLogError::Apply(format!("invalid id: {id}"))) +} + +/// Scope key from a vote event (canonicalized at apply time). fn parent_from_event_scope(scope: &str) -> ItemId { - if scope.contains('/') { - ItemId::parse(scope).unwrap_or_else(|| ItemId::from_legacy_scope(scope)) - } else { - ItemId::from_legacy_scope(scope) + let s = scope.trim(); + if s.is_empty() { + return ItemId::root(); } + ItemId::from_storage(s) + .or_else(|| ItemId::parse(s)) + .unwrap_or_else(|| ItemId::from_legacy_scope(s)) } pub fn apply_records( @@ -68,15 +76,11 @@ pub fn apply_records( vote_parents.insert(parent); } Event::NodeEnsured { id } => { - let parsed = ItemId::parse(id) - .or_else(|| ItemId::from_url(id)) - .ok_or_else(|| EventLogError::Apply(format!("invalid node id: {id}")))?; + let parsed = parse_event_id(id)?; ensure_path_writes(&mut batch, &parsed); } Event::EntityImported { id, payload, .. } => { - let parsed = ItemId::parse(id) - .or_else(|| ItemId::from_url(id)) - .ok_or_else(|| EventLogError::Apply(format!("invalid entity id: {id}")))?; + let parsed = parse_event_id(id)?; let view = entity_view_from_payload(&parsed, payload); entity_view_writes(&mut batch, &parsed, view.as_ref()); entity_store @@ -92,7 +96,6 @@ pub fn apply_records( .commit_with(durable::Durability::DisableWal) .map_err(|e| EventLogError::Apply(e.to_string()))?; - // Cap recent-vote windows (idempotent, blind; not part of the cursor batch). for parent in vote_parents { projection_store .trim_recent_votes(&parent) diff --git a/server/src/reddit.rs b/server/src/reddit.rs index 72caf7c33b9dd44ea15b62b59e91227cf96a3431..20b7f9e3f8be39268a1767d09f5cf81eaa6ae0df 100644 --- a/server/src/reddit.rs +++ b/server/src/reddit.rs @@ -183,7 +183,7 @@ pub fn entity_view_from_payload( id: &ItemId, payload: &Value, ) -> Option { - if id.as_str().starts_with("reddit.com") { + if id.as_str().contains("reddit.com") { return parse_reddit_view(id, payload); } None @@ -495,41 +495,59 @@ fn rate_limit_reset_secs(resp: &reqwest::Response) -> u64 { .unwrap_or(5) } +fn reddit_path_segments(id: &ItemId) -> Option> { + let s = id.as_str(); + let rest = s + .strip_prefix("https://reddit.com/") + .or_else(|| s.strip_prefix("http://reddit.com/")) + .or_else(|| s.strip_prefix("reddit.com/"))?; + let segments: Vec = rest + .split('/') + .filter(|p| !p.is_empty()) + .map(str::to_string) + .collect(); + Some(segments) +} + pub fn map_item_to_reddit_api(id: &ItemId, api_base: &str) -> String { - let path = id.as_str(); - if !path.starts_with("reddit.com/") && path != "reddit.com" { - return String::new(); - } + let segments = match reddit_path_segments(id) { + Some(s) => s, + None if matches!( + id.as_str(), + "https://reddit.com" | "http://reddit.com" | "reddit.com" + ) => + { + return String::new(); + } + None => return String::new(), + }; let base = api_base.trim_end_matches('/'); - let segments: Vec<&str> = path.split('/').collect(); - - if let Some(i) = segments.iter().position(|&p| p == "comments") { + if let Some(i) = segments.iter().position(|p| p == "comments") { if segments.len() > i + 1 { - let api_path = segments[1..=i + 1].join("/"); + let api_path = segments[..=i + 1].join("/"); return format!("{base}/{api_path}.json?raw_json=1"); } } - if segments.len() == 3 && segments[1] == "r" { - return format!("{base}/r/{}/about.json?raw_json=1", segments[2]); + if segments.len() == 2 && segments[0] == "r" { + return format!("{base}/r/{}/about.json?raw_json=1", segments[1]); } String::new() } /// Listing URL for a node's children. Currently only subreddits -/// (`reddit.com/r/` → `/r/.json`) expose a child listing. +/// (`https://reddit.com/r/` → `/r/.json`) expose a child listing. pub fn map_children_url(id: &ItemId, api_base: &str) -> String { - let path = id.as_str(); - if !path.starts_with("reddit.com/") { - return String::new(); - } + let segments = match reddit_path_segments(id) { + Some(s) => s, + None => return String::new(), + }; let base = api_base.trim_end_matches('/'); - let segments: Vec<&str> = path.split('/').collect(); - if segments.len() == 3 && segments[1] == "r" { - return format!("{base}/r/{}.json?raw_json=1&limit=25", segments[2]); + if segments.len() == 2 && segments[0] == "r" { + return format!("{base}/r/{}.json?raw_json=1&limit=25", segments[1]); } String::new() } @@ -548,8 +566,8 @@ fn parse_children(_parent: &ItemId, payload: &Value) -> Vec<(ItemId, Value)> { Some(p) if !p.is_empty() => p, _ => continue, }; - let path = format!("reddit.com{}", permalink.trim_end_matches('/')); - if let Some(id) = ItemId::from_storage(&path) { + let raw = format!("https://reddit.com{}", permalink.trim_end_matches('/')); + if let Some(id) = ItemId::from_url(&raw) { out.push((id, child.clone())); } } @@ -683,7 +701,7 @@ mod tests { #[test] fn map_subreddit_about_url() { - let id = ItemId::parse("reddit.com/r/rust").unwrap(); + let id = ItemId::from_url("https://reddit.com/r/rust").unwrap(); assert_eq!( map_item_to_reddit_api(&id, "https://www.reddit.com"), "https://www.reddit.com/r/rust/about.json?raw_json=1" @@ -699,7 +717,8 @@ mod tests { let json = include_str!("../../test/fixtures/reddit/r_rust_about.json"); let v: Value = serde_json::from_str(json).unwrap(); let entity = - entity_view_from_payload(&ItemId::parse("reddit.com/r/rust").unwrap(), &v).unwrap(); + entity_view_from_payload(&ItemId::from_url("https://reddit.com/r/rust").unwrap(), &v) + .unwrap(); assert_eq!(entity.title, "The Rust Programming Language"); } @@ -707,7 +726,8 @@ mod tests { fn parse_post_listing_extracts_thumb_and_full_preview() { let json = include_str!("../../test/fixtures/reddit/post_preview.json"); let v: Value = serde_json::from_str(json).unwrap(); - let id = ItemId::parse("reddit.com/r/nsfw/comments/1tpy6a1/angel_eyes").unwrap(); + let id = + ItemId::from_url("https://reddit.com/r/nsfw/comments/1tpy6a1/angel_eyes").unwrap(); let entity = entity_view_from_payload(&id, &v).unwrap(); assert_eq!(entity.title, "Angel Eyes"); assert!(entity.thumb_url.as_ref().unwrap().contains("width=140")); diff --git a/server/src/reducer.rs b/server/src/reducer.rs index 6578f64a41726845517cdbf59a359c69e0aa56db..5179ddeca7a7cb0fb92dfc4aa9d5a80bd9125611 100644 --- a/server/src/reducer.rs +++ b/server/src/reducer.rs @@ -248,16 +248,18 @@ mod from_recorded_tests { #[test] fn ensure_path_wires_children() { let mut tree = GlobalTree::new(); - let id = ItemId::parse("reddit.com/r/rust").unwrap(); + let id = ItemId::from_url("https://reddit.com/r/rust").unwrap(); tree.ensure_path(&id); let root = tree.get(&ItemId::root()).unwrap(); assert!(root .children - .contains(&ItemId::parse("reddit.com").unwrap())); - let reddit = tree.get(&ItemId::parse("reddit.com").unwrap()).unwrap(); + .contains(&ItemId::from_url("https://reddit.com").unwrap())); + let reddit = tree + .get(&ItemId::from_url("https://reddit.com").unwrap()) + .unwrap(); assert!(reddit .children - .contains(&ItemId::parse("reddit.com/r").unwrap())); + .contains(&ItemId::from_url("https://reddit.com/r").unwrap())); let sub = tree.get(&id).unwrap(); assert_eq!(sub.id, id); } diff --git a/server/src/render/reddit.rs b/server/src/render/reddit.rs index 7f840aa33b734a31d8cf3341a0581c8bcb9bbcf3..595e202436040b0bfc419e68f083f94757ba5d0c 100644 --- a/server/src/render/reddit.rs +++ b/server/src/render/reddit.rs @@ -9,7 +9,7 @@ use crate::{ }; pub fn is_reddit_post(id: &ItemId) -> bool { - id.as_str().starts_with("reddit.com/") && id.as_str().contains("/comments/") + id.as_str().contains("reddit.com/") && id.as_str().contains("/comments/") } /// Post detail card (inside [`crate::fetch::html::entity_panel`]). diff --git a/server/src/state.rs b/server/src/state.rs index 78126f08d90f8069a586279d79258c27a9f9f7a4..513b329ef45fc332e63b3f8ed8498a1f59feb07c 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -217,15 +217,21 @@ impl AppState { ratio_right: i32, ) -> Result<(), String> { let ts = crate::html::now_ms(); - let vote = VoteData::from_recorded(ts, a, b, ratio_left, ratio_right) - .ok_or_else(|| "invalid vote: need two distinct non-empty items".to_string())?; + let a_raw = a.trim(); + let b_raw = b.trim(); + if a_raw.is_empty() || b_raw.is_empty() || a_raw == b_raw { + return Err("invalid vote: need two distinct non-empty items".to_string()); + } + // Validate items canonicalize (or are opaque keys) before append. + let _ = VoteData::from_recorded(ts, a_raw, b_raw, ratio_left, ratio_right) + .ok_or_else(|| "invalid vote: need two distinct parseable items".to_string())?; let event = Event::VoteRecorded { ts, - a: vote.a.as_str().to_string(), - b: vote.b.as_str().to_string(), - ratio_left: vote.ratio_left, - ratio_right: vote.ratio_right, + a: a_raw.to_string(), + b: b_raw.to_string(), + ratio_left, + ratio_right, scope: parent.as_str().to_string(), }; @@ -253,7 +259,7 @@ mod tests { let log = EventLog::new(log_path.to_string_lossy().into_owned()); let payload = json!({"kind":"t5","data":{"title":"Rust","display_name":"rust"}}); let event = Event::EntityImported { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), ts: 1, payload: payload.clone(), }; @@ -266,14 +272,14 @@ mod tests { .await .unwrap(); let tree = projection_store - .scope_tree(&ItemId::parse("reddit.com/r/rust").unwrap()) + .scope_tree(&ItemId::parse("https://reddit.com/r/rust").unwrap()) .unwrap(); let node = tree - .get(&ItemId::parse("reddit.com/r/rust").unwrap()) + .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) .unwrap(); assert_eq!(node.data.as_ref().unwrap().title, "Rust"); let stored = entity_store - .get(&ItemId::parse("reddit.com/r/rust").unwrap()) + .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) .unwrap() .unwrap(); assert_eq!(stored["data"]["display_name"], "rust"); @@ -289,13 +295,13 @@ mod tests { event_record( 1, Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }, ), event_record( 2, Event::EntityImported { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), ts: 2, payload: payload.clone(), }, @@ -325,7 +331,7 @@ mod tests { &[event_record( 1, Event::NodeEnsured { - id: "reddit.com/r/stale".into(), + id: "https://reddit.com/r/stale".into(), }, )], ) @@ -352,11 +358,11 @@ mod tests { let root = tree.get(&ItemId::root()).unwrap(); assert!(root.children.contains(&ItemId::parse("alpha").unwrap())); assert!(projection_store - .load_node(&ItemId::parse("reddit.com/r/stale").unwrap()) + .load_node(&ItemId::parse("https://reddit.com/r/stale").unwrap()) .unwrap() .is_none()); let stored = entity_store - .get(&ItemId::parse("reddit.com/r/rust").unwrap()) + .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) .unwrap() .unwrap(); assert_eq!(stored["data"]["display_name"], "rust"); @@ -370,7 +376,7 @@ mod tests { log.append(&event_record( 1, Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }, )) .await @@ -387,7 +393,7 @@ mod tests { &[event_record( 2, Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }, )], ) @@ -454,7 +460,7 @@ mod tests { port: 0, }) .await; - let id = ItemId::parse("reddit.com/r/rust").unwrap(); + let id = ItemId::parse("https://reddit.com/r/rust").unwrap(); state.ensure_node(&id).await.unwrap(); @@ -465,11 +471,11 @@ mod tests { let projected = state.projection_store.load_tree().unwrap(); assert!(projected.get(&id).is_some()); let reddit = projected - .get(&ItemId::parse("reddit.com").unwrap()) + .get(&ItemId::from_url("https://reddit.com").unwrap()) .unwrap(); assert!(reddit .children - .contains(&ItemId::parse("reddit.com/r").unwrap())); + .contains(&ItemId::from_url("https://reddit.com/r").unwrap())); } #[tokio::test] @@ -510,7 +516,7 @@ mod tests { log.append(&event_record( 1, Event::NodeEnsured { - id: "reddit.com/r/rust".into(), + id: "https://reddit.com/r/rust".into(), }, )) .await @@ -518,7 +524,7 @@ mod tests { log.append(&event_record( 2, Event::NodeEnsured { - id: "reddit.com/r/python".into(), + id: "https://reddit.com/r/python".into(), }, )) .await @@ -544,13 +550,13 @@ mod tests { 2 ); let tree = second - .scope_tree(&ItemId::parse("reddit.com/r/rust").unwrap()) + .scope_tree(&ItemId::parse("https://reddit.com/r/rust").unwrap()) .unwrap(); assert!(tree - .get(&ItemId::parse("reddit.com/r/rust").unwrap()) + .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) .is_some()); assert!(tree - .get(&ItemId::parse("reddit.com/r/python").unwrap()) + .get(&ItemId::parse("https://reddit.com/r/python").unwrap()) .is_none()); } @@ -611,7 +617,7 @@ mod tests { #[test] fn parse_item_param_from_url() { let id = parse_item_param("https://reddit.com/r/rust"); - assert_eq!(id.as_str(), "reddit.com/r/rust"); + assert_eq!(id.as_str(), "https://reddit.com/r/rust"); } #[test] diff --git a/server/src/url_rules/engine.rs b/server/src/url_rules/engine.rs new file mode 100644 index 0000000000000000000000000000000000000000..e29b6b48c08deb7bffe031b1e542b1e25a7bef15 --- /dev/null +++ b/server/src/url_rules/engine.rs @@ -0,0 +1,187 @@ +//! Composable URL normalization primitives. + +use std::collections::HashMap; + +use url::Url; + +/// Mutable URL view used by rule combinators before serializing to a canonical string. +#[derive(Debug, Clone)] +pub struct ParsedUrl { + pub scheme: String, + pub host: String, + pub path_segments: Vec, + pub query: HashMap, + pub fragment: Option, +} + +impl ParsedUrl { + pub fn parse(raw: &str) -> Option { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return None; + } + + let with_scheme = if trimmed.contains("://") { + trimmed.to_string() + } else if trimmed.starts_with("r/") || trimmed.starts_with("/r/") { + let rest = trimmed.trim_start_matches('/').trim_start_matches("r/"); + format!("https://reddit.com/r/{rest}") + } else if trimmed.contains('.') && !trimmed.starts_with('/') { + format!("https://{trimmed}") + } else { + trimmed.to_string() + }; + + let url = Url::parse(&with_scheme).ok()?; + let host = url.host_str()?.to_string(); + let path_segments: Vec = url + .path_segments() + .map(|segs| segs.filter(|s| !s.is_empty()).map(str::to_string).collect()) + .unwrap_or_default(); + + let mut query = HashMap::new(); + for (k, v) in url.query_pairs() { + query.insert(k.into_owned(), v.into_owned()); + } + + Some(Self { + scheme: url.scheme().to_string(), + path_segments, + query, + fragment: url.fragment().map(str::to_string), + host, + }) + } + + pub fn with_path_segments(&self, segments: &[String]) -> Self { + let mut u = self.clone(); + u.path_segments = segments.to_vec(); + u + } + + pub fn to_url(&self) -> Option { + let mut url = if self.path_segments.is_empty() { + Url::parse(&format!("{}://{}", self.scheme, self.host)).ok()? + } else { + let path = format!("/{}", self.path_segments.join("/")); + Url::parse(&format!("{}://{}{}", self.scheme, self.host, path)).ok()? + }; + if !self.query.is_empty() { + let mut pairs: Vec<_> = self.query.iter().collect(); + pairs.sort_by(|a, b| a.0.cmp(b.0)); + url.query_pairs_mut().clear(); + for (k, v) in pairs { + url.query_pairs_mut().append_pair(k, v); + } + } + if let Some(ref frag) = self.fragment { + url.set_fragment(Some(frag)); + } + Some(url) + } + + pub fn canonical_string(&self) -> Option { + let url = self.to_url()?; + let mut s = url.to_string(); + if self.path_segments.is_empty() { + s = s.trim_end_matches('/').to_string(); + } + Some(s) + } +} + +pub fn force_https(u: &mut ParsedUrl) { + if u.scheme == "http" { + u.scheme = "https".to_string(); + } +} + +pub fn drop_fragment(u: &mut ParsedUrl) { + u.fragment = None; +} + +pub fn strip_www(u: &mut ParsedUrl) { + if u.host.starts_with("www.") { + u.host = u.host[4..].to_string(); + } +} + +pub fn lowercase_host(u: &mut ParsedUrl) { + u.host = u.host.to_ascii_lowercase(); +} + +pub fn lowercase_path(u: &mut ParsedUrl) { + for seg in &mut u.path_segments { + *seg = seg.to_ascii_lowercase(); + } +} + +pub fn clear_query(u: &mut ParsedUrl) { + u.query.clear(); +} + +pub fn keep_only_query(u: &mut ParsedUrl, keys: &[&str]) { + u.query + .retain(|k, _| keys.iter().any(|want| want == &k.as_str())); +} + +pub fn strip_tracking_params(u: &mut ParsedUrl) { + u.query.retain(|k, _| { + let lower = k.to_ascii_lowercase(); + !(lower.starts_with("utm_") + || matches!( + lower.as_str(), + "fbclid" | "gclid" | "ref" | "ref_src" | "ref_source" | "mc_cid" | "mc_eid" + )) + }); +} + +pub fn truncate_after_segment(u: &mut ParsedUrl, name: &str, keep: usize) { + if let Some(i) = u.path_segments.iter().position(|s| s == name) { + let end = (i + 1 + keep).min(u.path_segments.len()); + u.path_segments.truncate(end); + } +} + +pub fn drop_listing_suffix(u: &mut ParsedUrl, suffixes: &[&str]) { + if u.path_segments.len() >= 3 && u.path_segments.first().map(String::as_str) == Some("r") { + if let Some(last) = u.path_segments.last() { + if suffixes.iter().any(|s| *s == last.as_str()) { + u.path_segments.pop(); + } + } + } +} + +pub fn normalize_reddit_host(u: &mut ParsedUrl) { + if matches!( + u.host.as_str(), + "old.reddit.com" | "new.reddit.com" | "www.reddit.com" + ) { + u.host = "reddit.com".to_string(); + } +} + +pub fn rewrite_youtu_be(u: &mut ParsedUrl) { + if u.host == "youtu.be" && u.path_segments.len() == 1 { + let id = u.path_segments[0].clone(); + u.host = "youtube.com".to_string(); + u.path_segments = vec!["watch".to_string()]; + u.query.insert("v".to_string(), id); + } +} + +pub fn rewrite_youtube_shorts(u: &mut ParsedUrl) { + if u.host == "youtube.com" && u.path_segments.first().map(String::as_str) == Some("shorts") { + if let Some(id) = u.path_segments.get(1).cloned() { + u.path_segments = vec!["watch".to_string()]; + u.query.insert("v".to_string(), id); + } + } +} + +pub fn normalize_youtube_host(u: &mut ParsedUrl) { + if matches!(u.host.as_str(), "m.youtube.com" | "www.youtube.com") { + u.host = "youtube.com".to_string(); + } +} diff --git a/server/src/url_rules/mod.rs b/server/src/url_rules/mod.rs new file mode 100644 index 0000000000000000000000000000000000000000..03d53bd3e82d704a01ba3fd8dd02b7d31422c0de --- /dev/null +++ b/server/src/url_rules/mod.rs @@ -0,0 +1,13 @@ +//! URL canonicalization and hierarchy rules for [`crate::path_types::ItemId`]. + +mod engine; +mod registry; + +pub use registry::{ + canonicalize_raw, looks_like_url, navigable_breadcrumbs, parent_url, resolve_id, CanonicalResult, +}; + +/// Resolve raw input to canonical URL. +pub fn resolve_canonical(raw: &str) -> Option { + canonicalize_raw(raw.trim()).map(|r| r.canonical) +} diff --git a/server/src/url_rules/registry.rs b/server/src/url_rules/registry.rs new file mode 100644 index 0000000000000000000000000000000000000000..14514e9af8385fb2b9b2f35eb9ee14d453d4b97c --- /dev/null +++ b/server/src/url_rules/registry.rs @@ -0,0 +1,235 @@ +//! Per-domain canonicalization and hierarchy rules. + +use std::collections::HashSet; + +use super::engine::{ + clear_query, drop_fragment, drop_listing_suffix, force_https, keep_only_query, lowercase_host, + lowercase_path, normalize_reddit_host, normalize_youtube_host, rewrite_youtu_be, + rewrite_youtube_shorts, strip_tracking_params, strip_www, truncate_after_segment, ParsedUrl, +}; + +/// Result of canonicalizing a raw URL string. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CanonicalResult { + pub canonical: String, + /// When the input normalizes to a different string, the original is an alias. + pub alias_of: Option, +} + +fn apply_global(u: &mut ParsedUrl) { + force_https(u); + drop_fragment(u); + strip_www(u); + lowercase_host(u); + strip_tracking_params(u); +} + +fn normalize_reddit(u: &mut ParsedUrl) { + normalize_reddit_host(u); + lowercase_path(u); + truncate_after_segment(u, "comments", 1); + drop_listing_suffix(u, &["hot", "top", "new", "rising", "controversial"]); + clear_query(u); +} + +fn normalize_youtube(u: &mut ParsedUrl) { + rewrite_youtu_be(u); + normalize_youtube_host(u); + rewrite_youtube_shorts(u); + keep_only_query(u, &["v", "list"]); +} + +fn normalize_default(_u: &mut ParsedUrl) { + // Global rules only. +} + +fn domain_key(host: &str) -> &'static str { + if host == "reddit.com" || host.ends_with(".reddit.com") { + "reddit.com" + } else if host == "youtube.com" || host == "youtu.be" { + "youtube.com" + } else { + "default" + } +} + +fn normalize_for_host(u: &mut ParsedUrl) { + apply_global(u); + match domain_key(&u.host) { + "reddit.com" => normalize_reddit(u), + "youtube.com" => normalize_youtube(u), + _ => normalize_default(u), + } +} + +/// Structural path segments that must not become standalone tree nodes when more path follows. +fn structural_trailing(host: &str) -> &'static [&'static str] { + match domain_key(host) { + "reddit.com" => &["comments"], + _ => &[], + } +} + +/// Canonicalize a raw URL. Returns `None` if the input is not URL-like. +pub fn canonicalize_raw(raw: &str) -> Option { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return None; + } + let mut u = ParsedUrl::parse(trimmed)?; + let input_snapshot = u.canonical_string()?; + normalize_for_host(&mut u); + let canonical = u.canonical_string()?; + let alias_of = if input_snapshot != canonical { + Some(trimmed.to_string()) + } else { + None + }; + Some(CanonicalResult { + canonical, + alias_of, + }) +} + +/// Resolve a stored or event id string to its canonical URL identity. +pub fn resolve_id(raw: &str) -> Option { + canonicalize_raw(raw).map(|r| r.canonical) +} + +/// Navigable ancestor URLs from domain root up to and including `canonical` (full URLs). +pub fn navigable_breadcrumbs(canonical: &str) -> Vec { + let Some(u) = ParsedUrl::parse(canonical) else { + return vec![canonical.to_string()]; + }; + let structural: HashSet<&str> = structural_trailing(&u.host).iter().copied().collect(); + let n = u.path_segments.len(); + let mut out = Vec::new(); + + // Domain root (no path segments). + if let Some(base) = u.with_path_segments(&[]).canonical_string() { + out.push(base); + } + + for i in 0..n { + let segs: Vec = u.path_segments[..=i].to_vec(); + let is_last = i == n - 1; + let seg = u.path_segments[i].as_str(); + if structural.contains(seg) && !is_last { + continue; + } + if let Some(url) = u.with_path_segments(&segs).canonical_string() { + if out.last() != Some(&url) { + out.push(url); + } + } + } + out +} + +/// Immediate parent scope URL, or `None` for tree root / opaque single-segment ids. +pub fn parent_url(canonical: &str) -> Option { + let crumbs = navigable_breadcrumbs(canonical); + if crumbs.len() <= 1 { + None + } else { + crumbs.get(crumbs.len() - 2).cloned() + } +} + +/// True when `raw` looks like a URL (has scheme or host-like shape). +pub fn looks_like_url(raw: &str) -> bool { + let t = raw.trim(); + t.contains("://") + || t.starts_with("r/") + || t.starts_with("/r/") + || (t.contains('.') && t.contains('/')) + || t.starts_with("reddit.com") + || t.starts_with("www.") + || t.starts_with("youtu.be/") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn reddit_post_drops_slug_and_normalizes_host() { + let r = canonicalize_raw( + "https://old.reddit.com/r/AmItheAsshole/comments/1trnvdl/aita_for_cancelling/", + ) + .unwrap(); + assert_eq!( + r.canonical, + "https://reddit.com/r/amitheasshole/comments/1trnvdl" + ); + } + + #[test] + fn reddit_strips_query_and_listing() { + assert_eq!( + canonicalize_raw("https://www.reddit.com/r/rust/?sort=top") + .unwrap() + .canonical, + "https://reddit.com/r/rust" + ); + assert_eq!( + canonicalize_raw("https://www.reddit.com/r/programming/hot") + .unwrap() + .canonical, + "https://reddit.com/r/programming" + ); + } + + #[test] + fn reddit_short_path() { + assert_eq!( + canonicalize_raw("r/rust").unwrap().canonical, + "https://reddit.com/r/rust" + ); + } + + #[test] + fn reddit_breadcrumbs_skip_phantom_comments() { + let post = "https://reddit.com/r/aww/comments/1trnvdl"; + let crumbs = navigable_breadcrumbs(post); + assert!(!crumbs.iter().any(|c| c.ends_with("/comments"))); + assert_eq!( + crumbs.last().map(String::as_str), + Some(post) + ); + assert!(crumbs.contains(&"https://reddit.com/r/aww".to_string())); + } + + #[test] + fn reddit_parent_of_post_is_subreddit() { + assert_eq!( + parent_url("https://reddit.com/r/aww/comments/1trnvdl").as_deref(), + Some("https://reddit.com/r/aww") + ); + } + + #[test] + fn youtube_youtu_be_and_watch_same_canonical() { + let a = canonicalize_raw("https://youtu.be/dQw4w9WgXcQ").unwrap().canonical; + let b = canonicalize_raw("https://www.youtube.com/watch?v=dQw4w9WgXcQ&t=10").unwrap(); + assert_eq!(a, b.canonical); + assert_eq!(a, "https://youtube.com/watch?v=dQw4w9WgXcQ"); + } + + #[test] + fn legacy_schemeless_upgrades() { + assert_eq!( + canonicalize_raw("reddit.com/r/rust/comments/aaa/announcing_rust_199") + .unwrap() + .canonical, + "https://reddit.com/r/rust/comments/aaa" + ); + } + + #[test] + fn alias_recorded_when_input_differs() { + let r = canonicalize_raw("https://youtu.be/abc123").unwrap(); + assert_eq!(r.canonical, "https://youtube.com/watch?v=abc123"); + assert!(r.alias_of.is_some()); + } +} Side B — contributor: tommy-mor Side B — commit message: [9be54f7d] Add GitHub OAuth auth and move durable out of tree. Gate votes behind session + pseudonym claim, project identity events into durable maps, and depend on tommy-mor/durable from git instead of the in-repo crates. Co-authored-by: Cursor Side B — unified diff (full patch): diff --git a/Cargo.lock b/Cargo.lock index 49a908ef935c430dbe63c6a28d8a24e38b489486..aa02997ad85777195f135bfd9456bcee0fc9a590 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -69,12 +69,6 @@ version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" -[[package]] -name = "autocfg" -version = "1.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" - [[package]] name = "axum" version = "0.7.9" @@ -211,21 +205,6 @@ dependencies = [ "syn", ] -[[package]] -name = "bit-set" -version = "0.8.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "08807e080ed7f9d5433fa9b275196cfc35414f66a0c79d864dc51a0d825231a3" -dependencies = [ - "bit-vec", -] - -[[package]] -name = "bit-vec" -version = "0.8.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5e764a1d40d510daf35e07be9eb06e75770908c27d411ee6c92109c9840eaaf7" - [[package]] name = "bitflags" version = "1.3.2" @@ -435,19 +414,19 @@ dependencies = [ [[package]] name = "durable" version = "0.2.0" +source = "git+https://github.com/tommy-mor/durable.git?branch=main#a6c14eaa809693140eea0c22b07ef24d8e74adaf" dependencies = [ "ciborium", "durable-derive", - "proptest", "rocksdb", "serde", - "tempfile", "thiserror", ] [[package]] name = "durable-derive" version = "0.2.0" +source = "git+https://github.com/tommy-mor/durable.git?branch=main#a6c14eaa809693140eea0c22b07ef24d8e74adaf" dependencies = [ "proc-macro2", "quote", @@ -1238,15 +1217,6 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" -[[package]] -name = "num-traits" -version = "0.2.19" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" -dependencies = [ - "autocfg", -] - [[package]] name = "once_cell" version = "1.21.4" @@ -1358,7 +1328,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d" dependencies = [ "phf_shared", - "rand 0.8.6", + "rand", ] [[package]] @@ -1467,31 +1437,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "proptest" -version = "1.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4b45fcc2344c680f5025fe57779faef368840d0bd1f42f216291f0dc4ace4744" -dependencies = [ - "bit-set", - "bit-vec", - "bitflags 2.11.1", - "num-traits", - "rand 0.9.4", - "rand_chacha 0.9.0", - "rand_xorshift", - "regex-syntax", - "rusty-fork", - "tempfile", - "unarray", -] - -[[package]] -name = "quick-error" -version = "1.2.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0" - [[package]] name = "quote" version = "1.0.45" @@ -1520,18 +1465,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" dependencies = [ "libc", - "rand_chacha 0.3.1", - "rand_core 0.6.4", -] - -[[package]] -name = "rand" -version = "0.9.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" -dependencies = [ - "rand_chacha 0.9.0", - "rand_core 0.9.5", + "rand_chacha", + "rand_core", ] [[package]] @@ -1541,17 +1476,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" dependencies = [ "ppv-lite86", - "rand_core 0.6.4", -] - -[[package]] -name = "rand_chacha" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" -dependencies = [ - "ppv-lite86", - "rand_core 0.9.5", + "rand_core", ] [[package]] @@ -1563,24 +1488,6 @@ dependencies = [ "getrandom 0.2.17", ] -[[package]] -name = "rand_core" -version = "0.9.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" -dependencies = [ - "getrandom 0.3.4", -] - -[[package]] -name = "rand_xorshift" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "513962919efc330f829edb2535844d1b912b0fbe2ca165d613e4e8788bb05a5a" -dependencies = [ - "rand_core 0.9.5", -] - [[package]] name = "redox_syscall" version = "0.5.18" @@ -1747,18 +1654,6 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" -[[package]] -name = "rusty-fork" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cc6bf79ff24e648f6da1f8d1f011e9cac26491b619e6b9280f2b47f1774e6ee2" -dependencies = [ - "fnv", - "quick-error", - "tempfile", - "wait-timeout", -] - [[package]] name = "ryu" version = "1.0.23" @@ -1940,7 +1835,7 @@ dependencies = [ "durable", "futures-util", "maud", - "rand 0.8.6", + "rand", "reqwest", "serde", "serde_json", @@ -2335,12 +2230,6 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" -[[package]] -name = "unarray" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eaea85b334db583fe3274d12b4cd1880032beab409c0d774be044d4480ab9a94" - [[package]] name = "unicode-ident" version = "1.0.24" @@ -2407,15 +2296,6 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" -[[package]] -name = "wait-timeout" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09ac3b126d3914f9849036f826e054cbabdc8519970b8998ddaf3b5bd3c65f11" -dependencies = [ - "libc", -] - [[package]] name = "want" version = "0.3.1" diff --git a/Cargo.toml b/Cargo.toml index 9c387a8e106861dae210eaf07857eee3b8dad92a..156820ec5151b709a254dafc177da10488fee566 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,3 +1,3 @@ [workspace] -members = ["server", "durable", "durable-derive"] +members = ["server"] resolver = "2" diff --git a/durable-derive/Cargo.toml b/durable-derive/Cargo.toml deleted file mode 100644 index 1e9979ef32073a9fdaff80269342637ba641f798..0000000000000000000000000000000000000000 --- a/durable-derive/Cargo.toml +++ /dev/null @@ -1,15 +0,0 @@ -[package] -name = "durable-derive" -version = "0.2.0" -edition = "2021" -authors = ["Durable Contributors"] -description = "#[derive(Durable)] macro for the durable crate" -license = "MIT OR Apache-2.0" - -[lib] -proc-macro = true - -[dependencies] -syn = { version = "2.0", features = ["full"] } -quote = "1.0" -proc-macro2 = "1.0" diff --git a/durable-derive/src/lib.rs b/durable-derive/src/lib.rs deleted file mode 100644 index e646ca36ae9813bbe0b3d20af4ba81e4b0e306c2..0000000000000000000000000000000000000000 --- a/durable-derive/src/lib.rs +++ /dev/null @@ -1,90 +0,0 @@ -//! `#[derive(Durable)]` for the `durable` crate. -//! -//! Turns a struct whose fields are durable schema types into a navigable schema: -//! -//! - implements `durable::Schema` for the struct, -//! - generates a `{Name}Fields` extension trait (implemented for -//! `durable::Path`) with one navigator method per field, and -//! - adds `Name::root()` / `Name::namespaced(name)` constructors. -//! -//! Each field is assigned a stable numeric id from its declaration order, which -//! is encoded into the on-disk key. Reordering fields changes the layout; add new -//! fields at the end. - -use proc_macro::TokenStream; -use quote::quote; -use syn::{parse_macro_input, Data, DeriveInput, Fields, Ident}; - -#[proc_macro_derive(Durable)] -pub fn derive_durable(input: TokenStream) -> TokenStream { - let input = parse_macro_input!(input as DeriveInput); - let name = &input.ident; - let vis = &input.vis; - let (impl_generics, ty_generics, where_clause) = input.generics.split_for_impl(); - - let fields = match &input.data { - Data::Struct(data) => match &data.fields { - Fields::Named(named) => &named.named, - _ => { - return syn::Error::new_spanned( - name, - "#[derive(Durable)] requires a struct with named fields", - ) - .to_compile_error() - .into(); - } - }, - _ => { - return syn::Error::new_spanned(name, "#[derive(Durable)] is only supported on structs") - .to_compile_error() - .into(); - } - }; - - let mut trait_methods = Vec::new(); - let mut impl_methods = Vec::new(); - - for (index, field) in fields.iter().enumerate() { - let field_ident = field.ident.as_ref().expect("named field"); - let field_ty = &field.ty; - let field_id = index as u32; - trait_methods.push(quote! { - fn #field_ident(&self) -> ::durable::Path<#field_ty>; - }); - impl_methods.push(quote! { - fn #field_ident(&self) -> ::durable::Path<#field_ty> { - self.child_field(#field_id) - } - }); - } - - let trait_name = Ident::new(&format!("{name}Fields"), name.span()); - let trait_doc = format!("Field navigators for [`{name}`], implemented for `durable::Path<{name}>`."); - - let expanded = quote! { - impl #impl_generics ::durable::Schema for #name #ty_generics #where_clause {} - - #[doc = #trait_doc] - #vis trait #trait_name { - #(#trait_methods)* - } - - impl #impl_generics #trait_name for ::durable::Path<#name #ty_generics> #where_clause { - #(#impl_methods)* - } - - impl #impl_generics #name #ty_generics #where_clause { - /// The root path of this schema (empty prefix; one root per database). - #vis fn root() -> ::durable::Path<#name #ty_generics> { - ::durable::Path::root() - } - - /// A root path namespaced under `name`, to share a database between schemas. - #vis fn namespaced(name: &str) -> ::durable::Path<#name #ty_generics> { - ::durable::Path::namespaced(name) - } - } - }; - - expanded.into() -} diff --git a/durable/.gitignore b/durable/.gitignore deleted file mode 100644 index 90c273ef12d6313594dde541c3465f2a47739729..0000000000000000000000000000000000000000 --- a/durable/.gitignore +++ /dev/null @@ -1,35 +0,0 @@ -all.txt - -# Generated by Cargo -# will have compiled files and executables -debug/ -target/ - -# These are backup files generated by rustfmt -**/*.rs.bk - -# MSVC Windows builds of rustc generate these, which store debugging information -*.pdb - -# Generated by cargo mutants -# Contains mutation testing data -**/mutants.out*/ - -# RustRover -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ - - -# Added by cargo - -/target - - -# Added by cargo -# -# already existing elements were commented out - -#/target diff --git a/durable/Cargo.toml b/durable/Cargo.toml deleted file mode 100644 index 95be0c71d64862c7cdb78e64fca768dcf83be961..0000000000000000000000000000000000000000 --- a/durable/Cargo.toml +++ /dev/null @@ -1,19 +0,0 @@ -[package] -name = "durable" -version = "0.2.0" -edition = "2021" -authors = ["Durable Contributors"] -description = "Deeply nested, precisely updatable RocksDB-backed data structures with paths-as-data" -license = "MIT OR Apache-2.0" - -[dependencies] -rocksdb = "0.21" -serde = { version = "1.0", features = ["derive"] } -thiserror = "1.0" -ciborium = "0.2.2" -durable-derive = { path = "../durable-derive", version = "0.2.0" } - -[dev-dependencies] -tempfile = "3.8" -proptest = "1.4" -serde = { version = "1.0", features = ["derive"] } diff --git a/durable/LICENSE b/durable/LICENSE deleted file mode 100644 index 261eeb9e9f8b2b4b0d119366dda99c6fd7d35c64..0000000000000000000000000000000000000000 --- a/durable/LICENSE +++ /dev/null @@ -1,201 +0,0 @@ - Apache License - Version 2.0, January 2004 - http://www.apache.org/licenses/ - - TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION - - 1. Definitions. - - "License" shall mean the terms and conditions for use, reproduction, - and distribution as defined by Sections 1 through 9 of this document. - - "Licensor" shall mean the copyright owner or entity authorized by - the copyright owner that is granting the License. - - "Legal Entity" shall mean the union of the acting entity and all - other entities that control, are controlled by, or are under common - control with that entity. For the purposes of this definition, - "control" means (i) the power, direct or indirect, to cause the - direction or management of such entity, whether by contract or - otherwise, or (ii) ownership of fifty percent (50%) or more of the - outstanding shares, or (iii) beneficial ownership of such entity. - - "You" (or "Your") shall mean an individual or Legal Entity - exercising permissions granted by this License. - - "Source" form shall mean the preferred form for making modifications, - including but not limited to software source code, documentation - source, and configuration files. - - "Object" form shall mean any form resulting from mechanical - transformation or translation of a Source form, including but - not limited to compiled object code, generated documentation, - and conversions to other media types. - - "Work" shall mean the work of authorship, whether in Source or - Object form, made available under the License, as indicated by a - copyright notice that is included in or attached to the work - (an example is provided in the Appendix below). - - "Derivative Works" shall mean any work, whether in Source or Object - form, that is based on (or derived from) the Work and for which the - editorial revisions, annotations, elaborations, or other modifications - represent, as a whole, an original work of authorship. For the purposes - of this License, Derivative Works shall not include works that remain - separable from, or merely link (or bind by name) to the interfaces of, - the Work and Derivative Works thereof. - - "Contribution" shall mean any work of authorship, including - the original version of the Work and any modifications or additions - to that Work or Derivative Works thereof, that is intentionally - submitted to Licensor for inclusion in the Work by the copyright owner - or by an individual or Legal Entity authorized to submit on behalf of - the copyright owner. For the purposes of this definition, "submitted" - means any form of electronic, verbal, or written communication sent - to the Licensor or its representatives, including but not limited to - communication on electronic mailing lists, source code control systems, - and issue tracking systems that are managed by, or on behalf of, the - Licensor for the purpose of discussing and improving the Work, but - excluding communication that is conspicuously marked or otherwise - designated in writing by the copyright owner as "Not a Contribution." - - "Contributor" shall mean Licensor and any individual or Legal Entity - on behalf of whom a Contribution has been received by Licensor and - subsequently incorporated within the Work. - - 2. Grant of Copyright License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - copyright license to reproduce, prepare Derivative Works of, - publicly display, publicly perform, sublicense, and distribute the - Work and such Derivative Works in Source or Object form. - - 3. Grant of Patent License. Subject to the terms and conditions of - this License, each Contributor hereby grants to You a perpetual, - worldwide, non-exclusive, no-charge, royalty-free, irrevocable - (except as stated in this section) patent license to make, have made, - use, offer to sell, sell, import, and otherwise transfer the Work, - where such license applies only to those patent claims licensable - by such Contributor that are necessarily infringed by their - Contribution(s) alone or by combination of their Contribution(s) - with the Work to which such Contribution(s) was submitted. If You - institute patent litigation against any entity (including a - cross-claim or counterclaim in a lawsuit) alleging that the Work - or a Contribution incorporated within the Work constitutes direct - or contributory patent infringement, then any patent licenses - granted to You under this License for that Work shall terminate - as of the date such litigation is filed. - - 4. Redistribution. You may reproduce and distribute copies of the - Work or Derivative Works thereof in any medium, with or without - modifications, and in Source or Object form, provided that You - meet the following conditions: - - (a) You must give any other recipients of the Work or - Derivative Works a copy of this License; and - - (b) You must cause any modified files to carry prominent notices - stating that You changed the files; and - - (c) You must retain, in the Source form of any Derivative Works - that You distribute, all copyright, patent, trademark, and - attribution notices from the Source form of the Work, - excluding those notices that do not pertain to any part of - the Derivative Works; and - - (d) If the Work includes a "NOTICE" text file as part of its - distribution, then any Derivative Works that You distribute must - include a readable copy of the attribution notices contained - within such NOTICE file, excluding those notices that do not - pertain to any part of the Derivative Works, in at least one - of the following places: within a NOTICE text file distributed - as part of the Derivative Works; within the Source form or - documentation, if provided along with the Derivative Works; or, - within a display generated by the Derivative Works, if and - wherever such third-party notices normally appear. The contents - of the NOTICE file are for informational purposes only and - do not modify the License. You may add Your own attribution - notices within Derivative Works that You distribute, alongside - or as an addendum to the NOTICE text from the Work, provided - that such additional attribution notices cannot be construed - as modifying the License. - - You may add Your own copyright statement to Your modifications and - may provide additional or different license terms and conditions - for use, reproduction, or distribution of Your modifications, or - for any such Derivative Works as a whole, provided Your use, - reproduction, and distribution of the Work otherwise complies with - the conditions stated in this License. - - 5. Submission of Contributions. Unless You explicitly state otherwise, - any Contribution intentionally submitted for inclusion in the Work - by You to the Licensor shall be under the terms and conditions of - this License, without any additional terms or conditions. - Notwithstanding the above, nothing herein shall supersede or modify - the terms of any separate license agreement you may have executed - with Licensor regarding such Contributions. - - 6. Trademarks. This License does not grant permission to use the trade - names, trademarks, service marks, or product names of the Licensor, - except as required for reasonable and customary use in describing the - origin of the Work and reproducing the content of the NOTICE file. - - 7. Disclaimer of Warranty. Unless required by applicable law or - agreed to in writing, Licensor provides the Work (and each - Contributor provides its Contributions) on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or - implied, including, without limitation, any warranties or conditions - of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A - PARTICULAR PURPOSE. You are solely responsible for determining the - appropriateness of using or redistributing the Work and assume any - risks associated with Your exercise of permissions under this License. - - 8. Limitation of Liability. In no event and under no legal theory, - whether in tort (including negligence), contract, or otherwise, - unless required by applicable law (such as deliberate and grossly - negligent acts) or agreed to in writing, shall any Contributor be - liable to You for damages, including any direct, indirect, special, - incidental, or consequential damages of any character arising as a - result of this License or out of the use or inability to use the - Work (including but not limited to damages for loss of goodwill, - work stoppage, computer failure or malfunction, or any and all - other commercial damages or losses), even if such Contributor - has been advised of the possibility of such damages. - - 9. Accepting Warranty or Additional Liability. While redistributing - the Work or Derivative Works thereof, You may choose to offer, - and charge a fee for, acceptance of support, warranty, indemnity, - or other liability obligations and/or rights consistent with this - License. However, in accepting such obligations, You may act only - on Your own behalf and on Your sole responsibility, not on behalf - of any other Contributor, and only if You agree to indemnify, - defend, and hold each Contributor harmless for any liability - incurred by, or claims asserted against, such Contributor by reason - of your accepting any such warranty or additional liability. - - END OF TERMS AND CONDITIONS - - APPENDIX: How to apply the Apache License to your work. - - To apply the Apache License to your work, attach the following - boilerplate notice, with the fields enclosed by brackets "[]" - replaced with your own identifying information. (Don't include - the brackets!) The text should be enclosed in the appropriate - comment syntax for the file format. We also recommend that a - file or class name and description of purpose be included on the - same "printed page" as the copyright notice for easier - identification within third-party archives. - - Copyright [yyyy] [name of copyright owner] - - Licensed under the Apache License, Version 2.0 (the "License"); - you may not use this file except in compliance with the License. - You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - - Unless required by applicable law or agreed to in writing, software - distributed under the License is distributed on an "AS IS" BASIS, - WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - See the License for the specific language governing permissions and - limitations under the License. diff --git a/durable/README.md b/durable/README.md deleted file mode 100644 index cd3c84ad801be756b643b8f479c2ff39e9904233..0000000000000000000000000000000000000000 --- a/durable/README.md +++ /dev/null @@ -1,168 +0,0 @@ -# durable - -Deeply nested, precisely updatable RocksDB-backed data structures for Rust, -built around **paths as data**. - -Most embedded-storage wrappers make you serialize a whole struct into one blob. -Updating one field means reading, deserializing, mutating, re-serializing, and -rewriting the entire value. `durable` takes the opposite approach: you describe -your data as a *schema* of composable types, and address any location with a -typed **path**. A path lowers to a deterministic RocksDB key with no I/O, so a -mutation touches exactly the keys it names — nothing else. - -```rust -use durable::{Db, Durable, Durability, Leaf, Map, Sum}; - -#[derive(Durable)] -struct Store { - scores: Map>, - title: Leaf, -} - -fn main() -> durable::Result<()> { - let db = Db::open("scores.db")?; - let root = Store::root(); - let alice = "alice".to_string(); - - // Three precise writes, one atomic batch, one WAL flush. - db.apply( - &[ - root.scores().key(&alice).add(10), // blind merge — no read - root.scores().key(&alice).add(5), - root.title().set(&"leaderboard".to_string()), - ], - Durability::SyncWal, - )?; - - assert_eq!(root.scores().key(&alice).get(&db)?, 15); - Ok(()) -} -``` - -## The model - -### Schema types - -A *schema* is a type-level description of a location's shape. Compose them -freely: - -| Type | Meaning | Key terminal ops | -|------|---------|------------------| -| `Leaf` | one CBOR-encoded value | `get`, `set`, `delete` | -| `Map` | keys `K` → sub-schema `V` | `key`, `keys`, `entries`, `len`, `contains`, `clear` | -| `List` | index-addressed sequence | `at`, `push`, `pop`, `iter`, `len`, `clear` | -| `Deque` | double-ended queue (O(1) ends) | `push_back`, `push_front`, `pop_front`, `pop_back`, `front`, `back`, `iter` | -| `Sum` | numeric accumulator | `add` (blind merge), `get`, `set` | -| `#[derive(Durable)] struct` | fixed named fields | one navigator method per field | - -Leaf- and `Sum`-valued maps additionally get `get`, `iter`, and -`transform_values` (a one-scan bulk rewrite that yields reified writes — e.g. -"decay every edge weight"). - -Nest them arbitrarily: - -```rust -use durable::{Deque, Durable, Leaf, Map, Sum}; -# use serde::{Serialize, Deserialize}; -# #[derive(Serialize, Deserialize)] struct Vote; -#[derive(Durable)] -#[allow(dead_code)] -struct GroupState { - edges: Map<(u32, u32), Sum>, - recent_votes: Deque>, - item_count: Sum, -} - -#[derive(Durable)] -#[allow(dead_code)] -struct Store { - scopes: Map, -} -``` - -Now `Store::root().scopes().key(&scope).edges().key(&(i, j)).add(1.0)` updates a -single edge weight without reading or rewriting anything else in the scope. - -### Paths are data - -`Path` is just a byte prefix plus a phantom schema type. Navigation is pure -and allocation-light; nothing hits the database until you read or apply. Because -paths are values you can build them once and reuse them, pass them around, and -compose them. - -### Mutations are reified - -Terminal mutating operations don't perform side effects — they return a -[`Write`], a typed wrapper around a plain-data [`Op`] (`Put` / `Delete` / -`DeletePrefix` / `Merge`). Collect several and apply them atomically: - -```rust,ignore -let writes = vec![ - edges.key(&(0, 1)).add(2.0), - edges.key(&(1, 0)).add(1.0), - voted_pairs.key(&(0, 1)).set(&true), -]; -db.apply(&writes, Durability::DisableWal)?; -``` - -Reified writes are inspectable and testable — you can assert on the `Op` a path -produces, log it, or serialize it. - -### Blind vs. read-modify-write - -The cost model is explicit, not hidden: - -- **Blind** (no read): `Leaf::set`/`delete`, `Sum::add`/`set`, `Map::clear`. - These are pure `Op` data and compose freely in a batch. -- **Read-modify-write**: `List::push`/`pop`, `Deque` pushes/pops (they read a - length/cursor). In a batch, appends are deferred and resolved at commit so - several land at contiguous indices in one atomic write. -- **Scan**: `Map::keys`/`iter`/`len`, `transform_values`. Prefix range scans. - -`Sum` deserves a special mention: it's backed by a RocksDB associative merge -operator, so `add` is a blind O(1) write whose folding happens lazily during -compaction — ideal for counters and edge weights. - -## Durability - -Every batch commits with an explicit policy: - -- `Durability::SyncWal` — write the WAL and fsync before returning (survives - power loss). -- `Durability::WalOnly` — write the WAL without forcing an fsync. -- `Durability::DisableWal` — skip the WAL. Use only for projections rebuildable - from another durable source of truth. - -## Key layout - -Every location lowers to a key built from length-prefixed segments -(`uvarint(len) ++ bytes`), which makes segment sequences self-delimiting: a -parent prefix only ever prefixes its own descendants, so sibling subtrees never -collide. Within a location prefix `P`: - -- `P` (exact) holds a `Leaf`/`Sum` value; -- `P ++ [0x01] ++ seg` holds child data (map entries, struct fields, elements); -- `P ++ [0x00] ++ seg` holds collection metadata (lengths, deque cursors). - -Deleting a subtree is a single RocksDB range delete over `[P, upper_bound(P))`. - -## What this is not - -- Not multi-process safe. One writer process; serialize writes at the app layer. -- Not distributed, not SQL. -- Map iteration order is encoded-byte order, not logical key order. -- On-disk struct field ids come from declaration order — add new fields at the - end; reordering changes the layout. -- Schema evolution is your responsibility. Because durable shines as a - *rebuildable projection*, the simplest migration is often to drop the data and - replay from your canonical log. - -## Testing - -```bash -cargo test -p durable -``` - -Covers the codec, the merge operator, every collection kind end-to-end, atomic -batches, durability modes, persistence across reopen, and property tests against -`BTreeMap`/`VecDeque`/sum-of-deltas models. diff --git a/durable/docs/design.md b/durable/docs/design.md deleted file mode 100644 index b2f63587c4ff395d49ab6b6bd73fed2e320002e6..0000000000000000000000000000000000000000 --- a/durable/docs/design.md +++ /dev/null @@ -1,93 +0,0 @@ -# durable — design notes - -This document describes how `durable` actually works, so the layout and cost -model are auditable rather than mysterious. - -## Goals - -1. **Precise updates.** A mutation touches only the keys it names. No - read-deserialize-mutate-reserialize-write of a whole struct. -2. **Deep nesting.** Maps, lists, deques, and structs compose to arbitrary - depth, all in one RocksDB column family. -3. **Type safety.** Illegal navigation and illegal operations fail to compile. -4. **Paths and mutations as data.** Addresses and edits are values you can - build, reuse, inspect, and apply in atomic batches. - -Non-goals: multi-process concurrency, distribution, SQL, ad-hoc range queries -over logical key order. - -## Key encoding - -A location is a sequence of **segments**. Each segment is length-prefixed: -`uvarint(len) ++ bytes`. The full RocksDB key is the concatenation of a parent -prefix and a one-byte discriminator plus a segment per step. - -Length-prefixing makes segment sequences *self-delimiting*: no segment can be a -byte-prefix of a different segment, so a parent prefix only ever prefixes its own -descendants. Sibling subtrees never overlap. - -Within a location prefix `P`: - -| Key | Holds | -|-----|-------| -| `P` (exact) | a `Leaf` / `Sum` scalar value | -| `P ++ [0x01] ++ seg` | child data: map entry, struct field, list/deque element | -| `P ++ [0x00] ++ seg` | collection metadata: list `len`, deque `head`/`tail` | - -- **Map** entry under key `k`: segment is `cbor(k)`. Iteration is a range scan - over `P ++ [0x01]`; logical keys are deduplicated by their first segment - (nested values contribute several physical keys sharing that segment). -- **List** element `i`: segment is `i` as 8 big-endian bytes; `len` lives in - metadata. -- **Deque** element `i` (an `i64`, possibly negative): segment is an - order-preserving encoding (`(i as u64) ^ (1<<63)` big-endian) so the byte order - matches signed numeric order. `head`/`tail` cursors live in metadata; both ends - are O(1) and never renumber. -- **Struct** field: segment is the field's declaration-order id as a uvarint. - -Deleting a subtree is one RocksDB range delete over `[P, prefix_upper_bound(P))` -(falling back to a scan only when the prefix is empty or all `0xff`). - -## Types and navigation - -`Path` carries the lowered prefix bytes and a phantom schema `S`. Navigation -methods are implemented per concrete schema, so `Path>` has `key`, -`Path>` has `at`, a derived struct's `Path` has its field navigators, and -so on. Each step appends a segment and returns a `Path` of the child schema. - -`#[derive(Durable)]` generates, for a struct, the `Schema` impl, a `{Name}Fields` -extension trait of navigators implemented for `Path`, and `Name::root()` / -`Name::namespaced(name)` constructors. - -## Mutations and the cost model - -Terminal mutating operations return reified `Write`s wrapping a plain-data `Op` -(`Put` / `Delete` / `DeletePrefix` / `Merge`). They are applied via `Db::apply` -or pushed onto a `Batch`, which commits as a single RocksDB write with an -explicit `Durability`. - -- **Blind** ops carry fully-determined keys and never read: `Leaf::set`/`delete`, - `Sum::add`/`set`/`delete`, collection `clear`. -- **Read-modify-write** ops read a length or cursor: list/deque pushes and pops. - In a `Batch`, appends are deferred and resolved at commit so contiguous appends - get contiguous indices and the whole batch is one atomic write. -- **Scans**: `keys`/`iter`/`len`/`contains`/`transform_values`. - -### Sum and the merge operator - -`Sum` is backed by a RocksDB associative merge operator registered at -`Db::open`. Accumulator values are stored tagged (`[type_tag, 8 LE bytes]`) so a -single operator folds `f64`, `i64`, and `u64` correctly. `add(delta)` is a blind -`Merge` write: O(1), no read, folded lazily during compaction. This is the right -primitive for counters and graph edge weights. - -## Durability and recovery - -`SyncWal` fsyncs the WAL before returning; `WalOnly` writes the WAL without an -fsync; `DisableWal` skips it. `DisableWal` is intended for projections that can -be rebuilt from another durable source of truth — its writes may be lost on an -unclean crash. - -`durable` is deliberately not a recovery plan on its own. Pair it with a -canonical log if you need crash recovery, and prefer "drop and replay" over -in-place migration when a schema changes. diff --git a/durable/examples/ranking.rs b/durable/examples/ranking.rs deleted file mode 100644 index 59425a8ece3556746c11fd3fda0339ec641f002a..0000000000000000000000000000000000000000 --- a/durable/examples/ranking.rs +++ /dev/null @@ -1,95 +0,0 @@ -//! A pairwise-ranking scope stored with precise, point-addressable updates. -//! -//! Run with: `cargo run -p durable --example ranking` -//! -//! This mirrors the motivating use case: a "scope" holds an edge-weight graph, a -//! capped window of recent votes, and a counter. A vote updates a handful of -//! keys in one atomic batch — it never reads or rewrites the whole scope. - -use durable::{Db, Deque, Durability, Durable, Leaf, Map, Sum}; -use serde::{Deserialize, Serialize}; - -#[derive(Serialize, Deserialize, Clone, Debug)] -struct Vote { - winner: u32, - loser: u32, - weight: f64, -} - -#[derive(Durable)] -#[allow(dead_code)] -struct Scope { - /// Directed edge weights: (from, to) -> accumulated weight. - edges: Map<(u32, u32), Sum>, - /// Most recent votes, newest at the back. - recent: Deque>, - /// Total votes recorded in this scope. - votes: Sum, -} - -#[derive(Durable)] -#[allow(dead_code)] -struct Store { - scopes: Map, -} - -const RECENT_CAP: u64 = 5; - -fn record_vote(db: &Db, scope: &str, vote: Vote) -> durable::Result<()> { - let s = Store::root().scopes().key(&scope.to_string()); - - // One atomic batch: bump the winning edge, flag the count, push the vote. - let mut batch = db.batch(); - batch.write(s.edges().key(&(vote.winner, vote.loser)).add(vote.weight)); - batch.write(s.votes().add(1)); - batch.push_back(&s.recent(), &vote)?; - batch.commit_with(Durability::SyncWal)?; - - // Keep only the most recent N votes (O(1) per eviction). - while s.recent().len(db)? > RECENT_CAP { - s.recent().pop_front(db)?; - } - Ok(()) -} - -fn main() -> durable::Result<()> { - let dir = tempfile::tempdir().unwrap(); - let db = Db::open(dir.path())?; - - for i in 0..8 { - let (winner, loser) = (i % 3, (i + 1) % 3); - record_vote( - &db, - "rust", - Vote { - winner, - loser, - weight: 1.0 + (i as f64) * 0.1, - }, - )?; - } - - let s = Store::root().scopes().key(&"rust".to_string()); - - println!("total votes: {}", s.votes().get(&db)?); - println!("recent window (cap {RECENT_CAP}): {}", s.recent().len(&db)?); - - let mut edges = s.edges().iter(&db)?; - edges.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap()); - println!("edges by weight:"); - for ((from, to), weight) in edges { - println!(" {from} -> {to}: {weight:.1}"); - } - - // Decay every edge by 10% in a single scan + atomic batch. - let decay = s - .edges() - .transform_values(&db, |_e, w| Some(w * 0.9))?; - db.apply(&decay, Durability::SyncWal)?; - println!( - "edge (0->1) after decay: {:.3}", - s.edges().key(&(0, 1)).get(&db)? - ); - - Ok(()) -} diff --git a/durable/src/codec.rs b/durable/src/codec.rs deleted file mode 100644 index e9ded9fd10153bb88c34e089e88258c3a0475793..0000000000000000000000000000000000000000 --- a/durable/src/codec.rs +++ /dev/null @@ -1,196 +0,0 @@ -//! Key encoding for durable paths. -//! -//! Every durable location lowers to a deterministic RocksDB key built from -//! *length-prefixed segments*. A segment is `uvarint(len) ++ bytes`, which makes -//! segment sequences self-delimiting: no segment can be a prefix of a *different* -//! segment, so sibling subtrees never overlap and a parent prefix only ever -//! prefixes its own descendants. -//! -//! Within a location prefix `P` we reserve a one-byte discriminator: -//! -//! - `P` (exact) → a [`crate::Leaf`] scalar value lives here. -//! - `P ++ [DATA] ++ seg` → a child (map entry, struct field, list element). -//! - `P ++ [META] ++ seg` → collection metadata (e.g. a list length). -//! -//! Because children and metadata live *under* `P`, deleting a whole subtree is a -//! single RocksDB range delete over `[P, prefix_upper_bound(P))`. - -/// Discriminator for child data living under a location. -pub const DATA: u8 = 0x01; -/// Discriminator for collection metadata living under a location. -pub const META: u8 = 0x00; - -/// Append an unsigned LEB128 varint to `out`. -pub fn put_uvarint(out: &mut Vec, mut value: u64) { - loop { - let mut byte = (value & 0x7f) as u8; - value >>= 7; - if value != 0 { - byte |= 0x80; - } - out.push(byte); - if value == 0 { - break; - } - } -} - -/// Decode an unsigned LEB128 varint from the front of `bytes`. -/// -/// Returns the value and the number of bytes consumed, or `None` if the input is -/// truncated or overlong. -pub fn read_uvarint(bytes: &[u8]) -> Option<(u64, usize)> { - let mut result: u64 = 0; - let mut shift = 0; - for (i, &byte) in bytes.iter().enumerate() { - if shift >= 64 { - return None; - } - result |= ((byte & 0x7f) as u64) << shift; - if byte & 0x80 == 0 { - return Some((result, i + 1)); - } - shift += 7; - } - None -} - -/// Append a length-prefixed segment to `out`. -pub fn put_segment(out: &mut Vec, bytes: &[u8]) { - put_uvarint(out, bytes.len() as u64); - out.extend_from_slice(bytes); -} - -/// Read one length-prefixed segment from the front of `bytes`. -/// -/// Returns the segment payload and the total number of bytes consumed -/// (including the length prefix). -pub fn read_segment(bytes: &[u8]) -> Option<(&[u8], usize)> { - let (len, header) = read_uvarint(bytes)?; - let len = len as usize; - let end = header.checked_add(len)?; - if end > bytes.len() { - return None; - } - Some((&bytes[header..end], header + len)) -} - -/// Build the key for child `seg` under location prefix `parent`. -pub fn child_key(parent: &[u8], seg: &[u8]) -> Vec { - let mut key = Vec::with_capacity(parent.len() + 2 + seg.len()); - key.extend_from_slice(parent); - key.push(DATA); - put_segment(&mut key, seg); - key -} - -/// The prefix under which all of `parent`'s child data lives. -pub fn child_scan_prefix(parent: &[u8]) -> Vec { - let mut key = Vec::with_capacity(parent.len() + 1); - key.extend_from_slice(parent); - key.push(DATA); - key -} - -/// Build a metadata key `name` under location prefix `parent`. -pub fn meta_key(parent: &[u8], name: &[u8]) -> Vec { - let mut key = Vec::with_capacity(parent.len() + 2 + name.len()); - key.extend_from_slice(parent); - key.push(META); - put_segment(&mut key, name); - key -} - -/// Smallest key strictly greater than every key prefixed by `prefix`. -/// -/// Returns `None` when `prefix` is empty or all `0xff` (i.e. the range extends to -/// the end of the keyspace), in which case callers must fall back to a scan. -pub fn prefix_upper_bound(prefix: &[u8]) -> Option> { - let mut end = prefix.to_vec(); - while let Some(last) = end.last_mut() { - if *last != 0xff { - *last += 1; - return Some(end); - } - end.pop(); - } - None -} - -/// Order-preserving encoding of an `i64` index (used by [`crate::Deque`]). -/// -/// Flipping the sign bit makes the unsigned big-endian byte order match signed -/// numeric order, so negative front indices sort before positive ones. -pub fn order_i64(index: i64) -> [u8; 8] { - ((index as u64) ^ (1u64 << 63)).to_be_bytes() -} - -/// Order-preserving encoding of a `u64` index (used by [`crate::List`]). -pub fn order_u64(index: u64) -> [u8; 8] { - index.to_be_bytes() -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn uvarint_roundtrip() { - for value in [0u64, 1, 127, 128, 300, 16384, u32::MAX as u64, u64::MAX] { - let mut buf = Vec::new(); - put_uvarint(&mut buf, value); - let (decoded, used) = read_uvarint(&buf).unwrap(); - assert_eq!(decoded, value); - assert_eq!(used, buf.len()); - } - } - - #[test] - fn read_uvarint_rejects_truncated() { - assert!(read_uvarint(&[0x80]).is_none()); - assert!(read_uvarint(&[]).is_none()); - } - - #[test] - fn segment_roundtrip_and_self_delimiting() { - let mut buf = Vec::new(); - put_segment(&mut buf, b"alpha"); - put_segment(&mut buf, b""); - put_segment(&mut buf, &[0x00, 0xff, 0x01]); - - let (a, n1) = read_segment(&buf).unwrap(); - assert_eq!(a, b"alpha"); - let (b, n2) = read_segment(&buf[n1..]).unwrap(); - assert_eq!(b, b""); - let (c, _) = read_segment(&buf[n1 + n2..]).unwrap(); - assert_eq!(c, &[0x00, 0xff, 0x01]); - } - - #[test] - fn segment_no_false_prefix() { - // seg("a") must not be a byte-prefix of seg("ab"): length-prefixing guards this. - let mut a = Vec::new(); - put_segment(&mut a, b"a"); - let mut ab = Vec::new(); - put_segment(&mut ab, b"ab"); - assert!(!ab.starts_with(&a)); - } - - #[test] - fn upper_bound_basics() { - assert_eq!(prefix_upper_bound(&[1, 2, 3]), Some(vec![1, 2, 4])); - assert_eq!(prefix_upper_bound(&[1, 2, 0xff]), Some(vec![1, 3])); - assert_eq!(prefix_upper_bound(&[0xff, 0xff]), None); - assert_eq!(prefix_upper_bound(&[]), None); - } - - #[test] - fn order_i64_is_monotonic() { - let mut values = [-5i64, -1, 0, 1, 5, i64::MIN, i64::MAX]; - values.sort(); - // Encoded byte order must match signed numeric order. - for pair in values.windows(2) { - assert!(order_i64(pair[0]) < order_i64(pair[1])); - } - } -} diff --git a/durable/src/lib.rs b/durable/src/lib.rs deleted file mode 100644 index b58fdcf4d5e85c6f34af10194155effa67ce200b..0000000000000000000000000000000000000000 --- a/durable/src/lib.rs +++ /dev/null @@ -1,435 +0,0 @@ -//! # durable -//! -//! Deeply nested, precisely updatable RocksDB-backed data structures for Rust, -//! built around **paths as data**. -//! -//! Instead of serializing a big struct into one blob, you describe your data with -//! a *schema* of composable types — [`Leaf`], [`Map`], [`List`], [`Deque`], -//! [`Sum`], and your own `#[derive(Durable)]` structs — and address any location -//! with a typed [`Path`]. A path lowers to a deterministic RocksDB key with no -//! I/O, so a mutation touches exactly the keys it names and nothing else. -//! -//! Terminal operations on a path return reified [`Write`] values (not side -//! effects). Compose several into one atomic [`Batch`] and commit them with an -//! explicit [`Durability`] policy. -//! -//! ``` -//! use durable::{Db, Durable, Durability, Leaf, Map, Sum}; -//! -//! #[derive(Durable)] -//! struct Store { -//! scores: Map>, -//! title: Leaf, -//! } -//! -//! // `#[derive(Durable)]` also generates a `StoreFields` navigator trait, -//! // in scope wherever `Store` is. -//! -//! # fn main() -> durable::Result<()> { -//! let dir = tempfile::tempdir().unwrap(); -//! let db = Db::open(dir.path())?; -//! -//! let root = Store::root(); -//! let alice = "alice".to_string(); -//! db.apply( -//! &[ -//! root.scores().key(&alice).add(10), // blind merge, no read -//! root.scores().key(&alice).add(5), -//! root.title().set(&"leaderboard".to_string()), -//! ], -//! Durability::SyncWal, -//! )?; -//! -//! assert_eq!(root.scores().key(&alice).get(&db)?, 15); -//! assert_eq!(root.title().get(&db)?, Some("leaderboard".to_string())); -//! # Ok(()) -//! # } -//! ``` - -mod codec; -mod path; -mod schema; - -use std::path::Path as FsPath; -use std::sync::Arc; - -use rocksdb::{Options, WriteBatch, WriteOptions, DB as RocksDb}; -use serde::{de::DeserializeOwned, Serialize}; -use thiserror::Error; - -pub use durable_derive::Durable; -pub use path::Path; -pub use schema::{Deque, Leaf, List, Map, Schema, Sum, Summable}; - -/// Errors returned by durable operations. -#[derive(Error, Debug)] -pub enum Error { - #[error("rocksdb error: {0}")] - RocksDb(#[from] rocksdb::Error), - #[error("serialization error: {0}")] - Serialize(String), - #[error("deserialization error: {0}")] - Deserialize(String), - #[error("data corruption: {0}")] - Corruption(String), -} - -/// Result alias used throughout the crate. -pub type Result = std::result::Result; - -/// CBOR-encode a value for leaf storage or key encoding. -pub(crate) fn encode_value(value: &T) -> Result> { - let mut bytes = Vec::new(); - ciborium::ser::into_writer(value, &mut bytes).map_err(|e| Error::Serialize(e.to_string()))?; - Ok(bytes) -} - -/// CBOR-decode a stored value. -pub(crate) fn decode_value(bytes: &[u8]) -> Result { - ciborium::de::from_reader(bytes).map_err(|e| Error::Deserialize(e.to_string())) -} - -/// Durability policy for a committed [`Batch`] or [`Db::apply`] call. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum Durability { - /// Write through the WAL and fsync it before returning (survives power loss). - SyncWal, - /// Write through the WAL without forcing an fsync. - WalOnly, - /// Skip the WAL entirely. Use only for projections rebuildable from another - /// durable source of truth. - DisableWal, -} - -/// A single reified storage operation. -/// -/// `Op` is the type-erased lowering of a typed terminal operation. It is plain -/// data: you can build, inspect, log, and store a list of ops, then apply them. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum Op { - /// Blind put of a value at an exact key. - Put { key: Vec, value: Vec }, - /// Blind delete of an exact key. - Delete { key: Vec }, - /// Delete every key under `prefix` (a whole subtree, including its root leaf). - DeletePrefix { prefix: Vec }, - /// Blind associative merge (used by [`Sum`]). - Merge { key: Vec, value: Vec }, -} - -/// A typed, reified mutation produced by a terminal path operation. -/// -/// A `Write` wraps a single [`Op`]. Collect several and hand them to -/// [`Db::apply`] (or push them onto a [`Batch`]) to commit atomically. -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct Write { - op: Op, -} - -impl Write { - pub(crate) fn new(op: Op) -> Self { - Self { op } - } - - /// The underlying reified operation. - pub fn op(&self) -> &Op { - &self.op - } - - /// Consume the write, yielding its operation. - pub fn into_op(self) -> Op { - self.op - } -} - -/// A handle to an open durable database. -/// -/// Cheap to clone (an `Arc` around the RocksDB handle). Durable assumes a single -/// writer process; serialize writes at the application layer. -#[derive(Clone)] -pub struct Db { - inner: Arc, -} - -impl Db { - /// Open (or create) a durable database at `path`. - pub fn open>(path: P) -> Result { - let mut opts = Options::default(); - opts.create_if_missing(true); - opts.set_merge_operator_associative("durable.sum", schema::sum_merge); - let db = RocksDb::open(&opts, path)?; - Ok(Self { - inner: Arc::new(db), - }) - } - - pub(crate) fn raw(&self) -> &RocksDb { - &self.inner - } - - /// Start an atomic batch of writes. - pub fn batch(&self) -> Batch { - Batch::new(self.clone()) - } - - /// Apply reified writes atomically with the given durability policy. - pub fn apply(&self, writes: &[Write], durability: Durability) -> Result<()> { - let mut batch = self.batch(); - for write in writes { - batch.write(write.clone()); - } - batch.commit_with(durability) - } - - /// Apply a single write with the given durability policy. - pub fn run(&self, write: Write, durability: Durability) -> Result<()> { - self.apply(std::slice::from_ref(&write), durability) - } -} - -/// An atomic batch of writes plus deferred collection appends. -/// -/// Blind writes are recorded immediately. Stateful appends ([`Batch::push_back`] -/// / [`Batch::push`]) are resolved at commit time against the current collection -/// length, so several appends in one batch land at contiguous indices and the -/// whole batch commits as one RocksDB write (one WAL flush). -pub struct Batch { - db: Db, - inner: WriteBatch, - prefix_deletes: Vec>, - appends: Vec, -} - -pub(crate) enum AppendEnd { - /// Append to the tail of a [`List`]; counter is `len`, index is `len`. - ListBack, - /// Append to the back of a [`Deque`]; counter is `tail`, index is `tail`. - DequeBack, - /// Push to the front of a [`Deque`]; counter is `head`, index is `head - 1`. - DequeFront, -} - -pub(crate) struct PendingAppend { - pub(crate) coll_prefix: Vec, - pub(crate) end: AppendEnd, - pub(crate) value: Vec, -} - -impl Batch { - fn new(db: Db) -> Self { - Self { - db, - inner: WriteBatch::default(), - prefix_deletes: Vec::new(), - appends: Vec::new(), - } - } - - /// Record a reified write in this batch. - pub fn write(&mut self, write: Write) { - match write.into_op() { - Op::Put { key, value } => self.inner.put(key, value), - Op::Delete { key } => self.inner.delete(key), - Op::Merge { key, value } => self.inner.merge(key, value), - Op::DeletePrefix { prefix } => self.prefix_deletes.push(prefix), - } - } - - /// Record several reified writes. - pub fn extend>(&mut self, writes: I) { - for write in writes { - self.write(write); - } - } - - pub(crate) fn raw_put(&mut self, key: Vec, value: Vec) { - self.inner.put(key, value); - } - - /// Append `value` to the back of a leaf list (resolved at commit). - pub fn push(&mut self, list: &Path>>, value: &T) -> Result<()> { - self.appends.push(PendingAppend { - coll_prefix: list.prefix().to_vec(), - end: AppendEnd::ListBack, - value: encode_value(value)?, - }); - Ok(()) - } - - /// Append `value` to the back of a leaf deque (resolved at commit). - pub fn push_back( - &mut self, - deque: &Path>>, - value: &T, - ) -> Result<()> { - self.appends.push(PendingAppend { - coll_prefix: deque.prefix().to_vec(), - end: AppendEnd::DequeBack, - value: encode_value(value)?, - }); - Ok(()) - } - - /// Push `value` to the front of a leaf deque (resolved at commit). - pub fn push_front( - &mut self, - deque: &Path>>, - value: &T, - ) -> Result<()> { - self.appends.push(PendingAppend { - coll_prefix: deque.prefix().to_vec(), - end: AppendEnd::DequeFront, - value: encode_value(value)?, - }); - Ok(()) - } - - /// Commit the batch, fsyncing the WAL (equivalent to - /// `commit_with(Durability::SyncWal)`). - pub fn commit(self) -> Result<()> { - self.commit_with(Durability::SyncWal) - } - - /// Commit the batch with an explicit durability policy. - pub fn commit_with(mut self, durability: Durability) -> Result<()> { - self.resolve_prefix_deletes()?; - self.resolve_appends()?; - - match durability { - Durability::SyncWal => { - self.db.raw().write(self.inner)?; - self.db.raw().flush_wal(true)?; - } - Durability::WalOnly => { - self.db.raw().write(self.inner)?; - } - Durability::DisableWal => { - let mut opts = WriteOptions::default(); - opts.disable_wal(true); - self.db.raw().write_opt(self.inner, &opts)?; - } - } - Ok(()) - } - - fn resolve_prefix_deletes(&mut self) -> Result<()> { - for prefix in std::mem::take(&mut self.prefix_deletes) { - match codec::prefix_upper_bound(&prefix) { - Some(end) => self.inner.delete_range(&prefix, &end), - None => { - // Range extends to the end of the keyspace: scan and delete. - let iter = self.db.raw().iterator(rocksdb::IteratorMode::From( - &prefix, - rocksdb::Direction::Forward, - )); - for item in iter { - let (key, _) = item?; - if !key.starts_with(&prefix) { - break; - } - self.inner.delete(&key); - } - } - } - } - Ok(()) - } - - fn resolve_appends(&mut self) -> Result<()> { - use std::collections::HashMap; - // Group appends by (collection, end) so contiguous appends get contiguous - // indices and each counter is read exactly once. - let mut order: Vec<(Vec, u8)> = Vec::new(); - let mut grouped: HashMap<(Vec, u8), Vec>> = HashMap::new(); - for append in std::mem::take(&mut self.appends) { - let tag = match append.end { - AppendEnd::ListBack => 0u8, - AppendEnd::DequeBack => 1u8, - AppendEnd::DequeFront => 2u8, - }; - let group_key = (append.coll_prefix, tag); - let entry = grouped.entry(group_key.clone()).or_insert_with(|| { - order.push(group_key); - Vec::new() - }); - entry.push(append.value); - } - - for (coll_prefix, tag) in order { - let values = grouped.remove(&(coll_prefix.clone(), tag)).unwrap(); - match tag { - 0 => self.resolve_list_back(&coll_prefix, values)?, - 1 => self.resolve_deque_end(&coll_prefix, values, true)?, - 2 => self.resolve_deque_end(&coll_prefix, values, false)?, - _ => unreachable!(), - } - } - Ok(()) - } - - fn resolve_list_back(&mut self, coll_prefix: &[u8], values: Vec>) -> Result<()> { - let len_key = codec::meta_key(coll_prefix, b"len"); - let mut len = read_u64(&self.db, &len_key)?.unwrap_or(0); - for value in values { - let elem = codec::child_key(coll_prefix, &codec::order_u64(len)); - self.inner.put(&elem, &value); - len += 1; - } - self.inner.put(&len_key, len.to_le_bytes()); - Ok(()) - } - - fn resolve_deque_end( - &mut self, - coll_prefix: &[u8], - values: Vec>, - back: bool, - ) -> Result<()> { - let head_key = codec::meta_key(coll_prefix, b"head"); - let tail_key = codec::meta_key(coll_prefix, b"tail"); - let mut head = read_i64(&self.db, &head_key)?.unwrap_or(0); - let mut tail = read_i64(&self.db, &tail_key)?.unwrap_or(0); - for value in values { - if back { - let elem = codec::child_key(coll_prefix, &codec::order_i64(tail)); - self.inner.put(&elem, &value); - tail += 1; - } else { - head -= 1; - let elem = codec::child_key(coll_prefix, &codec::order_i64(head)); - self.inner.put(&elem, &value); - } - } - self.inner.put(&head_key, head.to_le_bytes()); - self.inner.put(&tail_key, tail.to_le_bytes()); - Ok(()) - } -} - -pub(crate) fn read_u64(db: &Db, key: &[u8]) -> Result> { - match db.raw().get(key)? { - Some(bytes) => { - if bytes.len() != 8 { - return Err(Error::Corruption("expected 8-byte u64 meta".into())); - } - let mut buf = [0u8; 8]; - buf.copy_from_slice(&bytes); - Ok(Some(u64::from_le_bytes(buf))) - } - None => Ok(None), - } -} - -pub(crate) fn read_i64(db: &Db, key: &[u8]) -> Result> { - match db.raw().get(key)? { - Some(bytes) => { - if bytes.len() != 8 { - return Err(Error::Corruption("expected 8-byte i64 meta".into())); - } - let mut buf = [0u8; 8]; - buf.copy_from_slice(&bytes); - Ok(Some(i64::from_le_bytes(buf))) - } - None => Ok(None), - } -} diff --git a/durable/src/path.rs b/durable/src/path.rs deleted file mode 100644 index b2e8bfa4895b1be3a2e44eaf14cda31b443d0b81..0000000000000000000000000000000000000000 --- a/durable/src/path.rs +++ /dev/null @@ -1,575 +0,0 @@ -//! Typed paths: composable, data-only addresses into a durable schema. -//! -//! A [`Path`] is just a byte prefix plus a phantom schema type. Navigation -//! methods are gated by the concrete schema, so only legal steps compile, and -//! terminal operations return reified [`Write`]s (for mutations) or read directly -//! from a [`Db`]. - -use std::marker::PhantomData; - -use serde::{de::DeserializeOwned, Serialize}; - -use crate::{ - codec, - schema::{decode_sum, encode_sum, Deque, Leaf, List, Map, Schema, Sum, Summable}, - decode_value, encode_value, read_i64, read_u64, Db, Error, Op, Result, Write, -}; - -/// A typed address into a durable schema. -/// -/// Cheap to clone; carries only the lowered key prefix. Construct the root of a -/// schema with [`Path::root`] (typically via the `#[derive(Durable)]`-generated -/// `S::root()`), then navigate with schema-specific methods. -pub struct Path { - prefix: Vec, - _schema: PhantomData S>, -} - -impl Clone for Path { - fn clone(&self) -> Self { - Self { - prefix: self.prefix.clone(), - _schema: PhantomData, - } - } -} - -impl std::fmt::Debug for Path { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("Path") - .field("schema", &std::any::type_name::()) - .field("prefix", &self.prefix) - .finish() - } -} - -impl Path { - /// The empty-prefixed root of a schema. - /// - /// One root schema per database. Navigate from here. - pub fn root() -> Self { - Self::from_prefix(Vec::new()) - } - - /// A root namespaced under `name`, so multiple schemas can share one database. - pub fn namespaced(name: &str) -> Self { - let mut prefix = Vec::new(); - codec::put_segment(&mut prefix, name.as_bytes()); - Self::from_prefix(prefix) - } -} - -impl Path { - pub(crate) fn from_prefix(prefix: Vec) -> Self { - Self { - prefix, - _schema: PhantomData, - } - } - - /// The lowered RocksDB key prefix this path addresses. - pub fn prefix(&self) -> &[u8] { - &self.prefix - } - - fn child(&self, seg: &[u8]) -> Path { - Path::from_prefix(codec::child_key(&self.prefix, seg)) - } - - /// Navigate to a `#[derive(Durable)]` struct field. Called by generated code. - #[doc(hidden)] - pub fn child_field(&self, field_id: u32) -> Path { - let mut seg = Vec::new(); - codec::put_uvarint(&mut seg, field_id as u64); - self.child(&seg) - } -} - -// --------------------------------------------------------------------------- -// Leaf -// --------------------------------------------------------------------------- - -impl Path> { - /// Read the value at this leaf, if present. - pub fn get(&self, db: &Db) -> Result> { - match db.raw().get(&self.prefix)? { - Some(bytes) => Ok(Some(decode_value(&bytes)?)), - None => Ok(None), - } - } - - /// A reified blind write that sets this leaf to `value`. - pub fn set(&self, value: &T) -> Write { - let bytes = encode_value(value).expect("durable: leaf value serialization failed"); - Write::new(Op::Put { - key: self.prefix.clone(), - value: bytes, - }) - } - - /// A reified blind write that removes this leaf. - pub fn delete(&self) -> Write { - Write::new(Op::Delete { - key: self.prefix.clone(), - }) - } -} - -// --------------------------------------------------------------------------- -// Sum -// --------------------------------------------------------------------------- - -impl Path> { - /// Read the accumulated value (defaults to zero when absent). - pub fn get(&self, db: &Db) -> Result { - match db.raw().get(&self.prefix)? { - Some(bytes) => decode_sum::(&bytes) - .ok_or_else(|| Error::Corruption("malformed Sum accumulator".into())), - None => Ok(N::zero()), - } - } - - /// A reified blind merge that adds `delta` to the accumulator. - /// - /// This never reads the current value: it is an O(1) write whose effect is - /// resolved lazily by RocksDB's merge operator. - pub fn add(&self, delta: N) -> Write { - Write::new(Op::Merge { - key: self.prefix.clone(), - value: encode_sum(delta), - }) - } - - /// A reified blind write that sets the accumulator to an exact value. - pub fn set(&self, value: N) -> Write { - Write::new(Op::Put { - key: self.prefix.clone(), - value: encode_sum(value), - }) - } - - /// A reified blind write that removes the accumulator. - pub fn delete(&self) -> Write { - Write::new(Op::Delete { - key: self.prefix.clone(), - }) - } -} - -// --------------------------------------------------------------------------- -// Map -// --------------------------------------------------------------------------- - -impl Path> { - /// Navigate to the sub-schema stored under `key`. - pub fn key(&self, key: &K) -> Path { - let encoded = encode_value(key).expect("durable: map key serialization failed"); - self.child(&encoded) - } - - /// A reified write that deletes the entire map (all entries and metadata). - pub fn clear(&self) -> Write { - Write::new(Op::DeletePrefix { - prefix: self.prefix.clone(), - }) - } -} - -impl Path> { - /// All keys present in the map, in stored (encoded-byte) order. - pub fn keys(&self, db: &Db) -> Result> { - let scan = codec::child_scan_prefix(&self.prefix); - let iter = db - .raw() - .iterator(rocksdb::IteratorMode::From(&scan, rocksdb::Direction::Forward)); - let mut keys = Vec::new(); - let mut last: Option> = None; - for item in iter { - let (db_key, _) = item?; - if !db_key.starts_with(&scan) { - break; - } - let rest = &db_key[scan.len()..]; - let (key_seg, _) = codec::read_segment(rest) - .ok_or_else(|| Error::Corruption("malformed map entry key".into()))?; - if last.as_deref() == Some(key_seg) { - continue; // same logical key, deeper sub-key - } - last = Some(key_seg.to_vec()); - keys.push(decode_value(key_seg)?); - } - Ok(keys) - } - - /// The number of distinct keys in the map. - pub fn len(&self, db: &Db) -> Result { - Ok(self.keys(db)?.len()) - } - - /// Whether the map has no entries. - pub fn is_empty(&self, db: &Db) -> Result { - let scan = codec::child_scan_prefix(&self.prefix); - let mut iter = db - .raw() - .iterator(rocksdb::IteratorMode::From(&scan, rocksdb::Direction::Forward)); - match iter.next() { - Some(item) => { - let (db_key, _) = item?; - Ok(!db_key.starts_with(&scan)) - } - None => Ok(true), - } - } - - /// Whether `key` is present. - pub fn contains(&self, db: &Db, key: &K) -> Result { - let child = self.key(key); - // A present entry has at least one key at-or-under the child prefix. - let mut iter = db.raw().iterator(rocksdb::IteratorMode::From( - child.prefix(), - rocksdb::Direction::Forward, - )); - match iter.next() { - Some(item) => { - let (db_key, _) = item?; - Ok(db_key.starts_with(child.prefix())) - } - None => Ok(false), - } - } - - /// All keys paired with sub-paths into their values (composable navigation). - pub fn entries(&self, db: &Db) -> Result)>> { - let keys = self.keys(db)?; - Ok(keys - .into_iter() - .map(|k| { - let path = self.key(&k); - (k, path) - }) - .collect()) - } -} - -// Leaf-valued maps gain direct value iteration and bulk transforms. -impl Path>> { - /// Read the value stored under `key`. - pub fn get(&self, db: &Db, key: &K) -> Result> { - self.key(key).get(db) - } - - /// All `(key, value)` pairs in stored order. - pub fn iter(&self, db: &Db) -> Result> { - let scan = codec::child_scan_prefix(&self.prefix); - let iter = db - .raw() - .iterator(rocksdb::IteratorMode::From(&scan, rocksdb::Direction::Forward)); - let mut out = Vec::new(); - for item in iter { - let (db_key, value) = item?; - if !db_key.starts_with(&scan) { - break; - } - let rest = &db_key[scan.len()..]; - let (key_seg, used) = codec::read_segment(rest) - .ok_or_else(|| Error::Corruption("malformed map entry key".into()))?; - // Leaf entries are exactly one physical key; reject deeper sub-keys. - if used != rest.len() { - return Err(Error::Corruption("unexpected nested key in leaf map".into())); - } - out.push((decode_value(key_seg)?, decode_value(&value)?)); - } - Ok(out) - } - - /// Build reified writes that rewrite each value through `f`. - /// - /// Returning `Some(new)` sets the value, `None` deletes the entry. This reads - /// the map once (a prefix scan) and yields blind writes, so the whole - /// transform commits atomically in one batch (e.g. "decay every edge weight"). - pub fn transform_values( - &self, - db: &Db, - mut f: impl FnMut(&K, T) -> Option, - ) -> Result> { - let mut writes = Vec::new(); - for (k, v) in self.iter(db)? { - let entry = self.key(&k); - match f(&k, v) { - Some(new) => writes.push(entry.set(&new)), - None => writes.push(entry.delete()), - } - } - Ok(writes) - } -} - -// Sum-valued maps gain direct accumulator iteration and bulk transforms. -impl Path>> { - /// Read the accumulator stored under `key` (zero when absent). - pub fn get(&self, db: &Db, key: &K) -> Result { - self.key(key).get(db) - } - - /// All `(key, value)` accumulator pairs in stored order. - pub fn iter(&self, db: &Db) -> Result> { - let scan = codec::child_scan_prefix(&self.prefix); - let iter = db - .raw() - .iterator(rocksdb::IteratorMode::From(&scan, rocksdb::Direction::Forward)); - let mut out = Vec::new(); - for item in iter { - let (db_key, value) = item?; - if !db_key.starts_with(&scan) { - break; - } - let rest = &db_key[scan.len()..]; - let (key_seg, used) = codec::read_segment(rest) - .ok_or_else(|| Error::Corruption("malformed map entry key".into()))?; - if used != rest.len() { - return Err(Error::Corruption("unexpected nested key in sum map".into())); - } - let n = decode_sum::(&value) - .ok_or_else(|| Error::Corruption("malformed Sum accumulator".into()))?; - out.push((decode_value(key_seg)?, n)); - } - Ok(out) - } - - /// Build reified writes that rewrite each accumulator through `f`. - /// - /// `Some(new)` sets the accumulator (blind put), `None` deletes it. Reads the - /// map once, yields blind writes — ideal for "decay every edge weight". - pub fn transform_values( - &self, - db: &Db, - mut f: impl FnMut(&K, N) -> Option, - ) -> Result> { - let mut writes = Vec::new(); - for (k, v) in self.iter(db)? { - let entry = self.key(&k); - match f(&k, v) { - Some(new) => writes.push(entry.set(new)), - None => writes.push(entry.delete()), - } - } - Ok(writes) - } -} - -// --------------------------------------------------------------------------- -// List -// --------------------------------------------------------------------------- - -impl Path> { - /// Navigate to the element at `index` (no bounds check until read). - pub fn at(&self, index: u64) -> Path { - self.child(&codec::order_u64(index)) - } - - /// The number of elements. - pub fn len(&self, db: &Db) -> Result { - Ok(read_u64(db, &codec::meta_key(&self.prefix, b"len"))?.unwrap_or(0)) - } - - /// Whether the list is empty. - pub fn is_empty(&self, db: &Db) -> Result { - Ok(self.len(db)? == 0) - } - - /// A reified write that deletes the whole list (elements and length). - pub fn clear(&self) -> Write { - Write::new(Op::DeletePrefix { - prefix: self.prefix.clone(), - }) - } -} - -impl Path>> { - /// Read the element at `index`. - pub fn get(&self, db: &Db, index: u64) -> Result> { - if index >= self.len(db)? { - return Ok(None); - } - self.at(index).get(db) - } - - /// Append `value`, returning its index. Commits with `SyncWal`. - pub fn push(&self, db: &Db, value: &T) -> Result { - let mut batch = db.batch(); - let index = self.len(db)?; - batch.push(self, value)?; - batch.commit()?; - Ok(index) - } - - /// Remove and return the last element. Commits with `SyncWal`. - pub fn pop(&self, db: &Db) -> Result> { - let len = self.len(db)?; - if len == 0 { - return Ok(None); - } - let last = len - 1; - let value = self.at(last).get(db)?; - let mut batch = db.batch(); - batch.write(self.at(last).delete()); - batch.raw_put(codec::meta_key(&self.prefix, b"len"), last.to_le_bytes().to_vec()); - batch.commit()?; - Ok(value) - } - - /// All elements in index order. - pub fn iter(&self, db: &Db) -> Result> { - let len = self.len(db)?; - let mut out = Vec::with_capacity(len as usize); - for i in 0..len { - match self.at(i).get(db)? { - Some(v) => out.push(v), - None => return Err(Error::Corruption("list element missing below len".into())), - } - } - Ok(out) - } -} - -// --------------------------------------------------------------------------- -// Deque -// --------------------------------------------------------------------------- - -impl Path> { - fn head(&self, db: &Db) -> Result { - Ok(read_i64(db, &codec::meta_key(&self.prefix, b"head"))?.unwrap_or(0)) - } - - fn tail(&self, db: &Db) -> Result { - Ok(read_i64(db, &codec::meta_key(&self.prefix, b"tail"))?.unwrap_or(0)) - } - - /// The number of elements. - pub fn len(&self, db: &Db) -> Result { - Ok((self.tail(db)? - self.head(db)?).max(0) as u64) - } - - /// Whether the deque is empty. - pub fn is_empty(&self, db: &Db) -> Result { - Ok(self.len(db)? == 0) - } - - /// A reified write that deletes the whole deque (elements and metadata). - pub fn clear(&self) -> Write { - Write::new(Op::DeletePrefix { - prefix: self.prefix.clone(), - }) - } -} - -impl Path>> { - /// Push to the back. Commits with `SyncWal`. - pub fn push_back(&self, db: &Db, value: &T) -> Result<()> { - let mut batch = db.batch(); - batch.push_back(self, value)?; - batch.commit() - } - - /// Push to the front. Commits with `SyncWal`. - pub fn push_front(&self, db: &Db, value: &T) -> Result<()> { - let mut batch = db.batch(); - batch.push_front(self, value)?; - batch.commit() - } - - /// Remove and return the front element. Commits with `SyncWal`. - pub fn pop_front(&self, db: &Db) -> Result> { - let head = self.head(db)?; - let tail = self.tail(db)?; - if head >= tail { - return Ok(None); - } - let value = self.child::>(&codec::order_i64(head)).get(db)?; - let mut batch = db.batch(); - batch.write(self.child::>(&codec::order_i64(head)).delete()); - batch.raw_put( - codec::meta_key(&self.prefix, b"head"), - (head + 1).to_le_bytes().to_vec(), - ); - batch.commit()?; - Ok(value) - } - - /// Remove and return the back element. Commits with `SyncWal`. - pub fn pop_back(&self, db: &Db) -> Result> { - let head = self.head(db)?; - let tail = self.tail(db)?; - if head >= tail { - return Ok(None); - } - let last = tail - 1; - let value = self.child::>(&codec::order_i64(last)).get(db)?; - let mut batch = db.batch(); - batch.write(self.child::>(&codec::order_i64(last)).delete()); - batch.raw_put( - codec::meta_key(&self.prefix, b"tail"), - last.to_le_bytes().to_vec(), - ); - batch.commit()?; - Ok(value) - } - - /// Read the front element without removing it. - pub fn front(&self, db: &Db) -> Result> { - let head = self.head(db)?; - if head >= self.tail(db)? { - return Ok(None); - } - self.child::>(&codec::order_i64(head)).get(db) - } - - /// Read the back element without removing it. - pub fn back(&self, db: &Db) -> Result> { - let tail = self.tail(db)?; - if self.head(db)? >= tail { - return Ok(None); - } - self.child::>(&codec::order_i64(tail - 1)).get(db) - } - - /// All elements from front to back. - pub fn iter(&self, db: &Db) -> Result> { - let head = self.head(db)?; - let tail = self.tail(db)?; - let mut out = Vec::with_capacity((tail - head).max(0) as usize); - for idx in head..tail { - match self.child::>(&codec::order_i64(idx)).get(db)? { - Some(v) => out.push(v), - None => return Err(Error::Corruption("deque element missing in range".into())), - } - } - Ok(out) - } - - /// Drop elements from the back until the length is at most `max_len`, - /// committing with the given durability. A no-op when already short enough. - pub fn truncate_back( - &self, - db: &Db, - max_len: u64, - durability: crate::Durability, - ) -> Result<()> { - let head = self.head(db)?; - let tail = self.tail(db)?; - let len = (tail - head).max(0) as u64; - if len <= max_len { - return Ok(()); - } - let new_tail = tail - (len - max_len) as i64; - let mut batch = db.batch(); - for idx in new_tail..tail { - batch.write(self.child::>(&codec::order_i64(idx)).delete()); - } - batch.raw_put( - codec::meta_key(&self.prefix, b"tail"), - new_tail.to_le_bytes().to_vec(), - ); - batch.commit_with(durability) - } -} diff --git a/durable/src/schema.rs b/durable/src/schema.rs deleted file mode 100644 index f804939aee51683f5773f2415e122ed64ea40020..0000000000000000000000000000000000000000 --- a/durable/src/schema.rs +++ /dev/null @@ -1,219 +0,0 @@ -//! Type-level schema markers. -//! -//! A *schema* describes the shape of a durable location at the type level. It is -//! never instantiated; it only parameterizes a [`crate::Path`] so the compiler -//! knows which navigation steps and terminal operations are legal. -//! -//! - [`Leaf`] — a single CBOR-encoded scalar value. -//! - [`Map`] — keys of type `K` to sub-schema `V`. -//! - [`List`] — an index-addressed sequence of sub-schema `V`. -//! - [`Deque`] — a double-ended queue of sub-schema `V` (O(1) ends). -//! - [`Sum`] — a numeric accumulator updated with blind merge writes. -//! - any `#[derive(Durable)]` struct — a fixed set of named fields. - -use std::marker::PhantomData; - -/// Marker trait implemented by every durable schema. -/// -/// Implemented for [`Leaf`], [`Map`], [`List`], [`Deque`], [`Sum`], and by -/// `#[derive(Durable)]` for user structs. It is intentionally minimal; behaviour -/// lives on `Path` impls keyed by the concrete schema. -pub trait Schema {} - -/// A single CBOR-encoded scalar value of type `T`. -pub struct Leaf(PhantomData); -impl Schema for Leaf {} - -/// A map from keys of type `K` to sub-schema `V`. -pub struct Map(PhantomData<(K, V)>); -impl Schema for Map {} - -/// An index-addressed growable sequence of sub-schema `V`. -pub struct List(PhantomData); -impl Schema for List {} - -/// A double-ended queue of sub-schema `V` with O(1) push/pop at both ends. -pub struct Deque(PhantomData); -impl Schema for Deque {} - -/// A numeric accumulator. Updated with blind, associative merge writes so -/// incrementing is O(1) and never reads the current value. -pub struct Sum(PhantomData); -impl Schema for Sum {} - -/// Numbers that can back a [`Sum`] accumulator. -/// -/// Stored on disk as `[TAG, b0..b7]`: a one-byte type tag plus the 8-byte -/// little-endian payload. The tag lets a single RocksDB merge operator fold -/// `f64` and `i64` accumulators correctly. -pub trait Summable: Copy + 'static { - /// Disk type tag, unique per numeric type. - const TAG: u8; - /// Additive identity. - fn zero() -> Self; - /// Combine two values (sum). - fn combine(self, other: Self) -> Self; - /// Little-endian 8-byte payload. - fn to_le_payload(self) -> [u8; 8]; - /// Decode from a little-endian 8-byte payload. - fn from_le_payload(bytes: [u8; 8]) -> Self; -} - -impl Summable for f64 { - const TAG: u8 = 0; - fn zero() -> Self { - 0.0 - } - fn combine(self, other: Self) -> Self { - self + other - } - fn to_le_payload(self) -> [u8; 8] { - self.to_le_bytes() - } - fn from_le_payload(bytes: [u8; 8]) -> Self { - f64::from_le_bytes(bytes) - } -} - -impl Summable for i64 { - const TAG: u8 = 1; - fn zero() -> Self { - 0 - } - fn combine(self, other: Self) -> Self { - self.wrapping_add(other) - } - fn to_le_payload(self) -> [u8; 8] { - self.to_le_bytes() - } - fn from_le_payload(bytes: [u8; 8]) -> Self { - i64::from_le_bytes(bytes) - } -} - -impl Summable for u64 { - const TAG: u8 = 2; - fn zero() -> Self { - 0 - } - fn combine(self, other: Self) -> Self { - self.wrapping_add(other) - } - fn to_le_payload(self) -> [u8; 8] { - self.to_le_bytes() - } - fn from_le_payload(bytes: [u8; 8]) -> Self { - u64::from_le_bytes(bytes) - } -} - -/// Encode a `Summable` to its tagged on-disk form `[TAG, b0..b7]`. -pub(crate) fn encode_sum(value: N) -> Vec { - let mut out = Vec::with_capacity(9); - out.push(N::TAG); - out.extend_from_slice(&value.to_le_payload()); - out -} - -/// Decode a tagged accumulator payload back to `N`, validating the tag. -pub(crate) fn decode_sum(bytes: &[u8]) -> Option { - if bytes.len() != 9 || bytes[0] != N::TAG { - return None; - } - let mut payload = [0u8; 8]; - payload.copy_from_slice(&bytes[1..9]); - Some(N::from_le_payload(payload)) -} - -/// Fold one tagged operand into a running tagged accumulator. -/// -/// Used by the RocksDB merge operator. Operands of mismatched tags are skipped -/// rather than panicking, keeping compaction resilient to stray bytes. -fn fold_tagged(acc: &mut Option<[u8; 9]>, operand: &[u8]) { - if operand.len() != 9 { - return; - } - let tag = operand[0]; - let mut op_payload = [0u8; 8]; - op_payload.copy_from_slice(&operand[1..9]); - - match acc { - Some(existing) if existing[0] == tag => { - let mut acc_payload = [0u8; 8]; - acc_payload.copy_from_slice(&existing[1..9]); - let combined = match tag { - 0 => f64::from_le_bytes(acc_payload) - .combine(f64::from_le_bytes(op_payload)) - .to_le_payload(), - 1 => i64::from_le_bytes(acc_payload) - .combine(i64::from_le_bytes(op_payload)) - .to_le_payload(), - 2 => u64::from_le_bytes(acc_payload) - .combine(u64::from_le_bytes(op_payload)) - .to_le_payload(), - _ => return, - }; - existing[1..9].copy_from_slice(&combined); - } - Some(_) => {} // tag mismatch: ignore stray operand - None => { - let mut start = [0u8; 9]; - start[0] = tag; - start[1..9].copy_from_slice(&op_payload); - *acc = Some(start); - } - } -} - -/// Associative merge operator registered on every durable database so that -/// [`Sum`] accumulators can be incremented with blind `merge` writes. -pub(crate) fn sum_merge( - _key: &[u8], - existing: Option<&[u8]>, - operands: &rocksdb::MergeOperands, -) -> Option> { - let mut acc: Option<[u8; 9]> = None; - if let Some(existing) = existing { - if existing.len() == 9 { - let mut start = [0u8; 9]; - start.copy_from_slice(existing); - acc = Some(start); - } - } - for operand in operands.iter() { - fold_tagged(&mut acc, operand); - } - acc.map(|bytes| bytes.to_vec()) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn sum_roundtrip_tagged() { - assert_eq!(decode_sum::(&encode_sum(2.5f64)), Some(2.5)); - assert_eq!(decode_sum::(&encode_sum(-7i64)), Some(-7)); - assert_eq!(decode_sum::(&encode_sum(9u64)), Some(9)); - // Tag mismatch is rejected. - assert_eq!(decode_sum::(&encode_sum(2.5f64)), None); - } - - #[test] - fn fold_accumulates_same_tag() { - let mut acc = None; - fold_tagged(&mut acc, &encode_sum(1.5f64)); - fold_tagged(&mut acc, &encode_sum(2.0f64)); - let bytes = acc.unwrap(); - assert_eq!(decode_sum::(&bytes), Some(3.5)); - } - - #[test] - fn fold_skips_mismatched_tag() { - let mut acc = None; - fold_tagged(&mut acc, &encode_sum(5i64)); - fold_tagged(&mut acc, &encode_sum(1.0f64)); // ignored - let bytes = acc.unwrap(); - assert_eq!(decode_sum::(&bytes), Some(5)); - } -} diff --git a/durable/tests/integration.rs b/durable/tests/integration.rs deleted file mode 100644 index 6aef7049f5ae442c5153dddb37523308d77503f6..0000000000000000000000000000000000000000 --- a/durable/tests/integration.rs +++ /dev/null @@ -1,433 +0,0 @@ -//! End-to-end tests for the durable paths-as-data API. - -use durable::{Db, Durability, Durable, Deque, Leaf, List, Map, Op, Sum}; -use serde::{Deserialize, Serialize}; -use tempfile::TempDir; - -#[derive(Serialize, Deserialize, Clone, PartialEq, Debug)] -struct Vote { - a: String, - b: String, - ratio: i32, -} - -/// A scope's local ranking state — the kind of thing that used to be one CBOR -/// blob, now addressable field-by-field and key-by-key. -#[derive(Durable)] -#[allow(dead_code)] -struct GroupState { - edges: Map<(u32, u32), Sum>, - voted_pairs: Map<(u32, u32), Leaf>, - recent_votes: Deque>, - item_count: Sum, -} - -#[derive(Durable)] -#[allow(dead_code)] -struct Store { - scopes: Map, - nodes: Map>, - log: List>, -} - -fn open() -> (TempDir, Db) { - let dir = TempDir::new().unwrap(); - let db = Db::open(dir.path()).unwrap(); - (dir, db) -} - -#[test] -fn leaf_set_get_delete() { - let (_dir, db) = open(); - let root = Store::root(); - let k = "reddit.com/r/rust".to_string(); - - assert_eq!(root.nodes().key(&k).get(&db).unwrap(), None); - db.run(root.nodes().key(&k).set(&"Rust".to_string()), Durability::SyncWal) - .unwrap(); - assert_eq!(root.nodes().key(&k).get(&db).unwrap(), Some("Rust".to_string())); - - db.run(root.nodes().key(&k).delete(), Durability::SyncWal).unwrap(); - assert_eq!(root.nodes().key(&k).get(&db).unwrap(), None); -} - -#[test] -fn sum_accumulates_with_blind_merges() { - let (_dir, db) = open(); - let edges = Store::root().scopes().key(&"s".to_string()).edges(); - let e = (3u32, 7u32); - - // Several blind merges in one atomic batch — no reads involved. - db.apply( - &[ - edges.key(&e).add(2.0), - edges.key(&e).add(1.0), - edges.key(&e).add(0.5), - ], - Durability::SyncWal, - ) - .unwrap(); - assert_eq!(edges.key(&e).get(&db).unwrap(), 3.5); - - // Negative delta decrements; absent key reads as zero. - db.run(edges.key(&e).add(-1.5), Durability::SyncWal).unwrap(); - assert_eq!(edges.key(&e).get(&db).unwrap(), 2.0); - assert_eq!(edges.key(&(9, 9)).get(&db).unwrap(), 0.0); -} - -#[test] -fn sum_set_then_merge() { - let (_dir, db) = open(); - let count = Store::root().scopes().key(&"s".to_string()).item_count(); - db.run(count.set(10), Durability::SyncWal).unwrap(); - db.run(count.add(5), Durability::SyncWal).unwrap(); - assert_eq!(count.get(&db).unwrap(), 15); -} - -#[test] -fn reified_writes_are_inspectable_data() { - let edges = Store::root().scopes().key(&"s".to_string()).edges(); - let merge = edges.key(&(1u32, 2u32)).add(1.0); - assert!(matches!(merge.op(), Op::Merge { .. })); - - let put = Store::root().nodes().key(&"x".to_string()).set(&"y".to_string()); - assert!(matches!(put.op(), Op::Put { .. })); - - let clear = Store::root().scopes().clear(); - assert!(matches!(clear.op(), Op::DeletePrefix { .. })); -} - -#[test] -fn point_update_touches_only_its_own_key() { - let (_dir, db) = open(); - let root = Store::root(); - let rust = root.scopes().key(&"rust".to_string()); - let python = root.scopes().key(&"python".to_string()); - - // Populate two scopes with several edges and some recent votes. - let mut batch = db.batch(); - for j in 0..5u32 { - batch.write(rust.edges().key(&(0, j)).add(j as f64 + 1.0)); - batch.write(python.edges().key(&(0, j)).add(100.0)); - } - batch - .push_back( - &rust.recent_votes(), - &Vote { a: "a".into(), b: "b".into(), ratio: 2 }, - ) - .unwrap(); - batch.commit().unwrap(); - - // A single precise update to one edge in `rust`. - db.run(rust.edges().key(&(0, 2)).add(10.0), Durability::SyncWal) - .unwrap(); - - // Only that edge changed. - assert_eq!(rust.edges().key(&(0, 2)).get(&db).unwrap(), 13.0); - assert_eq!(rust.edges().key(&(0, 0)).get(&db).unwrap(), 1.0); - assert_eq!(rust.edges().key(&(0, 4)).get(&db).unwrap(), 5.0); - // The other scope is entirely untouched. - for j in 0..5u32 { - assert_eq!(python.edges().key(&(0, j)).get(&db).unwrap(), 100.0); - } - // And the unrelated recent_votes deque is intact. - assert_eq!(rust.recent_votes().len(&db).unwrap(), 1); -} - -#[test] -fn map_keys_len_contains_entries() { - let (_dir, db) = open(); - let nodes = Store::root().nodes(); - db.apply( - &[ - nodes.key(&"a".to_string()).set(&"1".to_string()), - nodes.key(&"b".to_string()).set(&"2".to_string()), - nodes.key(&"c".to_string()).set(&"3".to_string()), - ], - Durability::SyncWal, - ) - .unwrap(); - - let mut keys = nodes.keys(&db).unwrap(); - keys.sort(); - assert_eq!(keys, vec!["a".to_string(), "b".to_string(), "c".to_string()]); - assert_eq!(nodes.len(&db).unwrap(), 3); - assert!(nodes.contains(&db, &"b".to_string()).unwrap()); - assert!(!nodes.contains(&db, &"z".to_string()).unwrap()); - - let mut pairs = nodes.iter(&db).unwrap(); - pairs.sort(); - assert_eq!( - pairs, - vec![ - ("a".to_string(), "1".to_string()), - ("b".to_string(), "2".to_string()), - ("c".to_string(), "3".to_string()), - ] - ); -} - -#[test] -fn map_keys_dedup_across_nested_subkeys() { - // A map whose values are nested structs has many physical keys per logical - // key; `keys()`/`len()` must dedup to distinct logical keys. - let (_dir, db) = open(); - let scopes = Store::root().scopes(); - let rust = scopes.key(&"rust".to_string()); - - let mut batch = db.batch(); - batch.write(rust.edges().key(&(0, 1)).add(1.0)); - batch.write(rust.edges().key(&(0, 2)).add(1.0)); - batch.write(rust.item_count().add(3)); - batch - .push_back(&rust.recent_votes(), &Vote { a: "a".into(), b: "b".into(), ratio: 1 }) - .unwrap(); - batch.write(scopes.key(&"python".to_string()).item_count().add(1)); - batch.commit().unwrap(); - - let mut keys = scopes.keys(&db).unwrap(); - keys.sort(); - assert_eq!(keys, vec!["python".to_string(), "rust".to_string()]); - assert_eq!(scopes.len(&db).unwrap(), 2); -} - -#[test] -fn map_clear_deletes_subtree_only() { - let (_dir, db) = open(); - let rust = Store::root().scopes().key(&"rust".to_string()); - - let mut batch = db.batch(); - batch.write(rust.edges().key(&(0, 1)).add(1.0)); - batch.write(rust.edges().key(&(0, 2)).add(2.0)); - batch.write(rust.item_count().add(5)); - batch.commit().unwrap(); - - // Clear only the edges sub-map. - db.run(rust.edges().clear(), Durability::SyncWal).unwrap(); - - assert_eq!(rust.edges().len(&db).unwrap(), 0); - assert_eq!(rust.edges().key(&(0, 1)).get(&db).unwrap(), 0.0); - // Sibling field under the same scope is untouched. - assert_eq!(rust.item_count().get(&db).unwrap(), 5); -} - -#[test] -fn transform_values_decays_all_edges_in_one_batch() { - let (_dir, db) = open(); - let edges = Store::root().scopes().key(&"rust".to_string()).edges(); - - let mut batch = db.batch(); - for j in 1..=4u32 { - batch.write(edges.key(&(0, j)).add(j as f64 * 10.0)); - } - batch.commit().unwrap(); - - // Decay every edge by half, dropping any that fall to/under 5.0 — built as - // reified writes from one scan, applied atomically. - let writes = edges - .transform_values(&db, |_k, w| { - let decayed = w * 0.5; - if decayed <= 5.0 { - None - } else { - Some(decayed) - } - }) - .unwrap(); - db.apply(&writes, Durability::SyncWal).unwrap(); - - assert_eq!(edges.key(&(0, 1)).get(&db).unwrap(), 0.0); // 10*0.5=5.0 -> dropped - assert_eq!(edges.key(&(0, 2)).get(&db).unwrap(), 10.0); - assert_eq!(edges.key(&(0, 3)).get(&db).unwrap(), 15.0); - assert_eq!(edges.key(&(0, 4)).get(&db).unwrap(), 20.0); - assert_eq!(edges.len(&db).unwrap(), 3); -} - -#[test] -fn list_push_pop_iter() { - let (_dir, db) = open(); - let log = Store::root().log(); - - assert!(log.is_empty(&db).unwrap()); - assert_eq!(log.push(&db, &10).unwrap(), 0); - assert_eq!(log.push(&db, &20).unwrap(), 1); - assert_eq!(log.push(&db, &30).unwrap(), 2); - - assert_eq!(log.len(&db).unwrap(), 3); - assert_eq!(log.get(&db, 1).unwrap(), Some(20)); - assert_eq!(log.get(&db, 3).unwrap(), None); - assert_eq!(log.iter(&db).unwrap(), vec![10, 20, 30]); - - assert_eq!(log.pop(&db).unwrap(), Some(30)); - assert_eq!(log.len(&db).unwrap(), 2); - assert_eq!(log.iter(&db).unwrap(), vec![10, 20]); -} - -#[test] -fn batched_list_pushes_get_contiguous_indices() { - let (_dir, db) = open(); - let log = Store::root().log(); - - let mut batch = db.batch(); - batch.push(&log, &1).unwrap(); - batch.push(&log, &2).unwrap(); - batch.push(&log, &3).unwrap(); - batch.commit().unwrap(); - assert_eq!(log.iter(&db).unwrap(), vec![1, 2, 3]); - - // A second batch continues from the persisted length. - let mut batch = db.batch(); - batch.push(&log, &4).unwrap(); - batch.push(&log, &5).unwrap(); - batch.commit().unwrap(); - assert_eq!(log.iter(&db).unwrap(), vec![1, 2, 3, 4, 5]); - assert_eq!(log.len(&db).unwrap(), 5); -} - -#[test] -fn deque_behaves_like_a_double_ended_queue() { - let (_dir, db) = open(); - let dq = Store::root().scopes().key(&"s".to_string()).recent_votes(); - let v = |n: i32| Vote { a: format!("a{n}"), b: format!("b{n}"), ratio: n }; - - dq.push_back(&db, &v(1)).unwrap(); - dq.push_back(&db, &v(2)).unwrap(); - dq.push_front(&db, &v(0)).unwrap(); - - assert_eq!(dq.len(&db).unwrap(), 3); - assert_eq!(dq.iter(&db).unwrap(), vec![v(0), v(1), v(2)]); - assert_eq!(dq.front(&db).unwrap(), Some(v(0))); - assert_eq!(dq.back(&db).unwrap(), Some(v(2))); - - assert_eq!(dq.pop_front(&db).unwrap(), Some(v(0))); - assert_eq!(dq.pop_back(&db).unwrap(), Some(v(2))); - assert_eq!(dq.iter(&db).unwrap(), vec![v(1)]); - assert_eq!(dq.pop_front(&db).unwrap(), Some(v(1))); - assert_eq!(dq.pop_front(&db).unwrap(), None); - assert!(dq.is_empty(&db).unwrap()); -} - -#[test] -fn deque_supports_capped_recent_window() { - // The motivating use case: keep only the most recent N votes, O(1) per insert. - let (_dir, db) = open(); - let dq = Store::root().scopes().key(&"s".to_string()).recent_votes(); - const CAP: u64 = 3; - - for n in 0..10 { - dq.push_back(&db, &Vote { a: format!("{n}"), b: "x".into(), ratio: n }) - .unwrap(); - while dq.len(&db).unwrap() > CAP { - dq.pop_front(&db).unwrap(); - } - } - - let kept = dq.iter(&db).unwrap(); - assert_eq!(kept.len(), 3); - assert_eq!(kept.iter().map(|v| v.ratio).collect::>(), vec![7, 8, 9]); -} - -#[test] -fn deque_truncate_back_caps_length_keeping_front() { - let (_dir, db) = open(); - let dq = Store::root().scopes().key(&"s".to_string()).recent_votes(); - for n in 0..10 { - dq.push_back(&db, &Vote { a: format!("{n}"), b: "x".into(), ratio: n }) - .unwrap(); - } - // Keep only the 3 oldest at front (drop the back/newest beyond cap). - dq.truncate_back(&db, 3, Durability::SyncWal).unwrap(); - let kept = dq.iter(&db).unwrap(); - assert_eq!(kept.iter().map(|v| v.ratio).collect::>(), vec![0, 1, 2]); - - // Truncating to a larger-or-equal cap is a no-op. - dq.truncate_back(&db, 10, Durability::SyncWal).unwrap(); - assert_eq!(dq.len(&db).unwrap(), 3); -} - -#[test] -fn one_batch_commits_all_or_nothing_and_persists() { - let dir = TempDir::new().unwrap(); - let rust_key = "rust".to_string(); - { - let db = Db::open(dir.path()).unwrap(); - let rust = Store::root().scopes().key(&rust_key); - // A "vote" as one atomic batch: two edge merges, a pair flag, a recent - // vote, and a counter — all distinct keys, one WAL flush. - let mut batch = db.batch(); - batch.write(rust.edges().key(&(0, 1)).add(2.0)); - batch.write(rust.edges().key(&(1, 0)).add(1.0)); - batch.write(rust.voted_pairs().key(&(0, 1)).set(&true)); - batch - .push_back(&rust.recent_votes(), &Vote { a: "0".into(), b: "1".into(), ratio: 2 }) - .unwrap(); - batch.write(rust.item_count().add(2)); - batch.commit().unwrap(); - } - - // Reopen: SyncWal data survives. - let db = Db::open(dir.path()).unwrap(); - let rust = Store::root().scopes().key(&rust_key); - assert_eq!(rust.edges().key(&(0, 1)).get(&db).unwrap(), 2.0); - assert_eq!(rust.edges().key(&(1, 0)).get(&db).unwrap(), 1.0); - assert_eq!(rust.voted_pairs().key(&(0, 1)).get(&db).unwrap(), Some(true)); - assert_eq!(rust.recent_votes().len(&db).unwrap(), 1); - assert_eq!(rust.item_count().get(&db).unwrap(), 2); -} - -#[test] -fn disable_wal_visible_within_session() { - let (_dir, db) = open(); - let count = Store::root().scopes().key(&"s".to_string()).item_count(); - db.run(count.add(7), Durability::DisableWal).unwrap(); - assert_eq!(count.get(&db).unwrap(), 7); -} - -#[test] -fn wal_only_durability_writes() { - let (_dir, db) = open(); - let node = Store::root().nodes().key(&"k".to_string()); - db.run(node.set(&"v".to_string()), Durability::WalOnly).unwrap(); - assert_eq!(node.get(&db).unwrap(), Some("v".to_string())); -} - -#[test] -fn namespaced_roots_do_not_collide() { - let (_dir, db) = open(); - let a = Store::namespaced("a"); - let b = Store::namespaced("b"); - db.run(a.nodes().key(&"k".to_string()).set(&"av".to_string()), Durability::SyncWal) - .unwrap(); - db.run(b.nodes().key(&"k".to_string()).set(&"bv".to_string()), Durability::SyncWal) - .unwrap(); - - assert_eq!(a.nodes().key(&"k".to_string()).get(&db).unwrap(), Some("av".to_string())); - assert_eq!(b.nodes().key(&"k".to_string()).get(&db).unwrap(), Some("bv".to_string())); -} - -#[test] -fn persistence_across_reopen_for_all_collection_kinds() { - let dir = TempDir::new().unwrap(); - { - let db = Db::open(dir.path()).unwrap(); - let root = Store::root(); - let s = root.scopes().key(&"s".to_string()); - db.run(root.nodes().key(&"n".to_string()).set(&"N".to_string()), Durability::SyncWal) - .unwrap(); - root.log().push(&db, &42).unwrap(); - db.run(s.edges().key(&(1, 2)).add(9.0), Durability::SyncWal).unwrap(); - s.recent_votes() - .push_back(&db, &Vote { a: "a".into(), b: "b".into(), ratio: 3 }) - .unwrap(); - } - let db = Db::open(dir.path()).unwrap(); - let root = Store::root(); - let s = root.scopes().key(&"s".to_string()); - assert_eq!(root.nodes().key(&"n".to_string()).get(&db).unwrap(), Some("N".to_string())); - assert_eq!(root.log().iter(&db).unwrap(), vec![42]); - assert_eq!(s.edges().key(&(1, 2)).get(&db).unwrap(), 9.0); - assert_eq!( - s.recent_votes().front(&db).unwrap(), - Some(Vote { a: "a".into(), b: "b".into(), ratio: 3 }) - ); -} diff --git a/durable/tests/proptests.rs b/durable/tests/proptests.rs deleted file mode 100644 index 9580a5fa406d29e886ac46c91f5cc723ee21c346..0000000000000000000000000000000000000000 --- a/durable/tests/proptests.rs +++ /dev/null @@ -1,103 +0,0 @@ -//! Property tests: durable collections must behave like their std analogues. - -use std::collections::{BTreeMap, VecDeque}; - -use durable::{Db, Durability, Durable, Deque, Leaf, List, Map, Sum}; -use proptest::prelude::*; -use tempfile::TempDir; - -#[derive(Durable)] -#[allow(dead_code)] -struct Bag { - map: Map>, - list: List>, - deque: Deque>, - total: Sum, -} - -fn open() -> (TempDir, Db) { - let dir = TempDir::new().unwrap(); - let db = Db::open(dir.path()).unwrap(); - (dir, db) -} - -proptest! { - #[test] - fn map_matches_btreemap(entries in proptest::collection::vec((".*", any::()), 0..40)) { - let (_dir, db) = open(); - let map = Bag::root().map(); - let mut model = BTreeMap::new(); - - let mut batch = db.batch(); - for (k, v) in &entries { - batch.write(map.key(k).set(v)); - model.insert(k.clone(), *v); - } - batch.commit_with(Durability::WalOnly).unwrap(); - - prop_assert_eq!(map.len(&db).unwrap(), model.len()); - for (k, v) in &model { - prop_assert_eq!(map.get(&db, k).unwrap(), Some(*v)); - } - let mut got = map.iter(&db).unwrap(); - got.sort(); - let mut want: Vec<(String, i64)> = model.into_iter().collect(); - want.sort(); - prop_assert_eq!(got, want); - } - - #[test] - fn list_roundtrips_in_order(values in proptest::collection::vec(any::(), 0..50)) { - let (_dir, db) = open(); - let list = Bag::root().list(); - let mut batch = db.batch(); - for v in &values { - batch.push(&list, v).unwrap(); - } - batch.commit_with(Durability::WalOnly).unwrap(); - - prop_assert_eq!(list.len(&db).unwrap(), values.len() as u64); - prop_assert_eq!(list.iter(&db).unwrap(), values); - } - - #[test] - fn deque_matches_vecdeque(ops in proptest::collection::vec(any::<(bool, i64)>(), 0..60)) { - let (_dir, db) = open(); - let dq = Bag::root().deque(); - let mut model: VecDeque = VecDeque::new(); - - for (front, v) in &ops { - if *front { - dq.push_front(&db, v).unwrap(); - model.push_front(*v); - } else { - dq.push_back(&db, v).unwrap(); - model.push_back(*v); - } - } - prop_assert_eq!(dq.len(&db).unwrap(), model.len() as u64); - prop_assert_eq!(dq.iter(&db).unwrap(), Vec::from(model.clone())); - - // Drain alternately from both ends. - let mut toggle = true; - while !model.is_empty() { - if toggle { - prop_assert_eq!(dq.pop_front(&db).unwrap(), model.pop_front()); - } else { - prop_assert_eq!(dq.pop_back(&db).unwrap(), model.pop_back()); - } - toggle = !toggle; - } - prop_assert!(dq.is_empty(&db).unwrap()); - prop_assert_eq!(dq.pop_front(&db).unwrap(), None); - } - - #[test] - fn sum_equals_total_of_deltas(deltas in proptest::collection::vec(-1000i64..1000, 0..50)) { - let (_dir, db) = open(); - let total = Bag::root().total(); - let writes: Vec<_> = deltas.iter().map(|d| total.add(*d)).collect(); - db.apply(&writes, Durability::WalOnly).unwrap(); - prop_assert_eq!(total.get(&db).unwrap(), deltas.iter().sum::()); - } -} diff --git a/scripts/cursor-env-install.sh b/scripts/cursor-env-install.sh index 9aeffdd1164016d5955efc81295f6c7f42b3d925..c4d547a4bdcf39d46cd8929f1b9589537c17dea4 100755 --- a/scripts/cursor-env-install.sh +++ b/scripts/cursor-env-install.sh @@ -190,7 +190,6 @@ clojure -P -M clojure -M -e "(com.microsoft.playwright.CLI/main (into-array String [\"install\" \"chromium\" \"--with-deps\"]))" # Warm RocksDB + release server link (Clojure tests use release binary). -cargo build -p durable --quiet cargo build --release --package sorter2-server --quiet echo "cursor-env-install: ok (bb=$(bb --version 2>/dev/null || echo missing), CXX=${CXX})" diff --git a/server/Cargo.toml b/server/Cargo.toml index ad4912791aff59fb1d3293f66ad381ae618cd60b..dfa39beddecfa37dcdeaa602cb30f4b547528fbb 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -25,7 +25,7 @@ futures-util = { version = "0.3", default-features = false, features = ["std"] } rand = "0.8" urlencoding = "2" url = "2" -durable = { path = "../durable" } +durable = { git = "https://github.com/tommy-mor/durable.git", branch = "main" } [dev-dependencies] reqwest = { version = "0.12", features = ["json"] } diff --git a/server/src/api/ui_html.rs b/server/src/api/ui_html.rs index e6e61815f6a6ddd626f79be81bf70256c047380a..b86581b1f337650564274254d840e8a75b49524d 100644 --- a/server/src/api/ui_html.rs +++ b/server/src/api/ui_html.rs @@ -3,14 +3,20 @@ use axum::{ response::{IntoResponse, Response}, Form, }; +use axum_extra::extract::cookie::CookieJar; use std::collections::HashMap; use crate::{ + auth::{ + alias_status_js, alias_redirect_js, config, login_redirect_js, oauth, redirect_js, resolve_vote_actor, + session::{load_valid_session, session_has_pseudonym, session_id_from_jar}, + }, fetch, html::{input_panel, js_string_literal, ranking_panel, JsBuilder}, parser::parse_reddit_url, path_types::ItemId, state::{parse_item_param, AppState}, + storage_schema::pseudonym_owner, ui_action::{parse_html_ui_from_form, HtmlUiAction}, }; @@ -30,8 +36,21 @@ fn parent_from_scope(scope: &str) -> ItemId { parse_item_param(scope) } +fn vote_auth_redirect(state: &AppState, jar: &CookieJar) -> Option { + let db = state.projection_store.db(); + let session = session_id_from_jar(jar) + .as_deref() + .and_then(|id| load_valid_session(db, id)); + match session { + None => Some(login_redirect_js().into_response()), + Some(s) if !session_has_pseudonym(&s) => Some(alias_redirect_js().into_response()), + Some(_) => None, + } +} + pub async fn post_ui_html( State(state): State, + jar: CookieJar, Form(form): Form>, ) -> impl IntoResponse { let action = match parse_html_ui_from_form(&form) { @@ -48,9 +67,16 @@ pub async fn post_ui_html( scope, vote_compare, } => { + if let Some(resp) = vote_auth_redirect(&state, &jar) { + return resp; + } let parent = parent_from_scope(&scope); + let actor = resolve_vote_actor( + state.projection_store.db(), + session_id_from_jar(&jar).as_deref(), + ); if let Err(e) = state - .record_vote(&parent, &a, &b, ratio_left, ratio_right) + .record_vote(&parent, &a, &b, ratio_left, ratio_right, &actor) .await { return ui_js_warn(&e).into_response(); @@ -72,6 +98,58 @@ pub async fn post_ui_html( .morph_selector("#ranking-panel", panel) .into_response() } + HtmlUiAction::CheckPseudonym { pseudonym } => { + let db = state.projection_store.db(); + let session_id = match session_id_from_jar(&jar) { + Some(id) => id, + None => return alias_status_js("sign in first", false).into_response(), + }; + let session = match load_valid_session(db, &session_id) { + Some(s) => s, + None => return alias_status_js("session expired", false).into_response(), + }; + match oauth::validate_pseudonym(&pseudonym) { + Err(msg) => alias_status_js(msg, false).into_response(), + Ok(name) => match pseudonym_owner(db, &name) { + Ok(None) => alias_status_js("available", true).into_response(), + Ok(Some(owner)) if owner == session.uuid => { + alias_status_js("already yours", true).into_response() + } + Ok(Some(_)) => alias_status_js("taken", false).into_response(), + Err(e) => ui_js_warn(&e.to_string()).into_response(), + }, + } + } + HtmlUiAction::ClaimPseudonym { + pseudonym, + return_to, + } => { + let db = state.projection_store.db(); + let session_id = match session_id_from_jar(&jar) { + Some(id) => id, + None => return login_redirect_js().into_response(), + }; + let session = match load_valid_session(db, &session_id) { + Some(s) => s, + None => return login_redirect_js().into_response(), + }; + let name = match oauth::validate_pseudonym(&pseudonym) { + Ok(n) => n, + Err(msg) => return alias_status_js(msg, false).into_response(), + }; + if let Ok(Some(owner)) = pseudonym_owner(db, &name) { + if owner != session.uuid { + return alias_status_js("taken", false).into_response(); + } + } else if let Err(e) = state.claim_pseudonym(&session.uuid, &name).await { + return ui_js_warn(&e).into_response(); + } + if let Err(e) = crate::auth::session::update_session_pseudonym(&db, &session_id, &name) + { + return ui_js_warn(&e).into_response(); + } + redirect_js(&config::sanitize_return_to(&return_to)).into_response() + } HtmlUiAction::ParseQuery { query } => match parse_reddit_url(&query) { Ok(item) => { let _ = state.ensure_node(&item).await; diff --git a/server/src/auth/config.rs b/server/src/auth/config.rs new file mode 100644 index 0000000000000000000000000000000000000000..a1f042c655bf3e5234eeb87a7d889f64592807fb --- /dev/null +++ b/server/src/auth/config.rs @@ -0,0 +1,9 @@ +pub const AUTH_RETURN_COOKIE: &str = "sorter2_auth_return"; + +pub fn sanitize_return_to(raw: &str) -> String { + let s = raw.trim(); + if s.is_empty() || !s.starts_with('/') || s.starts_with("//") { + return "/".to_string(); + } + s.to_string() +} diff --git a/server/src/auth/identity.rs b/server/src/auth/identity.rs new file mode 100644 index 0000000000000000000000000000000000000000..20e51ac79ce1a5fd8c5ac2ab16b5799cb9a522d8 --- /dev/null +++ b/server/src/auth/identity.rs @@ -0,0 +1,27 @@ +//! Trust-weight calculation from linked OAuth providers. + +/// Base weight before any OAuth links. +pub const BASE_TRUST_WEIGHT: f64 = 1.0; + +/// Increment per linked provider (frozen at vote cast time). +pub const TRUST_WEIGHT_PER_LINK: f64 = 0.5; + +pub fn trust_weight_for_link_count(link_count: usize) -> f64 { + BASE_TRUST_WEIGHT + TRUST_WEIGHT_PER_LINK * link_count as f64 +} + +pub fn trust_weight_after_link(current: f64) -> f64 { + current + TRUST_WEIGHT_PER_LINK +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn trust_weight_scales_with_links() { + assert_eq!(trust_weight_for_link_count(0), 1.0); + assert_eq!(trust_weight_for_link_count(1), 1.5); + assert_eq!(trust_weight_for_link_count(2), 2.0); + } +} diff --git a/server/src/auth/mod.rs b/server/src/auth/mod.rs new file mode 100644 index 0000000000000000000000000000000000000000..5ed535ba199fa736f0048c32623b14c3b1e5de2d --- /dev/null +++ b/server/src/auth/mod.rs @@ -0,0 +1,412 @@ +//! GitHub OAuth login, session cookies, and vote actor resolution. + +pub mod config; +pub mod identity; +pub mod oauth; +pub mod session; + +use axum::{ + extract::{Query, State}, + http::StatusCode, + response::{Html, IntoResponse, Redirect, Response}, + Form, +}; +use axum_extra::extract::cookie::CookieJar; +use maud::{html, Markup}; +use reqwest::Client; +use serde::Deserialize; + +use crate::{ + events::Event, + fetch::now_ms, + form_template::template_json_compact, + html::layout, + state::AppState, + storage_schema::{oauth_link_owner, pseudonym_owner, Store, StoreFields}, + ui_action::UI_RPC_FIELD, +}; + +pub use session::{resolve_vote_actor, session_id_from_jar, VoteActor}; + +pub fn base_url_from_env(port: u16) -> String { + std::env::var("SORTER2_BASE_URL") + .unwrap_or_else(|_| format!("http://127.0.0.1:{port}")) +} + +fn new_actor_uuid() -> String { + let mut bytes = [0u8; 16]; + rand::Rng::fill(&mut rand::thread_rng(), &mut bytes); + format!( + "{:08x}-{:04x}-{:04x}-{:04x}-{:012x}", + u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]), + u16::from_be_bytes([bytes[4], bytes[5]]), + u16::from_be_bytes([bytes[6], bytes[7]]) | 0x4000, + u16::from_be_bytes([bytes[8], bytes[9]]) | 0x8000, + u128::from_be_bytes([ + 0, 0, bytes[10], bytes[11], bytes[12], bytes[13], bytes[14], bytes[15], 0, 0, 0, 0, + 0, 0, 0, 0, + ]) & 0x0000_FFFF_FFFF_FFFF + ) +} + +#[derive(Debug, Deserialize)] +pub struct LoginQuery { + #[serde(default)] + pub return_to: Option, +} + +#[derive(Debug, Deserialize)] +pub struct GitHubStartQuery { + #[serde(default)] + pub return_to: Option, + #[serde(default)] + pub mock_user: Option, +} + +fn return_from_query_or_jar(jar: &CookieJar, query: Option<&str>) -> String { + if let Some(raw) = query { + return config::sanitize_return_to(raw); + } + jar.get(config::AUTH_RETURN_COOKIE) + .map(|c| config::sanitize_return_to(c.value())) + .unwrap_or_else(|| "/".to_string()) +} + +fn oauth_providers(base_url: &str, return_to: &str) -> Vec<(&'static str, String)> { + let mut out = Vec::new(); + if oauth::GitHubConfig::from_env(base_url).is_some() { + out.push(( + "GitHub", + format!( + "/auth/github?return_to={}", + urlencoding::encode(return_to) + ), + )); + } + out +} + +fn alias_list(db: &durable::Db, uuid: &str) -> Vec { + Store::root() + .user_pseudonyms() + .key(&uuid.to_string()) + .iter(db) + .unwrap_or_default() +} + +fn login_body( + session: Option<&session::SessionActor>, + aliases: &[String], + providers: &[(&str, String)], +) -> Markup { + html! { + main class="panel login-page" { + div class="login-grid" { + section class="login-oauth" { + h1 { "sign in" } + @if providers.is_empty() { + p class="muted" { + "OAuth is not configured. Set GITHUB_CLIENT_ID and GITHUB_CLIENT_SECRET." + } + } @else { + ul class="oauth-provider-list" { + @for (name, href) in providers { + li { + a href=(href) class="button oauth-provider" data-testid=(format!("oauth-{}", name.to_lowercase())) { + (format!("Continue with {name}")) + } + } + } + } + } + @if let Some(actor) = session { + p class="muted small" { + "session active · weight " (format!("{:.1}", actor.trust_weight)) + } + form method="post" action="/auth/logout" data-navigate="full" { + button type="submit" { "log out" } + } + } + } + section class="login-aliases" { + h2 { "your aliases" } + ul id="alias-list" class="alias-list" { + @if aliases.is_empty() { + li class="muted" data-testid="alias-list-empty" { "none yet" } + } @else { + @for alias in aliases { + li { (alias) } + } + } + } + } + } + p { a href="/" { "← back" } } + } + } +} + +pub async fn login_page( + State(state): State, + jar: CookieJar, + Query(query): Query, +) -> Response { + let return_to = return_from_query_or_jar(&jar, query.return_to.as_deref()); + let jar = jar.add(session::auth_return_cookie_value(&return_to)); + + let db = state.projection_store.db(); + let session = session::session_id_from_jar(&jar) + .as_deref() + .and_then(|id| session::load_session_actor(db, id)); + let aliases = session + .as_ref() + .map(|s| alias_list(db, &s.uuid)) + .unwrap_or_default(); + let providers = oauth_providers(&base_url_from_env(state.cfg.port), &return_to); + + let markup = layout( + "login · sorter2", + login_body(session.as_ref(), &aliases, &providers), + state.views.get_views("/login"), + ); + (jar, Html(markup.into_string())).into_response() +} + +pub async fn alias_page( + State(state): State, + jar: CookieJar, + Query(query): Query, +) -> Result { + let return_to = return_from_query_or_jar(&jar, query.return_to.as_deref()); + let session_id = session::session_id_from_jar(&jar).ok_or(StatusCode::UNAUTHORIZED)?; + let db = state.projection_store.db(); + let session = session::load_valid_session(db, &session_id).ok_or(StatusCode::UNAUTHORIZED)?; + if session::session_has_pseudonym(&session) { + return Ok(Redirect::to(&return_to).into_response()); + } + + let check_rpc = template_json_compact(&serde_json::json!({ + "action": "check_pseudonym", + "pseudonym": {"$form": "pseudonym"}, + })) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + let claim_rpc = template_json_compact(&serde_json::json!({ + "action": "claim_pseudonym", + "pseudonym": {"$form": "pseudonym"}, + "return_to": return_to, + })) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + + let body = html! { + main class="panel alias-page" { + h1 { "choose alias" } + p class="muted" { "pick a unique display name for your votes" } + form id="alias-check-form" method="POST" action="/ui" { + input type="hidden" name=(UI_RPC_FIELD) value=(check_rpc); + label { "alias" } + input type="text" id="alias-input" name="pseudonym" autocomplete="off" + data-testid="alias-input" maxlength="64"; + p id="alias-status" class="muted" data-testid="alias-status" { "type to check availability" } + } + form id="alias-claim-form" method="POST" action="/ui" { + input type="hidden" name=(UI_RPC_FIELD) value=(claim_rpc); + input type="hidden" name="pseudonym" id="alias-claim-field" value=""; + button type="submit" class="btn-primary" data-testid="alias-claim" { "continue" } + } + p { a href="/login" { "← back to login" } } + } + }; + + Ok(Html( + layout( + "choose alias · sorter2", + body, + state.views.get_views("/login/alias"), + ) + .into_string(), + ) + .into_response()) +} + +pub async fn github_start( + State(state): State, + jar: CookieJar, + Query(query): Query, +) -> Result { + let cfg = oauth::GitHubConfig::from_env(&base_url_from_env(state.cfg.port)) + .ok_or(StatusCode::SERVICE_UNAVAILABLE)?; + let return_to = return_from_query_or_jar(&jar, query.return_to.as_deref()); + let state_token = session::new_oauth_state(); + let url = oauth::authorize_url(&cfg, &state_token, query.mock_user.as_deref()); + let jar = jar + .add(session::oauth_state_cookie_value(&state_token)) + .add(session::auth_return_cookie_value(&return_to)); + Ok((jar, Redirect::temporary(&url)).into_response()) +} + +#[derive(Debug, Deserialize)] +pub struct OAuthCallbackQuery { + pub code: String, + pub state: String, +} + +async fn finish_oauth_login( + state: &AppState, + jar: CookieJar, + provider: &str, + provider_id: String, +) -> Result<(CookieJar, String), StatusCode> { + let db = state.projection_store.db(); + let return_to = return_from_query_or_jar(&jar, None); + + let uuid = match oauth_link_owner(db, provider, &provider_id) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + { + Some(existing) => existing, + None => { + let uuid = new_actor_uuid(); + let ts = now_ms(); + state + .append_identity_events(vec![ + Event::PrincipalCreated { + uuid: uuid.clone(), + ts, + }, + Event::OauthLinked { + uuid: uuid.clone(), + provider: provider.to_string(), + provider_id, + ts, + }, + ]) + .await + .map_err(|e| { + tracing::warn!(err = %e, "identity event append failed"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + uuid + } + }; + + let aliases = alias_list(db, &uuid); + let pseudonym = aliases.last().cloned().unwrap_or_default(); + let (session_id, _) = session::create_session(db, &uuid, &pseudonym).map_err(|e| { + tracing::warn!(err = %e, "session create failed"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + + let jar = jar + .add(session::session_cookie_value(&session_id)) + .add(session::clear_oauth_state_cookie()); + + let dest = if pseudonym.is_empty() { + format!( + "/login/alias?return_to={}", + urlencoding::encode(&return_to) + ) + } else { + return_to + }; + + Ok((jar, dest)) +} + +pub async fn github_callback( + State(state): State, + jar: CookieJar, + Query(query): Query, +) -> Result { + let cfg = oauth::GitHubConfig::from_env(&base_url_from_env(state.cfg.port)) + .ok_or(StatusCode::SERVICE_UNAVAILABLE)?; + + let expected_state = session::oauth_state_from_jar(&jar).ok_or(StatusCode::BAD_REQUEST)?; + if expected_state != query.state { + return Err(StatusCode::BAD_REQUEST); + } + + let client = Client::builder() + .timeout(std::time::Duration::from_secs(15)) + .build() + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + + let token = oauth::exchange_code(&client, &cfg, &query.code) + .await + .map_err(|e| { + tracing::warn!(err = %e, "github oauth token exchange failed"); + StatusCode::BAD_GATEWAY + })?; + let user = oauth::fetch_user(&client, &cfg.api_base, &token) + .await + .map_err(|e| { + tracing::warn!(err = %e, "github user fetch failed"); + StatusCode::BAD_GATEWAY + })?; + + let provider = "github"; + let provider_id = oauth::provider_id(&user); + let (jar, dest) = finish_oauth_login(&state, jar, provider, provider_id).await?; + Ok((jar, Redirect::to(&dest)).into_response()) +} + +pub async fn logout(State(state): State, jar: CookieJar) -> impl IntoResponse { + if let Some(session_id) = session::session_id_from_jar(&jar) { + let _ = session::destroy_session(state.projection_store.db(), &session_id); + } + let jar = jar + .add(session::clear_session_cookie()) + .add(session::clear_auth_return_cookie()); + (jar, Redirect::to("/login")) +} + +#[derive(Deserialize)] +pub struct SwitchPseudonymForm { + pseudonym: String, +} + +pub async fn switch_pseudonym( + State(state): State, + jar: CookieJar, + Form(form): Form, +) -> Result { + let session_id = session::session_id_from_jar(&jar).ok_or(StatusCode::UNAUTHORIZED)?; + let db = state.projection_store.db(); + let actor = session::load_session_actor(db, &session_id).ok_or(StatusCode::UNAUTHORIZED)?; + let owner = pseudonym_owner(db, &form.pseudonym) + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .ok_or(StatusCode::NOT_FOUND)?; + if owner != actor.uuid { + return Err(StatusCode::FORBIDDEN); + } + session::update_session_pseudonym(db, &session_id, &form.pseudonym).map_err(|_| { + StatusCode::INTERNAL_SERVER_ERROR + })?; + Ok(Redirect::to("/login").into_response()) +} + +pub fn redirect_js(path: &str) -> crate::html::JsBuilder { + crate::html::JsBuilder::new().raw(&format!( + "window.location.href={};", + crate::html::js_string_literal(path) + )) +} + +pub fn login_redirect_js() -> crate::html::JsBuilder { + crate::html::JsBuilder::new().raw( + "window.location.href='/login?return_to='+encodeURIComponent(window.location.pathname+window.location.search);", + ) +} + +pub fn alias_redirect_js() -> crate::html::JsBuilder { + crate::html::JsBuilder::new().raw( + "window.location.href='/login/alias?return_to='+encodeURIComponent(window.location.pathname+window.location.search);", + ) +} + +pub fn alias_status_js(message: &str, ok: bool) -> crate::html::JsBuilder { + let class = if ok { "alias-ok" } else { "alias-bad" }; + crate::html::JsBuilder::new().raw(&format!( + "var el=document.getElementById('alias-status'); if(el){{ el.textContent={}; el.className={}; }} var cf=document.getElementById('alias-claim-field'); if(cf) cf.value=document.getElementById('alias-input')?.value||'';", + crate::html::js_string_literal(message), + crate::html::js_string_literal(class), + )) +} diff --git a/server/src/auth/oauth.rs b/server/src/auth/oauth.rs new file mode 100644 index 0000000000000000000000000000000000000000..b80078fee0c5acd905b9125d29454e46c4e0d066 --- /dev/null +++ b/server/src/auth/oauth.rs @@ -0,0 +1,146 @@ +//! GitHub OAuth (raw reqwest, same style as reddit.rs). + +use reqwest::Client; +use serde::Deserialize; + +#[derive(Debug, Clone)] +pub struct GitHubConfig { + pub client_id: String, + pub client_secret: String, + pub redirect_uri: String, + pub oauth_base: String, + pub api_base: String, +} + +pub fn github_oauth_base() -> String { + std::env::var("GITHUB_OAUTH_BASE").unwrap_or_else(|_| "https://github.com".into()) +} + +pub fn github_api_base() -> String { + std::env::var("GITHUB_API_BASE").unwrap_or_else(|_| "https://api.github.com".into()) +} + +impl GitHubConfig { + pub fn from_env(base_url: &str) -> Option { + let client_id = std::env::var("GITHUB_CLIENT_ID").ok()?; + let client_secret = std::env::var("GITHUB_CLIENT_SECRET").ok()?; + if client_id.is_empty() || client_secret.is_empty() { + return None; + } + let oauth_base = github_oauth_base(); + let base = base_url.trim_end_matches('/'); + Some(Self { + client_id, + client_secret, + redirect_uri: format!("{base}/auth/github/callback"), + oauth_base, + api_base: github_api_base(), + }) + } +} + +#[derive(Debug, Deserialize)] +struct TokenResponse { + access_token: String, +} + +#[derive(Debug, Deserialize)] +pub struct GitHubUser { + pub id: u64, + pub login: String, +} + +pub fn authorize_url(cfg: &GitHubConfig, state: &str, mock_user: Option<&str>) -> String { + let mut url = format!( + "{}/login/oauth/authorize?client_id={}&redirect_uri={}&scope=read:user&state={}", + cfg.oauth_base.trim_end_matches('/'), + urlencoding::encode(&cfg.client_id), + urlencoding::encode(&cfg.redirect_uri), + urlencoding::encode(state), + ); + if let Some(user) = mock_user { + url.push_str("&mock_user="); + url.push_str(&urlencoding::encode(user)); + } + url +} + +pub async fn exchange_code( + client: &Client, + cfg: &GitHubConfig, + code: &str, +) -> Result { + let resp = client + .post(format!( + "{}/login/oauth/access_token", + cfg.oauth_base.trim_end_matches('/') + )) + .header("Accept", "application/json") + .form(&[ + ("client_id", cfg.client_id.as_str()), + ("client_secret", cfg.client_secret.as_str()), + ("code", code), + ("redirect_uri", cfg.redirect_uri.as_str()), + ]) + .send() + .await + .map_err(|e| format!("github token request failed: {e}"))?; + + if !resp.status().is_success() { + return Err(format!("github token HTTP {}", resp.status())); + } + + let body: TokenResponse = resp + .json() + .await + .map_err(|e| format!("github token parse failed: {e}"))?; + Ok(body.access_token) +} + +pub async fn fetch_user( + client: &Client, + api_base: &str, + access_token: &str, +) -> Result { + let resp = client + .get(format!("{}/user", api_base.trim_end_matches('/'))) + .header("Accept", "application/vnd.github+json") + .header("User-Agent", "sorter2") + .bearer_auth(access_token) + .send() + .await + .map_err(|e| format!("github user request failed: {e}"))?; + + if !resp.status().is_success() { + return Err(format!("github user HTTP {}", resp.status())); + } + + resp.json() + .await + .map_err(|e| format!("github user parse failed: {e}")) +} + +pub fn provider_id(user: &GitHubUser) -> String { + user.id.to_string() +} + +pub fn validate_pseudonym(raw: &str) -> Result { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return Err("enter a name"); + } + if trimmed.len() > 64 { + return Err("too long"); + } + if !trimmed + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-') + { + return Err("letters, numbers, _ and - only"); + } + Ok(trimmed.to_string()) +} + +pub fn sanitize_pseudonym(login: &str) -> String { + validate_pseudonym(login).unwrap_or_else(|_| "user".to_string()) +} diff --git a/server/src/auth/session.rs b/server/src/auth/session.rs new file mode 100644 index 0000000000000000000000000000000000000000..09659240b9455c6fca12db5652e1d31cf8c2acfc --- /dev/null +++ b/server/src/auth/session.rs @@ -0,0 +1,240 @@ +//! Session cookie resolution and durable session CRUD. + +use axum_extra::extract::cookie::{Cookie, CookieJar, SameSite}; +use durable::{Db, Durability}; +use rand::Rng; + +use crate::{ + auth::config::AUTH_RETURN_COOKIE, + fetch::now_ms, + identity::{DEFAULT_ACTOR_UUID, DEFAULT_PSEUDONYM}, + storage_dto::{SessionDataV1, SESSION_DATA_VERSION}, + storage_schema::{delete_session, load_session, user_trust_weight, write_session}, +}; + +pub const SESSION_COOKIE: &str = "sorter2_session"; +pub const OAUTH_STATE_COOKIE: &str = "sorter2_oauth_state"; + +/// Session lifetime (30 days). +pub const SESSION_TTL_MS: i64 = 30 * 24 * 60 * 60 * 1000; + +pub fn session_has_pseudonym(session: &SessionDataV1) -> bool { + !session.current_pseudonym.trim().is_empty() +} + +pub fn load_valid_session(db: &Db, session_id: &str) -> Option { + let session = load_session(db, session_id).ok()??; + if session.expires_at <= now_ms() { + return None; + } + Some(session) +} + +#[derive(Debug, Clone)] +pub struct VoteActor { + pub pseudonym: String, + pub trust_weight: f64, +} + +impl VoteActor { + pub fn anon() -> Self { + Self { + pseudonym: DEFAULT_PSEUDONYM.to_string(), + trust_weight: 1.0, + } + } +} + +#[derive(Debug, Clone)] +pub struct SessionActor { + pub session_id: String, + pub uuid: String, + pub pseudonym: String, + pub trust_weight: f64, + pub expires_at: i64, +} + +pub fn new_session_id() -> String { + let mut bytes = [0u8; 32]; + rand::thread_rng().fill(&mut bytes); + hex_encode(&bytes) +} + +pub fn new_oauth_state() -> String { + let mut bytes = [0u8; 16]; + rand::thread_rng().fill(&mut bytes); + hex_encode(&bytes) +} + +fn hex_encode(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +pub fn resolve_vote_actor(db: &Db, session_id: Option<&str>) -> VoteActor { + let Some(session_id) = session_id else { + return VoteActor::anon(); + }; + let Ok(Some(session)) = load_session(db, session_id) else { + return VoteActor::anon(); + }; + if session.expires_at <= now_ms() { + return VoteActor::anon(); + } + let trust_weight = user_trust_weight(db, &session.uuid).unwrap_or(1.0); + VoteActor { + pseudonym: session.current_pseudonym, + trust_weight, + } +} + +pub fn load_session_actor(db: &Db, session_id: &str) -> Option { + let session = load_session(db, session_id).ok()??; + if session.expires_at <= now_ms() { + return None; + } + let trust_weight = user_trust_weight(db, &session.uuid).ok()?; + Some(SessionActor { + session_id: session_id.to_string(), + uuid: session.uuid, + pseudonym: session.current_pseudonym, + trust_weight, + expires_at: session.expires_at, + }) +} + +pub fn create_session( + db: &Db, + uuid: &str, + pseudonym: &str, +) -> Result<(String, SessionDataV1), String> { + let session_id = new_session_id(); + let expires_at = now_ms() + SESSION_TTL_MS; + let data = SessionDataV1 { + version: SESSION_DATA_VERSION, + uuid: uuid.to_string(), + current_pseudonym: pseudonym.to_string(), + expires_at, + }; + let mut batch = db.batch(); + write_session(&mut batch, &session_id, &data); + batch + .commit_with(Durability::SyncWal) + .map_err(|e| e.to_string())?; + Ok((session_id, data)) +} + +pub fn update_session_pseudonym( + db: &Db, + session_id: &str, + pseudonym: &str, +) -> Result<(), String> { + let mut session = load_session(db, session_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| "session not found".to_string())?; + if session.expires_at <= now_ms() { + return Err("session expired".to_string()); + } + session.current_pseudonym = pseudonym.to_string(); + let mut batch = db.batch(); + write_session(&mut batch, session_id, &session); + batch + .commit_with(Durability::SyncWal) + .map_err(|e| e.to_string()) +} + +pub fn destroy_session(db: &Db, session_id: &str) -> Result<(), String> { + let mut batch = db.batch(); + delete_session(&mut batch, session_id); + batch + .commit_with(Durability::SyncWal) + .map_err(|e| e.to_string()) +} + +pub fn session_cookie_value(session_id: &str) -> Cookie<'static> { + Cookie::build((SESSION_COOKIE, session_id.to_string())) + .http_only(true) + .same_site(SameSite::Lax) + .path("/") + .build() +} + +pub fn clear_session_cookie() -> Cookie<'static> { + Cookie::build((SESSION_COOKIE, "")) + .http_only(true) + .same_site(SameSite::Lax) + .path("/") + .removal() + .build() +} + +pub fn oauth_state_cookie_value(state: &str) -> Cookie<'static> { + Cookie::build((OAUTH_STATE_COOKIE, state.to_string())) + .http_only(true) + .same_site(SameSite::Lax) + .path("/") + .build() +} + +pub fn clear_oauth_state_cookie() -> Cookie<'static> { + Cookie::build((OAUTH_STATE_COOKIE, "")) + .http_only(true) + .same_site(SameSite::Lax) + .path("/") + .removal() + .build() +} + +pub fn auth_return_cookie_value(return_to: &str) -> Cookie<'static> { + Cookie::build((AUTH_RETURN_COOKIE, return_to.to_string())) + .http_only(true) + .same_site(SameSite::Lax) + .path("/") + .build() +} + +pub fn clear_auth_return_cookie() -> Cookie<'static> { + Cookie::build((AUTH_RETURN_COOKIE, "")) + .http_only(true) + .same_site(SameSite::Lax) + .path("/") + .removal() + .build() +} + +pub fn auth_return_from_jar(jar: &CookieJar) -> Option { + jar.get(AUTH_RETURN_COOKIE).map(|c| c.value().to_string()) +} + +pub fn session_id_from_jar(jar: &CookieJar) -> Option { + jar.get(SESSION_COOKIE).map(|c| c.value().to_string()) +} + +pub fn oauth_state_from_jar(jar: &CookieJar) -> Option { + jar.get(OAUTH_STATE_COOKIE).map(|c| c.value().to_string()) +} + +pub fn actor_uuid_for_vote(db: &Db, session_id: Option<&str>) -> String { + let Some(session_id) = session_id else { + return DEFAULT_ACTOR_UUID.to_string(); + }; + load_session(db, session_id) + .ok() + .flatten() + .filter(|s| s.expires_at > now_ms()) + .map(|s| s.uuid) + .unwrap_or_else(|| DEFAULT_ACTOR_UUID.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn missing_session_falls_back_to_anon() { + let dir = tempfile::tempdir().unwrap(); + let db = Db::open(dir.path()).unwrap(); + let actor = resolve_vote_actor(&db, None); + assert_eq!(actor.pseudonym, DEFAULT_PSEUDONYM); + assert_eq!(actor.trust_weight, 1.0); + } +} diff --git a/server/src/events.rs b/server/src/events.rs index 015208311f6c5c23a0e8aab068d002a69c89c4e1..8d4c1e260f9c034f5f7ec07484dda17efd69ecff 100644 --- a/server/src/events.rs +++ b/server/src/events.rs @@ -31,6 +31,9 @@ pub fn event_timestamp(event: &Event) -> i64 { match event { Event::VoteRecorded { ts, .. } => *ts, Event::NodeEnsured { .. } => crate::fetch::now_ms(), + Event::PrincipalCreated { ts, .. } => *ts, + Event::OauthLinked { ts, .. } => *ts, + Event::PseudonymClaimed { ts, .. } => *ts, } } @@ -57,6 +60,24 @@ pub enum Event { }, /// Register a node path in the fractal tree (no external fetch). NodeEnsured { id: String }, + + /// New trust anchor (first identity event for a human). + PrincipalCreated { uuid: String, ts: i64 }, + + /// OAuth provider account linked to an existing UUID. + OauthLinked { + uuid: String, + provider: String, + provider_id: String, + ts: i64, + }, + + /// Display pseudonym claimed by a UUID (global uniqueness enforced at apply). + PseudonymClaimed { + uuid: String, + pseudonym: String, + ts: i64, + }, } impl Event { diff --git a/server/src/html/mod.rs b/server/src/html/mod.rs index 1e2e7a06856d8a62378741aaf5ed94a4ffed337e..46d18b87f1313bf0aeb29955d18f291961057509 100644 --- a/server/src/html/mod.rs +++ b/server/src/html/mod.rs @@ -126,7 +126,7 @@ pub fn now_ms() -> i64 { t.as_millis() as i64 } -fn layout(title: &str, body: Markup, views: u64) -> Markup { +pub(crate) fn layout(title: &str, body: Markup, views: u64) -> Markup { let ver = asset_version(); let css_href = format!("/static/sorter.css?v={ver}"); let js_src = format!("/static/sorter_ui.js?v={ver}"); @@ -144,6 +144,9 @@ fn layout(title: &str, body: Markup, views: u64) -> Markup { @if views > 0 { span class="view-meta muted" { (views) " views" } } + nav class="top-nav" { + a href="/login" { "login" } + } div id="errors" {} (body) script src=(js_src) {} diff --git a/server/src/lib.rs b/server/src/lib.rs index da6e33e3dd6d79967bbee7a2708a7d982c93b88c..84f2565b105fb302241b949af64bd5e49916eab2 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -1,4 +1,5 @@ pub mod api; +pub mod auth; pub mod event_log; pub mod events; pub mod fetch; @@ -42,6 +43,12 @@ pub fn create_app(state: AppState) -> Router { .route("/~/*item_path", get(crate::html::browse)) .route("/", get(crate::html::home)) .route("/vote", get(crate::html::vote::vote_page)) + .route("/login", get(crate::auth::login_page)) + .route("/login/alias", get(crate::auth::alias_page)) + .route("/auth/github", get(crate::auth::github_start)) + .route("/auth/github/callback", get(crate::auth::github_callback)) + .route("/auth/logout", post(crate::auth::logout)) + .route("/auth/switch", post(crate::auth::switch_pseudonym)) .route("/ui", post(crate::api::ui_html::post_ui_html)) .with_state(state) .layer(TraceLayer::new_for_http()) diff --git a/server/src/projection_apply.rs b/server/src/projection_apply.rs index f2859c7e0e718c5bd3d332224eb05d7a8c65a816..4fb4b48ee0a53b7f673fae0c9731eed5acc33332 100644 --- a/server/src/projection_apply.rs +++ b/server/src/projection_apply.rs @@ -7,11 +7,13 @@ use crate::{ event_log::EventLogError, events::{Event, EventRecord}, - identity::resolve_actor_uuid, + auth::identity::{trust_weight_after_link, BASE_TRUST_WEIGHT}, path_types::ItemId, projection_store::ProjectionStore, reducer::VoteData, - storage_schema::{ensure_path_writes, vote_writes}, + storage_schema::{ + ensure_path_writes, oauth_link_key, pseudonym_owner, vote_writes, Store, StoreFields, + }, }; fn parse_event_id(id: &str) -> Result { @@ -72,7 +74,7 @@ pub fn apply_records( *trust_weight, ) .ok_or_else(|| EventLogError::Apply(format!("invalid vote event: {a} vs {b}")))?; - let actor_uuid = resolve_actor_uuid(db, pseudonym) + let actor_uuid = crate::identity::resolve_actor_uuid(db, pseudonym) .map_err(|e| EventLogError::Apply(e))?; let parent = parent_from_event_scope(scope); vote_writes(&mut batch, &parent, &vote, &actor_uuid) @@ -82,6 +84,72 @@ pub fn apply_records( let parsed = parse_event_id(id)?; ensure_path_writes(&mut batch, &parsed); } + Event::PrincipalCreated { uuid, .. } => { + batch.write( + Store::root() + .user_weights() + .key(&uuid.clone()) + .set(&BASE_TRUST_WEIGHT), + ); + } + Event::OauthLinked { + uuid, + provider, + provider_id, + .. + } => { + let link_key = oauth_link_key(provider, provider_id); + if let Some(existing) = Store::root() + .oauth_links() + .key(&link_key) + .get(db) + .map_err(|e| EventLogError::Apply(e.to_string()))? + { + if existing != *uuid { + return Err(EventLogError::Apply(format!( + "oauth link {link_key} already owned by {existing}" + ))); + } + } else { + batch.write(Store::root().oauth_links().key(&link_key).set(uuid)); + let current = Store::root() + .user_weights() + .key(&uuid.clone()) + .get(db) + .map_err(|e| EventLogError::Apply(e.to_string()))? + .unwrap_or(BASE_TRUST_WEIGHT); + batch.write( + Store::root() + .user_weights() + .key(&uuid.clone()) + .set(&trust_weight_after_link(current)), + ); + } + } + Event::PseudonymClaimed { uuid, pseudonym, .. } => { + if let Some(owner) = pseudonym_owner(db, pseudonym) + .map_err(|e| EventLogError::Apply(e.to_string()))? + { + if owner != *uuid { + return Err(EventLogError::Apply(format!( + "pseudonym {pseudonym} already claimed by {owner}" + ))); + } + } else { + batch.write( + Store::root() + .pseudonyms() + .key(&pseudonym.clone()) + .set(uuid), + ); + batch + .push( + &Store::root().user_pseudonyms().key(&uuid.clone()), + &pseudonym.clone(), + ) + .map_err(|e| EventLogError::Apply(e.to_string()))?; + } + } } last_seq = record.seq; } @@ -93,3 +161,105 @@ pub fn apply_records( Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + events::EventRecord, + identity::resolve_actor_uuid, + projection_store::ProjectionStore, + storage_schema::{oauth_link_owner, user_trust_weight, StoreFields}, + }; + + fn record(seq: u64, event: Event) -> EventRecord { + EventRecord::new(seq, crate::events::event_timestamp(&event), event) + } + + #[test] + fn identity_events_project_pseudonym_and_oauth_link() { + let dir = tempfile::tempdir().unwrap(); + let db = durable::Db::open(dir.path()).unwrap(); + let store = ProjectionStore::from_db(&db).unwrap(); + let uuid = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"; + let ts = 1; + + apply_records( + &store, + &[ + record( + 1, + Event::PrincipalCreated { + uuid: uuid.into(), + ts, + }, + ), + record( + 2, + Event::OauthLinked { + uuid: uuid.into(), + provider: "github".into(), + provider_id: "42".into(), + ts, + }, + ), + record( + 3, + Event::PseudonymClaimed { + uuid: uuid.into(), + pseudonym: "octocat".into(), + ts, + }, + ), + ], + ) + .unwrap(); + + assert_eq!( + oauth_link_owner(store.db(), "github", "42").unwrap(), + Some(uuid.to_string()) + ); + assert_eq!(resolve_actor_uuid(store.db(), "octocat").unwrap(), uuid); + assert_eq!(user_trust_weight(store.db(), uuid).unwrap(), 1.5); + let aliases = Store::root() + .user_pseudonyms() + .key(&uuid.to_string()) + .iter(store.db()) + .unwrap(); + assert_eq!(aliases, vec!["octocat".to_string()]); + } + + #[test] + fn pseudonym_claim_rejects_second_owner() { + let dir = tempfile::tempdir().unwrap(); + let db = durable::Db::open(dir.path()).unwrap(); + let store = ProjectionStore::from_db(&db).unwrap(); + + apply_records( + &store, + &[record( + 1, + Event::PseudonymClaimed { + uuid: "uuid-a".into(), + pseudonym: "taken".into(), + ts: 1, + }, + )], + ) + .unwrap(); + + let err = apply_records( + &store, + &[record( + 2, + Event::PseudonymClaimed { + uuid: "uuid-b".into(), + pseudonym: "taken".into(), + ts: 2, + }, + )], + ) + .unwrap_err(); + assert!(err.to_string().contains("already claimed")); + } +} diff --git a/server/src/projection_store.rs b/server/src/projection_store.rs index 8c1a466183a96173fe52b144fb67c2454b715fbd..37b4d1e01ca0aec0b079a936aeeae5306c7d80ca 100644 --- a/server/src/projection_store.rs +++ b/server/src/projection_store.rs @@ -18,7 +18,7 @@ use crate::{ const PROJECTION_CURSOR_KEY: &str = "cursor"; const PROJECTION_SCHEMA_KEY: &str = "schema_version"; -const PROJECTION_SCHEMA_VERSION: u64 = 5; +const PROJECTION_SCHEMA_VERSION: u64 = 6; #[derive(Debug, thiserror::Error)] pub enum ProjectionStoreError { diff --git a/server/src/state.rs b/server/src/state.rs index dcb82ff4beeaac8f0820de0e0131ca1f6bd81dcc..86949c4d1a40719e3b5bb114387c1fdb8f57d7ca 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -227,6 +227,7 @@ impl AppState { b: &str, ratio_left: i32, ratio_right: i32, + actor: &crate::auth::VoteActor, ) -> Result<(), String> { let ts = crate::html::now_ms(); let a_raw = a.trim(); @@ -251,17 +252,49 @@ impl AppState { return Err("invalid vote: need two distinct items".to_string()); } - let event = Event::vote_recorded( + let event = Event::VoteRecorded { ts, - a_id.as_str(), - b_id.as_str(), - left, - right, - parent.as_str(), - ); + a: a_id.as_str().to_string(), + b: b_id.as_str().to_string(), + ratio_left: left, + ratio_right: right, + scope: parent.as_str().to_string(), + pseudonym: actor.pseudonym.clone(), + trust_weight: actor.trust_weight, + }; self.journal.append(event).await } + + /// Append identity events (OAuth link, pseudonym claim, etc.). + pub async fn append_identity_events(&self, events: Vec) -> Result<(), String> { + self.journal.append_many(events).await + } + + pub async fn claim_pseudonym( + &self, + uuid: &str, + pseudonym: &str, + ) -> Result<(), String> { + let ts = crate::html::now_ms(); + self.journal + .append(Event::PseudonymClaimed { + uuid: uuid.to_string(), + pseudonym: pseudonym.to_string(), + ts, + }) + .await + } + + /// Session id for the seeded default pseudonym (tests and local dev helpers). + pub fn create_default_session(&self) -> Result { + crate::auth::session::create_session( + self.projection_store.db(), + crate::identity::DEFAULT_ACTOR_UUID, + crate::identity::DEFAULT_PSEUDONYM, + ) + .map(|(id, _)| id) + } } #[cfg(test)] @@ -497,7 +530,7 @@ mod tests { .await; let err = state - .record_vote(&ItemId::root(), "alpha", "beta", 0, 0) + .record_vote(&ItemId::root(), "alpha", "beta", 0, 0, &crate::auth::VoteActor::anon()) .await .unwrap_err(); assert!(err.contains("positive preference")); @@ -517,7 +550,14 @@ mod tests { .await; state - .record_vote(&ItemId::root(), "alpha", "beta", 2, 1) + .record_vote( + &ItemId::root(), + "alpha", + "beta", + 2, + 1, + &crate::auth::VoteActor::anon(), + ) .await .unwrap(); @@ -611,7 +651,14 @@ mod tests { }; let second = AppState::new(cfg).await; second - .record_vote(&ItemId::root(), "alpha", "gamma", 3, 1) + .record_vote( + &ItemId::root(), + "alpha", + "gamma", + 3, + 1, + &crate::auth::VoteActor::anon(), + ) .await .unwrap(); diff --git a/server/src/storage_dto.rs b/server/src/storage_dto.rs index 22f5ac498ae3a5db4347f3fe94830539c7c01e3e..db8a3094454df66f6d0e0e6c9369017c014784be 100644 --- a/server/src/storage_dto.rs +++ b/server/src/storage_dto.rs @@ -13,6 +13,16 @@ use crate::{ pub const VOTE_RECORD_VERSION: u32 = 2; pub const ENTITY_DATA_VERSION: u32 = 1; +pub const SESSION_DATA_VERSION: u32 = 1; + +/// Browser session stored in durable (operational; not event-logged). +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SessionDataV1 { + pub version: u32, + pub uuid: String, + pub current_pseudonym: String, + pub expires_at: i64, +} #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Versioned { diff --git a/server/src/storage_schema.rs b/server/src/storage_schema.rs index fe2671b876f3df77fdfd402dbc845ff1eb3512cd..b942810afd96d3bf01b7765718303a39be4ef8d2 100644 --- a/server/src/storage_schema.rs +++ b/server/src/storage_schema.rs @@ -14,7 +14,7 @@ use crate::{ reducer::{EntityData, NodeState, ScopeVotes, VoteData, UuidVoteKey, uuid_vote_key}, storage_dto::{ decode_entity_data, decode_vote, encode_entity_data, encode_vote, parse_stored_id, - StoredEntityDataV1, StoredVoteV1, + SessionDataV1, StoredEntityDataV1, StoredVoteV1, }, }; @@ -34,12 +34,75 @@ pub struct NodeSchema { #[allow(dead_code)] pub struct Store { pub nodes: Map, + pub sessions: Map>, + pub oauth_links: Map>, pub pseudonyms: Map>, + pub user_pseudonyms: Map>>, + pub user_weights: Map>, pub proj_meta: Map>, pub view_counts: Map>, pub view_meta: Map>, } +pub fn encode_session(data: &SessionDataV1) -> SessionDataV1 { + data.clone() +} + +pub fn decode_session(data: SessionDataV1) -> SessionDataV1 { + data +} + +pub fn oauth_link_key(provider: &str, provider_id: &str) -> String { + format!("{provider}:{provider_id}") +} + +pub fn user_trust_weight(db: &Db, uuid: &str) -> durable::Result { + Ok(Store::root() + .user_weights() + .key(&uuid.to_string()) + .get(db)? + .unwrap_or(1.0)) +} + +pub fn load_session(db: &Db, session_id: &str) -> durable::Result> { + Store::root() + .sessions() + .key(&session_id.to_string()) + .get(db) +} + +pub fn write_session(batch: &mut Batch, session_id: &str, data: &SessionDataV1) { + batch.write( + Store::root() + .sessions() + .key(&session_id.to_string()) + .set(&encode_session(data)), + ); +} + +pub fn delete_session(batch: &mut Batch, session_id: &str) { + batch.write( + Store::root() + .sessions() + .key(&session_id.to_string()) + .delete(), + ); +} + +pub fn pseudonym_owner(db: &Db, pseudonym: &str) -> durable::Result> { + Store::root() + .pseudonyms() + .key(&pseudonym.to_string()) + .get(db) +} + +pub fn oauth_link_owner(db: &Db, provider: &str, provider_id: &str) -> durable::Result> { + Store::root() + .oauth_links() + .key(&oauth_link_key(provider, provider_id)) + .get(db) +} + pub const RECENT_VOTES_CAP: u64 = 200; fn id_key(id: &ItemId) -> String { diff --git a/server/src/ui_action.rs b/server/src/ui_action.rs index 227ab3f6e8e9cce1532454450773f65daa24bfe7..a90bf80b191873700e030ee30591576ec6b08cf9 100644 --- a/server/src/ui_action.rs +++ b/server/src/ui_action.rs @@ -44,6 +44,14 @@ pub enum HtmlUiAction { #[serde(default)] kind: FetchTarget, }, + /// Live alias availability check (alias chooser page). + CheckPseudonym { pseudonym: String }, + /// Claim first alias after OAuth, then redirect. + ClaimPseudonym { + pseudonym: String, + #[serde(default)] + return_to: String, + }, } #[derive(Debug, Error)] diff --git a/server/static/sorter_ui.js b/server/static/sorter_ui.js index 5d6d765f69f301722e59248c76e919043559fb78..67a6ad7b5a9770290066d648d17feea3716bde54 100644 --- a/server/static/sorter_ui.js +++ b/server/static/sorter_ui.js @@ -162,8 +162,36 @@ update(); } + function initAliasInput() { + var input = document.getElementById('alias-input'); + var form = document.getElementById('alias-check-form'); + var claimField = document.getElementById('alias-claim-field'); + if (!input || !form) return; + var timer; + function syncClaimField() { + if (claimField) claimField.value = input.value || ''; + } + function queueCheck() { + syncClaimField(); + clearTimeout(timer); + timer = setTimeout(function () { + postUiForm(form); + }, 250); + } + input.addEventListener('input', queueCheck); + syncClaimField(); + } + + document.addEventListener('input', function (e) { + if (e.target && e.target.id === 'alias-input') { + var claimField = document.getElementById('alias-claim-field'); + if (claimField) claimField.value = e.target.value || ''; + } + }); + function initSorterUi() { initVoteSlider(); + initAliasInput(); document.addEventListener('submit', async function (e) { var f = e.target; if (!f || f.tagName !== 'FORM') return; diff --git a/server/tests/integration_ui.rs b/server/tests/integration_ui.rs index 40e4cfe1eec36dad29a075fb01aefb4dffd856d0..11ff32489f5feff280657948b1c5a10de5a6d91e 100644 --- a/server/tests/integration_ui.rs +++ b/server/tests/integration_ui.rs @@ -3,12 +3,13 @@ use std::net::SocketAddr; use axum::Router; use sorter2_server::{ + auth::session::SESSION_COOKIE, create_app, create_app_state, path_types::ItemId, state::AppConfig, ui_action::UI_RPC_FIELD, }; use tempfile::TempDir; use tokio::net::TcpListener; -async fn start_test_server() -> (SocketAddr, TempDir) { +async fn start_test_server() -> (SocketAddr, TempDir, String) { let tmp = TempDir::new().unwrap(); let data = tmp.path().to_string_lossy().into_owned(); let cfg = AppConfig { @@ -18,6 +19,8 @@ async fn start_test_server() -> (SocketAddr, TempDir) { port: 0, }; let state = create_app_state(cfg).await; + let session_id = state.create_default_session().unwrap(); + let session_cookie = format!("{SESSION_COOKIE}={session_id}"); let app: Router = create_app(state); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -25,12 +28,12 @@ async fn start_test_server() -> (SocketAddr, TempDir) { tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); - (addr, tmp) + (addr, tmp, session_cookie) } #[tokio::test] async fn post_ui_vote_compare_morphs_edge_history() { - let (addr, _tmp) = start_test_server().await; + let (addr, _tmp, session_cookie) = start_test_server().await; let parent = "reddit.com/r/rust"; let a = "reddit.com/r/rust/comments/aaa/announcing_rust_199"; let b = "reddit.com/r/rust/comments/bbb/what_are_you_working_on"; @@ -53,6 +56,7 @@ async fn post_ui_vote_compare_morphs_edge_history() { let client = reqwest::Client::new(); let body = client .post(format!("http://{addr}/ui")) + .header("Cookie", &session_cookie) .form(&form) .send() .await @@ -85,7 +89,7 @@ async fn post_ui_vote_compare_morphs_edge_history() { #[tokio::test] async fn post_ui_record_vote_morphs_ranking_and_persists() { - let (addr, tmp) = start_test_server().await; + let (addr, tmp, session_cookie) = start_test_server().await; let rpc = serde_json::json!({ "action": "record_vote", "a": "alpha", @@ -100,6 +104,7 @@ async fn post_ui_record_vote_morphs_ranking_and_persists() { let client = reqwest::Client::new(); let body = client .post(format!("http://{addr}/ui")) + .header("Cookie", &session_cookie) .form(&form) .send() .await @@ -139,7 +144,7 @@ async fn post_ui_record_vote_morphs_ranking_and_persists() { #[tokio::test] async fn browse_url_renders_subreddit_page() { - let (addr, _tmp) = start_test_server().await; + let (addr, _tmp, _session_cookie) = start_test_server().await; let client = reqwest::Client::new(); let html = client .get(format!("http://{addr}/~/https://reddit.com/r/rust")) @@ -155,7 +160,7 @@ async fn browse_url_renders_subreddit_page() { #[tokio::test] async fn vote_page_renders_live_ranking_sidebar() { - let (addr, _tmp) = start_test_server().await; + let (addr, _tmp, session_cookie) = start_test_server().await; let client = reqwest::Client::new(); let seed_rpc = serde_json::json!({ "action": "record_vote", @@ -169,6 +174,7 @@ async fn vote_page_renders_live_ranking_sidebar() { form.insert(UI_RPC_FIELD.to_string(), seed_rpc); client .post(format!("http://{addr}/ui")) + .header("Cookie", &session_cookie) .form(&form) .send() .await @@ -199,7 +205,7 @@ async fn vote_page_renders_live_ranking_sidebar() { #[tokio::test] async fn post_ui_parse_query_redirects_to_subreddit() { - let (addr, _tmp) = start_test_server().await; + let (addr, _tmp, _session_cookie) = start_test_server().await; let rpc = serde_json::json!({ "action": "parse_query", "query": "r/rust" diff --git a/test/auth_login.clj b/test/auth_login.clj new file mode 100644 index 0000000000000000000000000000000000000000..a81e064ebb88fe359c56cf6718880dc1e4c1a7d9 --- /dev/null +++ b/test/auth_login.clj @@ -0,0 +1,76 @@ +(ns test.auth-login + (:require [clojure.string :as str] + [clojure.test :refer [deftest is testing]] + [com.blockether.spel.core :as core] + [com.blockether.spel.locator :as loc] + [com.blockether.spel.page :as page] + [test.support.harness :as harness] + [test.support.seed-auth :as seed-auth])) + +(defn- move-vote-slider-left [pg] + (page/evaluate pg + "(() => { const s = document.getElementById('vote-preference-slider'); if (!s) return; s.value = '20'; s.dispatchEvent(new Event('input', { bubbles: true })); })()")) + +(defn- type-alias! [pg text] + (page/evaluate pg + (.replace + "(() => { const i = document.getElementById('alias-input'); const f = document.getElementById('alias-check-form'); if (!i || !f) return; + i.value = __TEXT__; + const cf = document.getElementById('alias-claim-field'); if (cf) cf.value = i.value; + return fetch(f.action, { method: 'POST', credentials: 'same-origin', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: new URLSearchParams(new FormData(f)).toString() }) + .then(function (r) { return r.text(); }) + .then(function (t) { eval(t); }); })()" + "__TEXT__" + (pr-str text))) + (Thread/sleep 400)) + +(defn- element-text [pg test-id] + (let [raw (page/evaluate pg + (str "document.querySelector('[data-testid=\"" test-id "\"]')?.textContent || ''"))] + (when (string? raw) (str/trim raw)))) + +(defn- wait-for-text [pg test-id text timeout-ms] + (let [deadline (+ (System/currentTimeMillis) timeout-ms)] + (loop [] + (let [got (or (element-text pg test-id) "")] + (cond + (= got text) got + (< (System/currentTimeMillis) deadline) (do (Thread/sleep 200) (recur)) + :else (throw (ex-info "timeout waiting for text" {:test-id test-id :want text :got got}))))))) + +(deftest new-user-login-flow-returns-to-vote-pair + (testing "anonymous vote redirects through OAuth + alias chooser back to the same pair" + (let [servers (harness/with-auth-servers + (fn [data-dir] + (seed-auth/write-seeder-events! (str data-dir "/events.jsonl"))))] + (try + (harness/seed-rust-children! (:app-base servers)) + (let [vote-url (seed-auth/seeder-pair-vote-url (:app-base servers)) + alias "newbie-alias"] + (core/with-testing-page [pg] + (page/navigate pg vote-url) + (page/wait-for-selector pg "#vote-compare-form") + (move-vote-slider-left pg) + (loc/click (page/get-by-test-id pg "vote-post")) + (page/wait-for-selector pg "[data-testid=oauth-github]" {:timeout 15000}) + (is (str/includes? (or (element-text pg "alias-list-empty") "") "none yet")) + (loc/click (page/get-by-test-id pg "oauth-github")) + (page/wait-for-selector pg "[data-testid=alias-input]" {:timeout 15000}) + (type-alias! pg "seeder") + (wait-for-text pg "alias-status" "taken" 15000) + (type-alias! pg alias) + (wait-for-text pg "alias-status" "available" 15000) + (loc/click (page/get-by-test-id pg "alias-claim")) + (page/wait-for-selector pg "#vote-compare-form" {:timeout 15000}) + (is (str/includes? (page/url pg) "/vote?")) + (loc/click (page/get-by-test-id pg "vote-post")) + (page/wait-for-selector pg ".vote-edge-history-title" {:timeout 15000}) + (let [history (or (element-text pg "#vote-edge-history-region") "")] + (is (str/includes? history "votes on this pair")) + (is (str/includes? history "3:1") + "seeded seeder vote still visible") + (is (not (str/includes? history "no votes on this pair yet")))))) + (finally + ((:stop servers))))))) diff --git a/test/support/harness.clj b/test/support/harness.clj new file mode 100644 index 0000000000000000000000000000000000000000..3f05951f418258642dcacb4a10ccccc8bfbe8748 --- /dev/null +++ b/test/support/harness.clj @@ -0,0 +1,104 @@ +(ns test.support.harness + (:require [babashka.process :as process] + [clojure.java.io :as io] + [clojure.string :as str] + [test.support.mock-oauth :as mock-oauth] + [test.support.mock-reddit :as mock-reddit])) + +(defn repo-root [] + (.getCanonicalPath (io/file (System/getProperty "user.dir")))) + +(defn pick-port [] + (with-open [s (java.net.ServerSocket. 0)] + (.getLocalPort s))) + +(defn wait-health [base-url ms] + (let [deadline (+ (System/currentTimeMillis) ms) + url (str base-url "/healthz")] + (loop [] + (let [resp (try + (process/shell {:out :string :err :string} + "curl" "-sf" url) + (catch Exception _ nil))] + (if (and resp (zero? (:exit resp)) (= "ok" (str/trim (:out resp "")))) + true + (if (< (System/currentTimeMillis) deadline) + (do (Thread/sleep 200) (recur)) + false)))))) + +(defn curl-fetch-children [base item] + (process/shell {:out :string :err :string} + "curl" "-sfN" "--max-time" "20" + "-X" "POST" (str base "/ui") + "--data-urlencode" + (str "__rpc__={\"action\":\"fetch_entity\",\"item\":\"" item + "\",\"kind\":\"children\"}"))) + +(defn app-env + [data-dir app-port oauth-port reddit-port] + (into (into {} (System/getenv)) + {"SORTER2_SKIP_DOTENV" "1" + "SORTER2_DATA_DIR" data-dir + "SORTER2_EVENT_LOG" (str data-dir "/events.jsonl") + "SORTER2_VIEWS_LOG" (str data-dir "/views.jsonl") + "PORT" (str app-port) + "SORTER2_BASE_URL" (str "http://127.0.0.1:" app-port) + "GITHUB_CLIENT_ID" "test-client" + "GITHUB_CLIENT_SECRET" "test-secret" + "GITHUB_OAUTH_BASE" (str "http://127.0.0.1:" oauth-port) + "GITHUB_API_BASE" (str "http://127.0.0.1:" oauth-port) + "REDDIT_API_BASE" (str "http://127.0.0.1:" reddit-port) + "REDDIT_OAUTH_BASE" (str "http://127.0.0.1:" reddit-port) + "REDDIT_CLIENT_ID" "" + "REDDIT_CLIENT_SECRET" "" + "REDDIT_APP_ID" "" + "REDDIT_APP_SECRET" ""})) + +(defn with-auth-servers + "Start mock Reddit + mock OAuth + release sorter2-server. + `seed-fn` is `(fn [data-dir] ...)` called before the app boots. + Returns `{:stop ... :app-base ...}`." + [seed-fn] + (let [root (repo-root) + fixtures (mock-reddit/fixtures-dir root) + data-dir (.getAbsolutePath + (doto (io/file (System/getProperty "java.io.tmpdir") + (str "sorter2-auth-" (System/currentTimeMillis))) + (.mkdirs))) + reddit-port (pick-port) + oauth-port (pick-port) + app-port (pick-port) + app-base (str "http://127.0.0.1:" app-port) + bin (str root "/target/release/sorter2-server") + stop-mock-reddit (mock-reddit/start-mock-reddit reddit-port fixtures) + stop-mock-oauth (mock-oauth/start-mock-oauth oauth-port)] + (seed-fn data-dir) + (process/shell {:dir root} + "cargo" "build" "--release" "--package" "sorter2-server") + (let [proc (process/process {:dir root + :env (app-env data-dir app-port oauth-port reddit-port) + :out :string + :err :string} + bin)] + (when-not (wait-health app-base 25000) + (process/destroy proc) + (stop-mock-oauth) + (stop-mock-reddit) + (throw (ex-info "app healthz timeout" {:app-base app-base}))) + {:stop (fn [] + (process/destroy proc) + (stop-mock-oauth) + (stop-mock-reddit)) + :app-base app-base + :data-dir data-dir}))) + +(defn oauth-login-url + [app-base return-to mock-user] + (str app-base "/auth/github?return_to=" + (java.net.URLEncoder/encode return-to "UTF-8") + "&mock_user=" (java.net.URLEncoder/encode mock-user "UTF-8"))) + +(defn seed-rust-children! [app-base] + (let [fetch (curl-fetch-children app-base "reddit.com/r/rust")] + (when-not (zero? (:exit fetch)) + (throw (ex-info "fetch rust children failed" {:err (:err fetch)}))))) diff --git a/test/support/mock_oauth.clj b/test/support/mock_oauth.clj new file mode 100644 index 0000000000000000000000000000000000000000..909d7a7be8b159ea54a122d69c46af3db3318118 --- /dev/null +++ b/test/support/mock_oauth.clj @@ -0,0 +1,84 @@ +(ns test.support.mock-oauth + "In-process HTTP stub for GitHub OAuth (authorize, token, /user)." + (:require [clojure.string :as str]) + (:import [com.sun.net.httpserver HttpServer HttpHandler HttpExchange] + [java.net InetSocketAddress URLDecoder])) + +(defn- query-param [query key] + (when query + (some (fn [pair] + (let [[k v] (str/split pair "=" 2)] + (when (= k key) + (URLDecoder/decode (or v "") "UTF-8")))) + (str/split query #"&")))) + +(defn- parse-mock-user [raw] + (let [s (or raw "1002:newbie") + [id login] (str/split s #":" 2)] + {:id (Long/parseLong id) + :login (or login "newbie")})) + +(defn- send-json [^HttpExchange ex status body] + (let [bytes (.getBytes body "UTF-8")] + (.set (.getResponseHeaders ex) "Content-Type" "application/json") + (.sendResponseHeaders ex status (alength bytes)) + (doto (.getResponseBody ex) + (.write bytes) + (.close)))) + +(defn- send-redirect [^HttpExchange ex location] + (.set (.getResponseHeaders ex) "Location" location) + (.sendResponseHeaders ex 302 -1) + (.close (.getResponseBody ex))) + +(defn- read-form-code [^HttpExchange ex] + (let [body (slurp (.getInputStream ex))] + (query-param body "code"))) + +(defn- bearer-token [^HttpExchange ex] + (some-> (.getRequestHeaders ex) + (.getFirst "Authorization") + (str/replace #"^[Bb]earer " ""))) + +(defn- parse-token-user [token] + (when (str/starts-with? token "mock:") + (parse-mock-user (subs token 5)))) + +(defn start-mock-oauth + "Start mock GitHub OAuth on `port`. Returns a zero-arg `stop` function." + [port] + (let [server (HttpServer/create (InetSocketAddress. "127.0.0.1" port) 0) + handler + (proxy [HttpHandler] [] + (handle [^HttpExchange exchange] + (let [uri (.getRequestURI exchange) + path (.getPath uri) + query (.getQuery uri)] + (cond + (str/ends-with? path "/login/oauth/authorize") + (let [redirect-uri (query-param query "redirect_uri") + state (query-param query "state") + mock-user (query-param query "mock_user") + user (parse-mock-user mock-user) + code (str "mock:" (:id user) ":" (:login user)) + loc (str redirect-uri "?code=" (java.net.URLEncoder/encode code "UTF-8") + "&state=" (java.net.URLEncoder/encode state "UTF-8"))] + (send-redirect exchange loc)) + + (str/ends-with? path "/login/oauth/access_token") + (let [code (or (read-form-code exchange) "mock:1002:newbie")] + (send-json exchange 200 (str "{\"access_token\":\"" code "\",\"token_type\":\"bearer\"}"))) + + (= path "/user") + (let [token (bearer-token exchange) + user (or (parse-token-user token) {:id 1002 :login "newbie"})] + (send-json exchange 200 + (str "{\"id\":" (:id user) ",\"login\":\"" (:login user) "\"}"))) + + :else + (send-json exchange 404 "{\"error\":\"not found\"}")))))] + (.createContext server "/" handler) + (.setExecutor server nil) + (.start server) + (fn stop [] + (.stop server 0)))) diff --git a/test/support/seed_auth.clj b/test/support/seed_auth.clj new file mode 100644 index 0000000000000000000000000000000000000000..88061ceb15daabd2032cc11192ae19c8579cbe5a --- /dev/null +++ b/test/support/seed_auth.clj @@ -0,0 +1,41 @@ +(ns test.support.seed-auth + "Append-only event-log seeds for auth integration tests.") + +(def seeder-uuid "00000000-0000-0000-0000-000000000010") + +(def rust-scope "https://reddit.com/r/rust") +(def post-a "https://reddit.com/r/rust/comments/aaa") +(def post-b "https://reddit.com/r/rust/comments/bbb") + +(defn- esc [s] + (.replace s "\\" "\\\\")) + +(defn- line [seq ts event-json] + (str "{\"schema\":2,\"seq\":" seq ",\"ts\":" ts ",\"event\":" event-json "}" "\n")) + +(defn seeder-vote-events + "Events that register a seeder principal and one vote on the rust A/B pair." + [] + [(line 1 1 (str "{\"type\":\"principal_created\",\"uuid\":\"" (esc seeder-uuid) "\",\"ts\":1}")) + (line 2 2 (str "{\"type\":\"oauth_linked\",\"uuid\":\"" (esc seeder-uuid) + "\",\"provider\":\"github\",\"provider_id\":\"1001\",\"ts\":2}")) + (line 3 3 (str "{\"type\":\"pseudonym_claimed\",\"uuid\":\"" (esc seeder-uuid) + "\",\"pseudonym\":\"seeder\",\"ts\":3}")) + (line 4 4 (str "{\"type\":\"node_ensured\",\"id\":\"" (esc rust-scope) "\"}")) + (line 5 5 (str "{\"type\":\"node_ensured\",\"id\":\"" (esc post-a) "\"}")) + (line 6 6 (str "{\"type\":\"node_ensured\",\"id\":\"" (esc post-b) "\"}")) + (line 7 7 (str "{\"type\":\"vote_recorded\",\"ts\":7" + ",\"a\":\"" (esc post-a) "\",\"b\":\"" (esc post-b) "\"" + ",\"ratio_left\":3,\"ratio_right\":1" + ",\"scope\":\"" (esc rust-scope) "\"" + ",\"pseudonym\":\"seeder\",\"trust_weight\":1.5}"))]) + +(defn write-seeder-events! + [event-log-path] + (spit event-log-path (apply str (seeder-vote-events)))) + +(defn seeder-pair-vote-url [app-base] + (str app-base "/vote?parent=" + (java.net.URLEncoder/encode rust-scope "UTF-8") + "&left=" (java.net.URLEncoder/encode post-a "UTF-8") + "&right=" (java.net.URLEncoder/encode post-b "UTF-8"))) diff --git a/test/vote_compare.clj b/test/vote_compare.clj index bc0be9f90bb2eccbc756b20cbdf7212ddfed9cbc..8a4f39bb5a1c188b3612e8a5c95aeccd0be9d16d 100644 --- a/test/vote_compare.clj +++ b/test/vote_compare.clj @@ -1,101 +1,36 @@ (ns test.vote-compare - (:require [babashka.process :as process] - [clojure.java.io :as io] - [clojure.string :as str] + (:require [clojure.string :as str] [clojure.test :refer [deftest is testing]] [com.blockether.spel.core :as core] [com.blockether.spel.locator :as loc] [com.blockether.spel.page :as page] - [test.support.mock-reddit :as mock-reddit]) - (:import [java.net URLEncoder])) - -(defn- repo-root [] - (.getCanonicalPath (io/file (System/getProperty "user.dir")))) - -(defn- pick-port [] - (with-open [s (java.net.ServerSocket. 0)] - (.getLocalPort s))) - -(defn- wait-health [base-url ms] - (let [deadline (+ (System/currentTimeMillis) ms) - url (str base-url "/healthz")] - (loop [] - (let [resp (try - (process/shell {:out :string :err :string} - "curl" "-sf" url) - (catch Exception _ nil))] - (if (and resp (zero? (:exit resp)) (= "ok" (str/trim (:out resp "")))) - true - (if (< (System/currentTimeMillis) deadline) - (do (Thread/sleep 200) (recur)) - false)))))) - -(defn- curl-fetch-children [base item] - (process/shell {:out :string :err :string} - "curl" "-sfN" "--max-time" "20" - "-X" "POST" (str base "/ui") - "--data-urlencode" - (str "__rpc__={\"action\":\"fetch_entity\",\"item\":\"" item - "\",\"kind\":\"children\"}"))) - -(defn- vote-page-url [base parent] - (str base "/vote?parent=" - (URLEncoder/encode parent "UTF-8"))) + [test.support.harness :as harness] + [test.support.seed-auth :as seed-auth])) (deftest vote-compare-shows-recorded-vote-after-post - (testing "post vote on /vote morphs edge history (mock Reddit children seeded)" - (let [root (repo-root) - fixtures (mock-reddit/fixtures-dir root) - data-dir (.getAbsolutePath - (doto (io/file (System/getProperty "java.io.tmpdir") - (str "sorter2-vote-" (System/currentTimeMillis))) - (.mkdirs))) - reddit-port (pick-port) - app-port (pick-port) - reddit-base (str "http://127.0.0.1:" reddit-port) - app-base (str "http://127.0.0.1:" app-port) - bin (str root "/target/release/sorter2-server") - stop-mock (mock-reddit/start-mock-reddit reddit-port fixtures)] + (testing "post vote on /vote morphs edge history (mock Reddit + auth session)" + (let [servers (harness/with-auth-servers + (fn [data-dir] + (seed-auth/write-seeder-events! (str data-dir "/events.jsonl"))))] (try - (is (zero? (:exit (process/shell {:dir root} - "cargo" "build" "--release" "--package" "sorter2-server"))) - "release build succeeds") - (let [proc (process/process {:dir root - :env (into (into {} (System/getenv)) - {"SORTER2_SKIP_DOTENV" "1" - "SORTER2_DATA_DIR" data-dir - "SORTER2_EVENT_LOG" (str data-dir "/events.jsonl") - "PORT" (str app-port) - "REDDIT_API_BASE" reddit-base - "REDDIT_OAUTH_BASE" reddit-base - "REDDIT_CLIENT_ID" "" - "REDDIT_CLIENT_SECRET" "" - "REDDIT_APP_ID" "" - "REDDIT_APP_SECRET" ""}) - :out :string - :err :string} - bin)] - (try - (is (wait-health app-base 20000) "app healthz") - (let [fetch (curl-fetch-children app-base "reddit.com/r/rust")] - (is (zero? (:exit fetch)) "fetch posts via mock Reddit") - (is (str/includes? (:out fetch) "Idiomorph.morph"))) - (core/with-testing-page [pg] - (page/navigate pg (vote-page-url app-base "reddit.com/r/rust")) - (page/wait-for-selector pg "#vote-compare-form") - (let [before (loc/text-content (page/locator pg "#vote-edge-history-region"))] - (is (str/includes? before "no votes on this pair yet") - "empty edge history before first vote")) + (harness/seed-rust-children! (:app-base servers)) + (let [vote-url (seed-auth/seeder-pair-vote-url (:app-base servers))] + (core/with-testing-page [pg] + (page/navigate pg (harness/oauth-login-url (:app-base servers) "/" "1001:seeder")) + (page/wait-for-selector pg ".top-nav" {:timeout 15000}) + (page/navigate pg vote-url) + (page/wait-for-selector pg "#vote-compare-form") + (let [before (loc/text-content (page/locator pg "#vote-edge-history-region"))] + (is (str/includes? before "votes on this pair") + "seeded seeder vote visible before our vote") (loc/click (page/get-by-test-id pg "vote-post")) (page/wait-for-selector pg ".vote-edge-history-title") (let [after (loc/text-content (page/locator pg "#vote-edge-history-region"))] (is (str/includes? after "votes on this pair") "shows edge history title after vote") - (is (str/includes? after "1:1") - "shows submitted ratio after vote (default slider at center)") + (is (not= before after) + "edge history updated after authenticated vote") (is (not (str/includes? after "no votes on this pair yet")) - "does not revert to empty edge history"))) - (finally - (process/destroy proc)))) + "does not revert to empty edge history"))))) (finally - (stop-mock)))))) + ((:stop servers)))))))