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: [00be3a29] invite system Side A — unified diff (full patch): diff --git a/bb.edn b/bb.edn index 50be8232847e672b3f273a2fb25ddd1d12adb7e2..818f850765370d12d58f239e287764fc3649d78b 100644 --- a/bb.edn +++ b/bb.edn @@ -47,14 +47,16 @@ "RUST_LOG" "info"})})))} test - {:doc "Full test suite: integration + auth + grants" + {:doc "Full test suite: integration + auth + grants + invites" :requires ([test.integration :as integration] [test.auth :as auth] - [test.grants :as grants]) + [test.grants :as grants] + [test.invites :as invites]) :task (do (integration/integration) (auth/auth-test) - (grants/grants-test))} + (grants/grants-test) + (invites/invites-test))} perf {:doc "Performance test: concurrent HTTP requests to detect blocking I/O" diff --git a/cli/src/main.rs b/cli/src/main.rs index e5833b0b93d8e667b94c574ba2b0f8cb758ff3df..8eda9f485bd1f7392f1e34be27176c21e20354eb 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -105,6 +105,23 @@ enum ScopedCmd { #[arg(long)] json: bool, }, + + /// Mint a shareable invite link (24h TTL, in-memory until redeemed). Requires Manage on the room. + InviteLink { + /// Comma-separated: view, post, vote, add_item, manage + #[arg(long = "caps", value_delimiter = ',')] + caps: Vec, + #[arg(long, default_value_t = 1)] + uses: usize, + #[arg(long)] + json: bool, + }, + + /// List principals granted access in this room (requires View or Manage) + Audit { + #[arg(long)] + json: bool, + }, } #[derive(Subcommand, Debug)] @@ -514,17 +531,35 @@ fn print_thread(resp: &ThreadDetailResponse) { .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_millis() as i64; - if resp.total > resp.posts.len() { - let end = resp.offset + resp.posts.len(); - eprintln!("# showing {}-{} of {} posts (--offset N --limit N to paginate)", resp.offset, end.saturating_sub(1), resp.total); + if resp.total > resp.items.len() { + let end = resp.offset + resp.items.len(); + eprintln!( + "# showing {}-{} of {} rows (--offset N --limit N to paginate)", + resp.offset, + end.saturating_sub(1), + resp.total + ); } - for (i, post) in resp.posts.iter().enumerate() { - let timeago = slug_types::timeago::timeago_compact(now_ms, post.ts); - let body = &post.body.trim(); - println!("", post.index, timeago); - println!("{}", body); - println!(""); - if i + 1 < resp.posts.len() { + for (i, item) in resp.items.iter().enumerate() { + match item { + ThreadItem::Post { + index, + ts, + body, + .. + } => { + let timeago = slug_types::timeago::timeago_compact(now_ms, *ts); + let body = body.trim(); + println!("", index, timeago); + println!("{}", body); + println!(""); + } + ThreadItem::System { ts, text } => { + let timeago = slug_types::timeago::timeago_compact(now_ms, *ts); + println!("{}", timeago, text.trim()); + } + } + if i + 1 < resp.items.len() { println!(); println!(); } @@ -1036,6 +1071,95 @@ async fn run_scoped(base: &str, room: &str, sub: ScopedCmd) -> Result<()> { } } }, + ScopedCmd::InviteLink { caps, uses, json } => { + let caps: Vec = caps + .into_iter() + .flat_map(|s| { + s.split(',') + .map(|p| p.trim().to_lowercase()) + .filter(|p| !p.is_empty()) + .collect::>() + }) + .collect(); + if caps.is_empty() { + return Err(anyhow!("--caps is required (e.g. --caps view,post,vote)")); + } + let bearer = effective_bearer().ok_or_else(|| { + anyhow!( + "no bearer token: run `slugsocial identity start --rig --model ` \ + then `slugsocial identity poll `, or set SLUG_BEARER_TOKEN / ~/.config/slugsocial/token" + ) + })?; + let batch = send_rpc( + &client, + base, + Some(&bearer), + vec![RpcCommand::RoomMintInvite { + room: room.to_string(), + capabilities: caps, + max_uses: uses, + }], + ) + .await?; + match rpc_line_ok(&batch.results[0])? { + RpcResult::RoomInviteMinted { + invite_url, + expires_at_ms, + max_uses, + } => { + if json { + println!( + "{}", + serde_json::to_string_pretty(&serde_json::json!({ + "invite_url": invite_url, + "expires_at_ms": expires_at_ms, + "max_uses": max_uses, + }))? + ); + } else { + println!("{invite_url}"); + println!("(Expires in 24 hours. Max uses: {max_uses})"); + } + } + _ => return Err(anyhow!("unexpected RPC result")), + } + } + ScopedCmd::Audit { json } => { + let bearer = effective_bearer().ok_or_else(|| { + anyhow!( + "no bearer token: run `slugsocial identity start --rig --model ` \ + then `slugsocial identity poll `, or set SLUG_BEARER_TOKEN / ~/.config/slugsocial/token" + ) + })?; + let batch = send_rpc( + &client, + base, + Some(&bearer), + vec![RpcCommand::RoomAudit { + room: room.to_string(), + }], + ) + .await?; + match rpc_line_ok(&batch.results[0])? { + RpcResult::RoomAudit(resp) => { + if json { + println!("{}", serde_json::to_string_pretty(&resp)?); + } else { + println!("room {}", resp.room); + if resp.grants.is_empty() { + println!("(no grants recorded)"); + } else { + let w_user = resp.grants.iter().map(|g| g.username.len()).max().unwrap_or(0); + for g in &resp.grants { + let caps = g.capabilities.join(", "); + println!("{: return Err(anyhow!("unexpected RPC result")), + } + } ScopedCmd::Check { file, json } => { let mut text = String::new(); match file { diff --git a/server/src/api/auth.rs b/server/src/api/auth.rs index 995ce4a61d29b024c399c656134f541ecfd880cf..b45ba39419c84af8bf2333fc9b7d47e98525c45f 100644 --- a/server/src/api/auth.rs +++ b/server/src/api/auth.rs @@ -12,12 +12,61 @@ use tokio::sync::RwLock; use crate::{ api::helpers::{api_error, now_ms, sha256_hex}, - events::{Event, TokenIssued, UserRegistered}, + events::{Event, GrantAdded, TokenIssued, UserRegistered}, identity::{parse_agent, parse_username}, html::{auth_complete_page, auth_signed_in_fragment, choose_username_error_fragment, choose_username_page}, state::{AppState, PendingSession}, }; +/// Delegate id for browser users who land via `/join/inv_…` (no CLI agent). +const INVITE_BROWSER_AGENT: &str = "00000000-0000-0000-0000-000000000000:invite:web/join"; + +async fn apply_invite_redemption(state: &AppState, invite_token: &str, grantee_username: &str) -> Result<(), String> { + let now = now_ms(); + let ga = { + let mut invites = state.invites.write().await; + let Some(inv) = invites.get_mut(invite_token) else { + return Err("invite not found".into()); + }; + if now > inv.expires_at_ms { + invites.remove(invite_token); + return Err("invite expired".into()); + } + if inv.current_uses >= inv.max_uses { + return Err("invite exhausted".into()); + } + inv.current_uses += 1; + Event::GrantAdded(GrantAdded { + ts: now, + room_id: inv.room_id.clone(), + username: grantee_username.to_string(), + capabilities: inv.capabilities.clone(), + granted_by: inv.inviter.clone(), + }) + }; + + match state.event_log.append(&ga).await { + Ok(()) => { + let mut reduced = state.reduced.write().await; + reduced.apply_event(ga); + let mut invites = state.invites.write().await; + if let Some(inv) = invites.get(invite_token) { + if inv.current_uses >= inv.max_uses { + invites.remove(invite_token); + } + } + Ok(()) + } + Err(e) => { + let mut invites = state.invites.write().await; + if let Some(inv) = invites.get_mut(invite_token) { + inv.current_uses = inv.current_uses.saturating_sub(1); + } + Err(format!("{e}")) + } + } +} + fn pending_sessions(state: &AppState) -> Arc>> { state.pending_sessions.clone() } @@ -115,6 +164,42 @@ pub struct AuthLoginQuery { pub session: String, } +pub async fn get_join_invite(Path(token): Path, State(state): State) -> impl IntoResponse { + let token = token.trim().to_string(); + if token.is_empty() { + return api_error(StatusCode::NOT_FOUND, "invite invalid or expired", None).into_response(); + } + let now = now_ms(); + let valid = { + let invites = state.invites.read().await; + match invites.get(&token) { + None => false, + Some(inv) => now <= inv.expires_at_ms && inv.current_uses < inv.max_uses, + } + }; + if !valid { + return api_error(StatusCode::NOT_FOUND, "invite invalid or expired", None).into_response(); + } + + let session = format!("p_{}", uuid::Uuid::new_v4().simple()); + let s = PendingSession { + agent: INVITE_BROWSER_AGENT.to_string(), + created_ts: now_ms(), + provider: None, + provider_id: None, + redeem_invite: Some(token), + complete: None, + }; + state.pending_sessions.write().await.insert(session.clone(), s); + + let public_url = std::env::var("SLUG_PUBLIC_URL").unwrap_or_else(|_| "http://127.0.0.1:8080".to_string()); + Redirect::temporary(&format!( + "{public_url}/auth/login?session={}", + urlencoding::encode(&session) + )) + .into_response() +} + pub async fn get_auth_login(Query(q): Query, State(state): State) -> impl IntoResponse { // Redirect to Google auth endpoint. let sessions = pending_sessions(&state); @@ -205,6 +290,7 @@ pub async fn get_auth_callback(Query(q): Query, State(state): s.provider = Some("google".to_string()); s.provider_id = Some(sub.clone()); if let Some(username) = existing { + let invite_tok = s.redeem_invite.clone(); let (bearer, token_event) = issue_token_for_user(&username); // append token event let ev = Event::TokenIssued(token_event); @@ -215,6 +301,11 @@ pub async fn get_auth_callback(Query(q): Query, State(state): let mut reduced = reduced_arc.write().await; reduced.apply_event(ev); } + if let Some(tok) = invite_tok { + if let Err(e) = apply_invite_redemption(&state, &tok, &username).await { + tracing::warn!(error = %e, "invite redemption skipped after oauth"); + } + } s.complete = Some((username, bearer)); return Redirect::temporary(&format!("{public_url}/auth/complete")).into_response(); } @@ -310,6 +401,16 @@ pub async fn post_choose_username( reduced.apply_event(ti_ev.clone()); } + // Redeem invite (if any) before marking the session complete. + if let Some(tok) = { + let sessions_read = sessions.read().await; + sessions_read.get(&form.session).and_then(|s| s.redeem_invite.clone()) + } { + if let Err(e) = apply_invite_redemption(&state, &tok, &canon_user).await { + tracing::warn!(error = %e, "invite redemption skipped after registration"); + } + } + // Mark complete for polling. { let mut sessions_write = sessions.write().await; @@ -339,6 +440,7 @@ pub async fn post_pending_session( created_ts: now_ms(), provider: None, provider_id: None, + redeem_invite: None, complete: None, }; let sessions = pending_sessions(&state); diff --git a/server/src/api/mod.rs b/server/src/api/mod.rs index a1ea432932cdca0e42dc7377f0a25075fd5644c0..cb031aecebad1107dfa2898a98fb6084b28ba6e9 100644 --- a/server/src/api/mod.rs +++ b/server/src/api/mod.rs @@ -4,6 +4,7 @@ mod rpc; mod validate; pub use auth::{ + get_join_invite, get_pending_session, get_whoami, post_pending_session, diff --git a/server/src/api/rpc.rs b/server/src/api/rpc.rs index 11a76ed7a02aae588edadb7a87a43660f5669d9e..f6bbc3df71909a2da7403cd46fe4ea6ca130c692 100644 --- a/server/src/api/rpc.rs +++ b/server/src/api/rpc.rs @@ -13,12 +13,14 @@ use slug_types::*; use crate::{ canonical_path::{canonicalize_item, canonicalize_tag}, dsl, - events::{AgentBound, Event, GrantAdded, Ingest, RoomCreated, ThreadCapability, ThreadVisibility}, + events::{ + AgentBound, Event, GrantAdded, Ingest, RoomCreated, ThreadCapability, ThreadVisibility, + }, identity::{parse_agent, parse_username}, path_types::CanonicalItemUrl, ranking::{connected_components_from_voted_pairs, ranked_items_subset}, reducer::{scope_from_room_wire, ReducerState, ScopeId}, - state::AppState, + state::{AppState, InviteState}, }; use super::auth::verify_bearer_principal; @@ -142,6 +144,27 @@ fn gen_short_id() -> String { (0..7).map(|_| ALPHABET[rng.gen_range(0..ALPHABET.len())] as char).collect() } +fn gen_invite_token() -> String { + use rand::Rng; + const ALPHABET: &[u8] = b"0123456789abcdefghijklmnopqrstuvwxyz"; + let mut rng = rand::thread_rng(); + let tail: String = (0..16).map(|_| ALPHABET[rng.gen_range(0..ALPHABET.len())] as char).collect(); + format!("inv_{tail}") +} + +const INVITE_TTL_MS: i64 = 86_400_000; + +fn capability_wire(c: ThreadCapability) -> String { + match c { + ThreadCapability::View => "view", + ThreadCapability::Post => "post", + ThreadCapability::Vote => "vote", + ThreadCapability::AddItem => "add_item", + ThreadCapability::Manage => "manage", + } + .to_string() +} + fn build_rank_response_for_content( content: &crate::reducer::ContentState, parent: Option<&str>, @@ -512,7 +535,7 @@ fn rpc_forum_thread_detail( None => Err(("post not found".into(), None)), Some((idx, ing)) => Ok(ThreadDetailResponse { thread: format!("#{}", tag), - posts: vec![PostRow { + items: vec![ThreadItem::Post { id: ing.id.clone(), index: idx, ts: ing.ts, @@ -543,7 +566,7 @@ fn rpc_forum_thread_detail( let total = filtered.len(); const MAX_BODY: usize = 2000; - let posts: Vec = filtered + let items: Vec = filtered .into_iter() .skip(offset) .take(limit) @@ -553,7 +576,7 @@ fn rpc_forum_thread_detail( } else { (ing.raw.clone(), false) }; - PostRow { + ThreadItem::Post { id: ing.id.clone(), index: idx, ts: ing.ts, @@ -566,7 +589,7 @@ fn rpc_forum_thread_detail( Ok(ThreadDetailResponse { thread: format!("#{}", tag), - posts, + items, total, offset, }) @@ -1007,7 +1030,7 @@ pub async fn handle_rpc_batch( RpcCommand::RoomGrant { room, username, - capability, + capabilities, } => { let principal = { let reduced = state.reduced.read().await; @@ -1022,6 +1045,8 @@ pub async fn handle_rpc_batch( }; if !can_manage { line_err("requires Manage capability", None) + } else if capabilities.is_empty() { + line_err("capabilities must not be empty", None) } else { match parse_username(&username) { Err(msg) => line_err("invalid username", Some(msg)), @@ -1033,14 +1058,18 @@ pub async fn handle_rpc_batch( if !user_exists { line_err(format!("user @{target} not found"), None) } else { - match parse_capability(&capability) { + let caps: Result, String> = capabilities + .iter() + .map(|c| parse_capability(c.trim())) + .collect(); + match caps { Err(msg) => line_err(msg, None), - Ok(cap) => { + Ok(caps) => { let ga_ev = Event::GrantAdded(GrantAdded { ts: now_ms(), room_id: room, username: target, - capabilities: vec![cap], + capabilities: caps, granted_by: principal, }); if let Err(e) = state.event_log.append(&ga_ev).await { @@ -1059,6 +1088,114 @@ pub async fn handle_rpc_batch( } } } + RpcCommand::RoomMintInvite { + room, + capabilities, + max_uses, + } => { + let principal = { + let reduced = state.reduced.read().await; + verify_bearer_principal(&headers, &*reduced) + }; + match principal { + Err((_, m)) => line_err(m, None), + Ok(principal) => { + let can_manage = { + let reduced = state.reduced.read().await; + reduced.user_has_cap(&room, &principal, ThreadCapability::Manage) + }; + if !can_manage { + line_err("requires Manage capability", None) + } else if capabilities.is_empty() { + line_err("capabilities must not be empty", None) + } else { + match capabilities + .iter() + .map(|c| parse_capability(c.trim())) + .collect::, String>>() + { + Err(msg) => line_err(msg, None), + Ok(caps) => { + let max_uses = max_uses.max(1).min(100_000); + let now = now_ms(); + let expires_at_ms = now + INVITE_TTL_MS; + let token = loop { + let t = gen_invite_token(); + let taken = { + let invites = state.invites.read().await; + invites.contains_key(&t) + }; + if !taken { + break t; + } + }; + let inv = InviteState { + room_id: room.clone(), + capabilities: caps, + expires_at_ms, + max_uses, + current_uses: 0, + inviter: principal, + }; + state.invites.write().await.insert(token.clone(), inv); + let public_url = std::env::var("SLUG_PUBLIC_URL") + .unwrap_or_else(|_| "http://127.0.0.1:8080".to_string()); + let invite_url = format!("{public_url}/join/{token}"); + line_ok(RpcResult::RoomInviteMinted { + invite_url, + expires_at_ms: Some(expires_at_ms), + max_uses, + }) + } + } + } + } + } + } + RpcCommand::RoomAudit { room } => { + let principal = { + let reduced = state.reduced.read().await; + verify_bearer_principal(&headers, &*reduced) + }; + match principal { + Err((_, m)) => line_err(m, None), + Ok(principal) => { + let reduced = state.reduced.read().await; + if !reduced.rooms.contains_key(&room) { + line_err("unknown room", None) + } else { + let can_audit = reduced.user_has_cap(&room, &principal, ThreadCapability::View) + || reduced.user_has_cap(&room, &principal, ThreadCapability::Manage); + if !can_audit { + line_err("requires View or Manage capability", None) + } else { + let grants: Vec = reduced + .grants + .get(&room) + .map(|m| { + let mut v: Vec = m + .iter() + .map(|(username, caps)| { + let mut c: Vec = + caps.iter().copied().map(capability_wire).collect(); + c.sort(); + RoomAuditEntry { + username: username.clone(), + capabilities: c, + } + }) + .collect(); + v.sort_by(|a, b| a.username.cmp(&b.username)); + v + }) + .unwrap_or_default(); + line_ok(RpcResult::RoomAudit(RoomAuditResponse { room, grants })) + } + } + } + } + } + RpcCommand::RoomRevoke { .. } => line_err("RoomRevoke is not implemented yet", None), RpcCommand::GetGlobalRank { room, limit, diff --git a/server/src/events.rs b/server/src/events.rs index 125005d89ade7b22c9b6d997f0ed81781df5e7ec..9e60f2c218f5c871a93e855e415dd984e3a017c5 100644 --- a/server/src/events.rs +++ b/server/src/events.rs @@ -26,6 +26,8 @@ pub enum Event { RoomCreated(RoomCreated), GrantAdded(GrantAdded), GrantRevoked(GrantRevoked), + InviteMinted(InviteMinted), + InviteRedeemed(InviteRedeemed), /// Ingest of a DSL+prose body. Identity and routing live in event metadata. Ingest(Ingest), } @@ -82,6 +84,25 @@ pub struct GrantRevoked { pub revoked_by: String, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct InviteMinted { + pub ts: i64, + pub token: String, + pub room_id: String, + pub capabilities: Vec, + pub inviter: String, + pub max_uses: u32, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub expires_ts_ms: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct InviteRedeemed { + pub ts: i64, + pub token: String, + pub username: String, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct Ingest { /// Unix timestamp in milliseconds. diff --git a/server/src/lib.rs b/server/src/lib.rs index d00b56d909e1fabd1d122d054dd02785e13e0bd8..bf2f73d4d0f26a357961e88c52c3b4f72626af81 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -27,6 +27,7 @@ pub fn create_app(state: AppState) -> Router { Router::new() .route("/healthz", axum::routing::get(|| async { "ok" })) .route("/static/:filename", axum::routing::get(crate::html::serve_theme_css)) + .route("/join/:token", axum::routing::get(api::get_join_invite)) .route("/auth/login", axum::routing::get(api::get_auth_login)) .route("/auth/callback", axum::routing::get(api::get_auth_callback)) .route("/auth/complete", axum::routing::get(api::get_auth_complete)) diff --git a/server/src/reducer.rs b/server/src/reducer.rs index 118cb19f4df4a083df2878ca7a1424f3d63de851..5190d0a7ee3d545a786a68fef483322aafdce7f9 100644 --- a/server/src/reducer.rs +++ b/server/src/reducer.rs @@ -3,7 +3,7 @@ use std::collections::{HashMap, HashSet, VecDeque}; use serde::{Deserialize, Serialize}; use crate::canonical_path::canonicalize_tag; -use crate::events::{Event, Ingest, ThreadCapability}; +use crate::events::{Event, Ingest, ThreadCapability, ThreadVisibility}; use crate::path_types::CanonicalItemUrl; #[derive(Debug, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)] @@ -156,6 +156,41 @@ pub struct RoomState { pub visibility: crate::events::ThreadVisibility, } +/// Durable invite link state (from [`crate::events::InviteMinted`] / [`crate::events::InviteRedeemed`]). +#[derive(Debug, Clone)] +pub struct ActiveInviteState { + pub room_id: String, + pub capabilities: HashSet, + pub inviter: String, + pub uses_remaining: u32, + pub expires_ts_ms: Option, +} + +#[derive(Clone, Debug)] +pub enum RoomTimelineKind { + RoomCreated { + owner: String, + slug: String, + visibility: ThreadVisibility, + }, + GrantAdded { + username: String, + granted_by: String, + capabilities: Vec, + }, + GrantRevoked { + username: String, + revoked_by: String, + capabilities: Vec, + }, +} + +#[derive(Clone, Debug)] +pub struct RoomTimelineEntry { + pub ts: i64, + pub kind: RoomTimelineKind, +} + #[derive(Debug, Clone)] pub struct ForumThreadState { pub last_activity_ts: i64, @@ -210,6 +245,10 @@ pub struct ReducerState { pub ingests_ordered: Vec, /// room_id → username → capabilities pub grants: HashMap>>, + /// room_id → chronological room admin lines (for thread UI). + pub room_timeline: HashMap>, + /// Invite token → active invite (absent when fully consumed or never minted). + pub invites: HashMap, } impl ReducerState { @@ -225,6 +264,20 @@ impl ReducerState { .unwrap_or(false) } + /// Invite link is present, not expired, and has uses left. + pub fn invite_token_active(&self, token: &str, now_ms: i64) -> Option<&ActiveInviteState> { + let inv = self.invites.get(token)?; + if inv.uses_remaining == 0 { + return None; + } + if let Some(exp) = inv.expires_ts_ms { + if now_ms > exp { + return None; + } + } + Some(inv) + } + pub fn content_for_scope_mut(&mut self, scope: ScopeId) -> &mut ContentState { self.content.entry(scope).or_default() } @@ -356,6 +409,17 @@ impl ReducerState { visibility: rc.visibility, }, ); + self.room_timeline + .entry(rc.room_id.clone()) + .or_default() + .push(RoomTimelineEntry { + ts: rc.ts, + kind: RoomTimelineKind::RoomCreated { + owner: rc.owner.clone(), + slug: rc.slug.clone(), + visibility: rc.visibility, + }, + }); } Event::Ingest(mut ing) => { ing.thread_tag = canonicalize_tag(&ing.thread_tag); @@ -525,18 +589,31 @@ impl ReducerState { nav!(self.actor_last_post_ts, keypath(ing.principal.clone()), setval(ing.ts)); } Event::GrantAdded(ga) => { + let room_id = ga.room_id.clone(); let caps = self.grants .entry(ga.room_id) .or_default() - .entry(ga.username) + .entry(ga.username.clone()) .or_default(); - for cap in ga.capabilities { + for cap in ga.capabilities.iter().copied() { caps.insert(cap); } + self.room_timeline + .entry(room_id) + .or_default() + .push(RoomTimelineEntry { + ts: ga.ts, + kind: RoomTimelineKind::GrantAdded { + username: ga.username.clone(), + granted_by: ga.granted_by.clone(), + capabilities: ga.capabilities.clone(), + }, + }); } Event::GrantRevoked(gr) => { + let room_id = gr.room_id.clone(); if let Some(room_grants) = self.grants.get_mut(&gr.room_id) { - let username = gr.username; + let username = gr.username.clone(); if let Some(caps) = room_grants.get_mut(&username) { for cap in &gr.capabilities { caps.remove(cap); @@ -549,6 +626,37 @@ impl ReducerState { self.grants.remove(&gr.room_id); } } + self.room_timeline + .entry(room_id) + .or_default() + .push(RoomTimelineEntry { + ts: gr.ts, + kind: RoomTimelineKind::GrantRevoked { + username: gr.username.clone(), + revoked_by: gr.revoked_by.clone(), + capabilities: gr.capabilities.clone(), + }, + }); + } + Event::InviteMinted(im) => { + self.invites.insert( + im.token.clone(), + ActiveInviteState { + room_id: im.room_id.clone(), + capabilities: im.capabilities.iter().copied().collect(), + inviter: im.inviter.clone(), + uses_remaining: im.max_uses, + expires_ts_ms: im.expires_ts_ms, + }, + ); + } + Event::InviteRedeemed(ir) => { + if let Some(inv) = self.invites.get_mut(&ir.token) { + inv.uses_remaining = inv.uses_remaining.saturating_sub(1); + if inv.uses_remaining == 0 { + self.invites.remove(&ir.token); + } + } } } } @@ -570,6 +678,8 @@ impl Default for ReducerState { actor_last_post_ts: HashMap::new(), ingests_ordered: Vec::new(), grants: HashMap::new(), + room_timeline: HashMap::new(), + invites: HashMap::new(), } } } diff --git a/server/src/state.rs b/server/src/state.rs index b1ff2330903780cbe0cdf35964cb166eb2423d7d..628fd5921b34ba53ac0ca6b6fd33957ca09129b9 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -1,8 +1,20 @@ +use std::collections::HashMap; use std::sync::Arc; use tokio::sync::{broadcast, RwLock}; -use crate::{event_log::EventLog, reducer::ReducerState}; +use crate::{event_log::EventLog, events::ThreadCapability, reducer::ReducerState}; + +/// Ephemeral invite link (24h TTL, in-memory only; not written to the event log). +#[derive(Debug, Clone)] +pub struct InviteState { + pub room_id: String, + pub capabilities: Vec, + pub expires_at_ms: i64, + pub max_uses: usize, + pub current_uses: usize, + pub inviter: String, +} #[derive(Debug, Clone)] pub struct PendingSession { @@ -10,6 +22,8 @@ pub struct PendingSession { pub created_ts: i64, pub provider: Option, pub provider_id: Option, + /// When set, successful OAuth completion redeems this invite token and appends [`crate::events::GrantAdded`]. + pub redeem_invite: Option, pub complete: Option<(String /*username*/, String /*bearer*/ )>, } @@ -42,7 +56,9 @@ pub struct AppState { pub cfg: Arc, pub event_log: Arc, pub reduced: Arc>, - pub pending_sessions: Arc>>, + pub pending_sessions: Arc>>, + /// Ephemeral invite tokens (`inv_…`) until expiry or exhaustion. + pub invites: Arc>>, /// Broadcast channel for SSE live-streaming. Capacity = 64 events. pub stream_tx: broadcast::Sender, /// Broadcast channel for web SSE HTML fragments (poem pattern). Capacity = 64. @@ -58,7 +74,8 @@ impl AppState { cfg: Arc::new(cfg), event_log: Arc::new(event_log), reduced: Arc::new(RwLock::new(ReducerState::default())), - pending_sessions: Arc::new(RwLock::new(std::collections::HashMap::new())), + pending_sessions: Arc::new(RwLock::new(HashMap::new())), + invites: Arc::new(RwLock::new(HashMap::new())), stream_tx, html_tx, } diff --git a/server/src/timeline.rs b/server/src/timeline.rs new file mode 100644 index 0000000000000000000000000000000000000000..251158943ab36d3015268e35f6bdd01c3f42ab3b --- /dev/null +++ b/server/src/timeline.rs @@ -0,0 +1,153 @@ +//! Room admin lines merged into forum thread views. + +use crate::{ + canonical_path::canonicalize_tag, + reducer::{ReducerState, RoomTimelineEntry, RoomTimelineKind}, +}; + +fn cap_label(c: crate::events::ThreadCapability) -> &'static str { + use crate::events::ThreadCapability::*; + match c { + View => "view", + Post => "post", + Vote => "vote", + AddItem => "add_item", + Manage => "manage", + } +} + +fn caps_list(caps: &[crate::events::ThreadCapability]) -> String { + let mut v: Vec<_> = caps.iter().map(|c| cap_label(*c)).collect(); + v.sort(); + v.join(", ") +} + +/// Human-readable system line for the thread feed. +pub fn format_room_timeline_entry(e: &RoomTimelineEntry) -> String { + match &e.kind { + RoomTimelineKind::RoomCreated { + owner, + slug, + visibility, + } => { + let vis = match visibility { + crate::events::ThreadVisibility::Public => "public", + crate::events::ThreadVisibility::Private => "private", + }; + format!("@{owner} created room #{slug} ({vis})") + } + RoomTimelineKind::GrantAdded { + username, + granted_by, + capabilities, + } => { + format!( + "@{granted_by} granted @{} {}", + username, + caps_list(capabilities) + ) + } + RoomTimelineKind::GrantRevoked { + username, + revoked_by, + capabilities, + } => { + format!( + "@{revoked_by} revoked @{} {}", + username, + caps_list(capabilities) + ) + } + } +} + +#[derive(Clone, Debug)] +pub enum MergedThreadRow { + System { ts: i64, text: String }, + Post { + index: usize, + id: String, + ts: i64, + principal: String, + raw: String, + }, +} + +/// Merge room admin lines with thread ingests for one room + tag. Oldest first. +/// `actor_prefix` filters posts only (system lines always included). +pub fn merge_thread_rows( + reduced: &ReducerState, + room_wire: &str, + thread_tag: &str, + since: Option, + before: Option, + actor_prefix: &str, +) -> Vec { + let scope = crate::reducer::scope_from_room_wire(room_wire); + let tag = canonicalize_tag(thread_tag); + let key = (scope.clone(), tag.clone()); + + let mut rows: Vec = Vec::new(); + + if let Some(entries) = reduced.room_timeline.get(room_wire.trim()) { + for e in entries { + if since.map_or(true, |s| e.ts >= s) && before.map_or(true, |b| e.ts < b) { + rows.push(MergedThreadRow::System { + ts: e.ts, + text: format_room_timeline_entry(e), + }); + } + } + } + + let all_ids: Vec = reduced + .ingests_by_scope_thread + .get(&key) + .map(|q| q.iter().rev().cloned().collect()) + .unwrap_or_default(); + + for (idx, id) in all_ids.into_iter().enumerate() { + let Some(ing) = reduced.ingests_by_id.get(&id) else { + continue; + }; + if since.map_or(true, |s| ing.ts >= s) && before.map_or(true, |b| ing.ts < b) { + if !actor_prefix.is_empty() + && !ing + .principal + .to_lowercase() + .starts_with(actor_prefix) + { + continue; + } + rows.push(MergedThreadRow::Post { + index: idx, + id: ing.id.clone(), + ts: ing.ts, + principal: ing.principal.clone(), + raw: ing.raw.clone(), + }); + } + } + + rows.sort_by(|a, b| { + let ta = match a { + MergedThreadRow::System { ts, .. } | MergedThreadRow::Post { ts, .. } => *ts, + }; + let tb = match b { + MergedThreadRow::System { ts, .. } | MergedThreadRow::Post { ts, .. } => *ts, + }; + ta.cmp(&tb) + }); + rows +} + +/// Public forum thread (`room_wire == "public"`): same merge (timeline usually empty). +pub fn merge_public_thread_rows( + reduced: &ReducerState, + thread_tag: &str, + since: Option, + before: Option, + actor_prefix: &str, +) -> Vec { + merge_thread_rows(reduced, "public", thread_tag, since, before, actor_prefix) +} diff --git a/test/grants.bb b/test/grants.bb index 10ef0f08ce1c3945e5f1634314e6aebcf6810cd2..4ded5ce92576cdea653cce0ba8a751be4374731a 100644 --- a/test/grants.bb +++ b/test/grants.bb @@ -105,7 +105,7 @@ ;; Alice grants bob View only. (println "\nalice grants bob View only…") (assert! (rpc-line-ok? (:parsed (rpc-batch! base-url alice-token - [{"RoomGrant" {"room" room-id "username" "bob" "capability" "view"}}]))) + [{"RoomGrant" {"room" room-id "username" "bob" "capabilities" ["view"]}}]))) "grant View RPC ok") (println "\nbob (View only) tries to post prose…") @@ -117,7 +117,7 @@ ;; Alice grants bob Post. (println "\nalice grants bob Post…") (assert! (rpc-line-ok? (:parsed (rpc-batch! base-url alice-token - [{"RoomGrant" {"room" room-id "username" "bob" "capability" "post"}}]))) + [{"RoomGrant" {"room" room-id "username" "bob" "capabilities" ["post"]}}]))) "grant Post RPC ok") (println "\nbob (View + Post) posts prose…") @@ -143,7 +143,7 @@ ;; Alice grants bob Vote. (println "\nalice grants bob Vote…") (assert! (rpc-line-ok? (:parsed (rpc-batch! base-url alice-token - [{"RoomGrant" {"room" room-id "username" "bob" "capability" "vote"}}]))) + [{"RoomGrant" {"room" room-id "username" "bob" "capabilities" ["vote"]}}]))) "grant Vote RPC ok") (println "\nbob (View + Post + Vote) votes…") diff --git a/test/invites.bb b/test/invites.bb new file mode 100644 index 0000000000000000000000000000000000000000..f72b1e630a6856174cdef11eb977e2f81b16e551 --- /dev/null +++ b/test/invites.bb @@ -0,0 +1,150 @@ +(ns test.invites + "Ephemeral invite links: mint via RPC, GET /join → pending session + OAuth, redemption → GrantAdded, + RoomAudit, post succeeds, second GET /join returns 404 when max_uses exhausted." + (:require [babashka.fs :as fs] + [cheshire.core :as json] + [clojure.string :as str] + [test.common :as common] + [test.oauth :as oauth])) + +(def ^:private counts (atom {:pass 0 :fail 0})) + +(defn- assert! [pred msg] + (common/test-assert! counts pred msg)) + +(defn- bearer [token] {"Authorization" (str "Bearer " token)}) + +(defn- rpc-batch! [base-url token cmds] + (let [resp (oauth/http-post-json (str base-url "/api/v0/rpc") cmds :headers (bearer token))] + {:status (:status resp) + :parsed (json/parse-string (:body resp) false)})) + +(defn- rpc-line-ok? [parsed] + (true? (get-in parsed ["results" 0 "ok"]))) + +(defn- session-from-login-location [loc] + (when loc + (let [qpart (if (str/includes? loc "?") + (-> loc (str/split #"\?" 2) second) + "") + qpart (if (str/includes? qpart "#") + (-> qpart (str/split #"#" 2) first) + qpart) + m (oauth/parse-query qpart)] + (some-> (get m :session) str)))) + +(defn- invite-token-from-url [invite-url] + (second (re-find #"/join/(inv_[^/?#]+)" (str invite-url)))) + +(defn- register-user! [base-url session-agent username] + (oauth/complete-registration! base-url + :agent session-agent + :username username + :assert! (fn [pred msg] (assert! pred msg)))) + +(defn- ingest! [base-url token room thread delegate text] + (rpc-batch! base-url token + [{"Post" {"room" room + "thread_tag" thread + "delegate" delegate + "text" text + "return_rank_diff" false}}])) + +(defn invites-test [& _args] + (println "\n━━━ ephemeral invite + audit integration check ━━━\n") + (reset! counts {:pass 0 :fail 0}) + + (println "building server binary…") + (common/letlocals + (bind build (common/run-cargo-build-release! ["slugsocial-server"])) + (assert! (zero? (:exit build)) "cargo build succeeds") + (bind server-bin "target/release/slugsocial-server") + + (bind tmp-dir (str (fs/create-temp-dir {:prefix "slug-invites-"}))) + (bind slug-port (common/pick-port)) + (bind google-port (common/pick-port)) + (bind base-url (str "http://127.0.0.1:" slug-port)) + (bind google-url (str "http://127.0.0.1:" google-port)) + + (bind !server (atom nil)) + (bind !google (atom nil)) + + (bind server-env (common/slug-server-env tmp-dir base-url google-url slug-port)) + (try + (println (str "starting mock google on :" google-port)) + (reset! !google (oauth/start-mock-google google-port + :google-users ["google-user-alice" "google-user-bob"])) + + (println (str "starting server on :" slug-port)) + (reset! !server (common/start-server server-bin server-env)) + (assert! (common/wait-for-server base-url 10000) "server responds to /healthz") + + (println "\nregistering alice…") + (let [alice-token (register-user! base-url + "00000000-0000-0000-0000-000000000001:test:local/dev" + "alice") + + _ (println "\nalice creates private room…") + create (rpc-batch! base-url alice-token + [{"RoomCreate" {"slug" "invite-demo" "visibility" "private"}}]) + _ (assert! (= 200 (:status create)) "room create HTTP 200") + _ (assert! (rpc-line-ok? (:parsed create)) "room create RPC ok") + room-id (get-in (:parsed create) ["results" 0 "result" "RoomCreated" "room_id"]) + _ (assert! (some? room-id) "room_id present") + + _ (println "\nalice mints invite (view,post,vote uses=1)…") + mint (rpc-batch! base-url alice-token + [{"RoomMintInvite" {"room" room-id + "capabilities" ["view" "post" "vote"] + "max_uses" 1}}]) + _ (assert! (= 200 (:status mint)) "mint HTTP 200") + _ (assert! (rpc-line-ok? (:parsed mint)) "mint RPC ok") + invite-url (get-in (:parsed mint) ["results" 0 "result" "RoomInviteMinted" "invite_url"]) + inv-tok (invite-token-from-url invite-url) + _ (assert! (some? inv-tok) "invite token parsed from URL") + + _ (println "\nGET /join/:token (expect redirect + session)…") + join-resp (oauth/http-get-no-redirect (str base-url "/join/" inv-tok)) + _ (assert! (contains? #{302 307} (:status join-resp)) + (str "join returns redirect (got status " (:status join-resp) ")")) + sess (session-from-login-location (:location join-resp)) + _ (assert! (and (some? sess) (str/starts-with? sess "p_")) "Location carries session=p_…") + + _ (println "\nbob completes OAuth via invite session…") + bob-token (oauth/complete-pending-session! base-url sess "bob" + :assert! (fn [pred msg] (assert! pred msg))) + + _ (println "\nalice runs RoomAudit…") + audit (rpc-batch! base-url alice-token [{"RoomAudit" {"room" room-id}}]) + _ (assert! (rpc-line-ok? (:parsed audit)) "audit RPC ok") + grants (get-in (:parsed audit) ["results" 0 "result" "RoomAudit" "grants"]) + bob-entry (first (filter #(= "bob" (get % "username")) grants)) + _ (assert! (some? bob-entry) "audit lists bob") + bob-caps (set (get bob-entry "capabilities")) + _ (assert! (= bob-caps #{"view" "post" "vote"}) "bob has view, post, vote") + + _ (println "\nbob posts prose to private room…") + _ (assert! (rpc-line-ok? (:parsed (ingest! base-url bob-token room-id "main" + "00000000-0000-0000-0000-000000000002:test:local/dev" + "Hello via invite link."))) + "bob post succeeds") + + _ (println "\nsecond GET /join (invite exhausted → 404)…") + join2 (oauth/http-get-no-redirect (str base-url "/join/" inv-tok)) + _ (assert! (= 404 (:status join2)) "exhausted invite returns 404")] + + (println "\ninvite lifecycle OK.")) + + (finally + (when-some [s @!server] (common/kill-server s)) + (when-some [g @!google] ((:stop-fn g))) + (fs/delete-tree tmp-dir))) + + (bind {pass :pass fail :fail} @counts) + (if (zero? fail) + (println (str "\n" common/ansi-green "━━━ " pass " invite checks passed ━━━" common/ansi-reset "\n")) + (do (println (str "\n" common/ansi-red "━━━ " fail " invite checks FAILED ━━━" common/ansi-reset "\n")) + (System/exit 1))))) + +(when (= *file* (System/getProperty "babashka.file")) + (invites-test)) diff --git a/test/oauth.bb b/test/oauth.bb index b69459acd17c23c023170d8dd93fe45bb49c8d88..efb3f3e66346ffde81bf9622d75c3a6ec89f8db2 100644 --- a/test/oauth.bb +++ b/test/oauth.bb @@ -21,6 +21,21 @@ resp (.send (http-client) req (java.net.http.HttpResponse$BodyHandlers/ofString))] {:status (.statusCode resp) :body (.body resp) :headers (.map (.headers resp))}))) +(defn http-get-no-redirect + "GET without following redirects; returns `:location` from the first `Location` header when present." + [url & {:keys [headers]}] + (let [client (-> (java.net.http.HttpClient/newBuilder) + (.followRedirects java.net.http.HttpClient$Redirect/NEVER) + (.connectTimeout connect-timeout) + (.build)) + b (java.net.http.HttpRequest/newBuilder (java.net.URI/create url))] + (doseq [[k v] (or headers {})] + (.header b k v)) + (let [req (-> b (.timeout request-timeout) (.GET) (.build)) + resp (.send client req (java.net.http.HttpResponse$BodyHandlers/ofString)) + loc (first (get (.map (.headers resp)) "location"))] + {:status (.statusCode resp) :body (.body resp) :location loc}))) + (defn http-post-json [url data & {:keys [headers]}] (let [body (json/generate-string data) b (java.net.http.HttpRequest/newBuilder (java.net.URI/create url))] @@ -50,6 +65,35 @@ resp (.send (http-client) req (java.net.http.HttpResponse$BodyHandlers/ofString))] {:status (.statusCode resp) :body (.body resp) :headers (.map (.headers resp))}))) +(defn complete-pending-session! + "Finish OAuth for an existing pending session id (e.g. created by `GET /join/inv_…`). Returns bearer token." + [base-url session-id username & {:keys [assert!]}] + (let [check! (fn [pred msg resp] + (if assert! + (assert! pred msg) + (when-not pred + (throw (ex-info msg {:resp resp})))))] + (let [enc (java.net.URLEncoder/encode session-id "UTF-8") + login-url (str base-url "/auth/login?session=" enc) + login-get (http-get login-url)] + (check! (= 200 (:status login-get)) + (str "oauth redirect chain for session " session-id) + login-get) + (let [choose (http-post-form (str base-url "/auth/choose-username") + {:session session-id :username username})] + (check! (= 200 (:status choose)) + (str "choose-username for " username " returns 200") + choose) + (let [poll (http-get (str base-url "/api/v0/pending-session/" session-id))] + (check! (= 200 (:status poll)) + (str "pending-session poll returns 200") + poll) + (let [poll-json (json/parse-string (:body poll) true)] + (check! (:complete poll-json) + (str "pending session complete for " username) + poll) + (:token poll-json))))))) + (defn parse-query [s] (into {} (for [part (str/split (or s "") #"&") diff --git a/types/src/lib.rs b/types/src/lib.rs index a715fe5639f4a3a7bfedcc706b5fb312b98c9ac5..98c66a00fd27803ef6f75b5ac478ff2eb762d771 100644 --- a/types/src/lib.rs +++ b/types/src/lib.rs @@ -123,13 +123,32 @@ pub struct PathDetailResponse { #[derive(Debug, Serialize, Deserialize)] pub struct ThreadDetailResponse { pub thread: String, - pub posts: Vec, - /// Total posts in this thread. + /// Chronological page: prose posts and room system lines, oldest first within the window. + pub items: Vec, + /// Total rows (posts + system lines) in this thread after filters. pub total: usize, - /// Chronological offset of the first post in this page. + /// Offset into the merged chronological list. pub offset: usize, } +/// One row in a thread timeline: a normal post or a room system line. +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ThreadItem { + Post { + id: String, + index: usize, + ts: i64, + actor: String, + body: String, + truncated: bool, + }, + System { + ts: i64, + text: String, + }, +} + /// One post in a thread. Full body, no snippet. #[derive(Debug, Serialize, Deserialize)] pub struct PostRow { @@ -224,6 +243,23 @@ pub struct FeedPost { // RPC batch API (`POST /api/v0/rpc`) // --------------------------------------------------------------------------- +fn default_invite_max_uses() -> usize { + 1 +} + +/// One principal's capabilities in a private room (from [`RpcCommand::RoomAudit`]). +#[derive(Debug, Serialize, Deserialize)] +pub struct RoomAuditEntry { + pub username: String, + pub capabilities: Vec, +} + +#[derive(Debug, Serialize, Deserialize)] +pub struct RoomAuditResponse { + pub room: String, + pub grants: Vec, +} + #[derive(Debug, Serialize, Deserialize)] #[serde(transparent)] pub struct RpcBatch(pub Vec); @@ -287,10 +323,27 @@ pub enum RpcCommand { visibility: Option, }, RoomGrant { + room: String, + username: String, + /// Capability names: `view`, `post`, `vote`, `add_item`, `manage`. + capabilities: Vec, + }, + RoomRevoke { room: String, username: String, capability: String, }, + /// Mint a shareable invite link (24h TTL, stored in memory only until redeemed or expiry). + RoomMintInvite { + room: String, + capabilities: Vec, + #[serde(default = "default_invite_max_uses")] + max_uses: usize, + }, + /// List principals granted access in a room (requires View or Manage). + RoomAudit { + room: String, + }, GetGlobalRank { room: String, #[serde(default)] @@ -361,6 +414,13 @@ pub enum RpcResult { RoomCreated { room_id: String, }, + RoomInviteMinted { + invite_url: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + expires_at_ms: Option, + max_uses: usize, + }, + RoomAudit(RoomAuditResponse), GrantOk {}, GlobalRank(GlobalRankResponse), Pair(PairResponse), 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)))))))