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: [06b48801] Implement extensible URL foundation (#145) * Implement extensible URL foundation Co-authored-by: tommy * Make query params non-identity by default Co-authored-by: tommy * Fix external href helper test scope Co-authored-by: tommy * Avoid broken external previews for blocked hosts Co-authored-by: tommy * Clarify external empty state copy Co-authored-by: tommy * Add on-demand GitHub external resolver Co-authored-by: tommy * Add fenced JSON bodies for GitHub resolver Co-authored-by: tommy --------- Co-authored-by: Cursor Agent Side A — unified diff (full patch): diff --git a/agents.md b/agents.md index 59bd2174f4d2f972a45123fe10d408aa5881ee93..c66f33789441eea193b4354fce4c03b7fffdd639 100644 --- a/agents.md +++ b/agents.md @@ -55,6 +55,7 @@ Strict **CSP** that blocks `eval` would break the current app. Other projects ma | Ingests, grants, rooms, identity tokens, agent binds, redactions, etc. | **JSONL** | Appended in `server/src/api/rpc.rs`, `server/src/api/auth.rs` (and related paths) before updating `ReducerState` | | **`RoomMintInvite` links** | **RAM only** | `AppState.invites` — not appended as `InviteMinted` today; **lost on restart** (`server/src/state.rs`, `server/src/api/rpc.rs`). Event types `InviteMinted` / `InviteRedeemed` exist for replay and a possible future persisted mint (`server/src/reducer.rs`). | | **OAuth / pending sessions** | **RAM only** | `AppState.pending_sessions` (`server/src/state.rs`, `server/src/api/auth.rs`) | +| **External resolver cooldowns** | **RAM only** | `AppState.resolver_runs` — debounce/rate-limit guard for on-demand resolver buttons. Resolver results themselves are durable synthetic `Ingest` events in `events.jsonl`. | | **Reducer projection** | **Derived** | Rebuilt from log on startup; not separately persisted | If you add a new ephemeral map or start persisting something that was RAM-only, **update this table and the code comments** (`server/src/state.rs` is a good anchor). diff --git a/server/src/api/ui_html.rs b/server/src/api/ui_html.rs index 7cbda3876451687aa7a55547fc06bbe86ac9d260..cd501ba6d4d656eabad03afed4583efdf885695d 100644 --- a/server/src/api/ui_html.rs +++ b/server/src/api/ui_html.rs @@ -18,14 +18,15 @@ use crate::{ rpc::{rpc_post_redact, rpc_post_with_bearer, rpc_room_delete}, }, canonical_path::canonicalize_tag, + external_resolver::resolve_github_children, + html::vote_compare_post_success_js, html::{ - fragment_new_thread_slot, login_to_post_hint_markup, - parse_html_ui_from_form, room_members_section_markup, thread_feed_html, - thread_feed_html_for_room, thread_feed_region_markup, thread_ui_collapse_redacted_post, - thread_ui_expand_post_full, thread_ui_expand_redacted_post, ui_js_warn, user_can_post_room, - user_can_view_room, HtmlUiAction, JsBuilder, ThreadNav, + fragment_new_thread_slot, login_to_post_hint_markup, parse_html_ui_from_form, + room_members_section_markup, thread_feed_html, thread_feed_html_for_room, + thread_feed_region_markup, thread_ui_collapse_redacted_post, thread_ui_expand_post_full, + thread_ui_expand_redacted_post, ui_js_warn, user_can_post_room, user_can_view_room, + HtmlUiAction, JsBuilder, ThreadNav, }, - html::vote_compare_post_success_js, reducer::{scope_from_room_wire, ScopeId}, state::AppState, }; @@ -104,26 +105,35 @@ async fn dispatch_ui_action( ) .into_response(); } - match rpc_post_with_bearer(state, &session.bearer, room.clone(), thread_tag.clone(), text).await { - Ok(RpcResult::PostOk { .. }) => { - post_success_response( - state, - &room, - &thread_tag, - error_target.as_ref(), - form_id.as_ref(), - Some(session.username.as_str()), - ) - .await - .into_response() - } + match rpc_post_with_bearer( + state, + &session.bearer, + room.clone(), + thread_tag.clone(), + text, + ) + .await + { + Ok(RpcResult::PostOk { .. }) => post_success_response( + state, + &room, + &thread_tag, + error_target.as_ref(), + form_id.as_ref(), + Some(session.username.as_str()), + ) + .await + .into_response(), Ok(_) => form_js_error( error_target.as_ref(), "unexpected response", "Post did not return PostOk.", ) .into_response(), - Err((msg, hint)) => form_js_error(error_target.as_ref(), &msg, hint.as_deref().unwrap_or("")).into_response(), + Err((msg, hint)) => { + form_js_error(error_target.as_ref(), &msg, hint.as_deref().unwrap_or("")) + .into_response() + } } } HtmlUiAction::CheckIngest { @@ -150,9 +160,19 @@ async fn dispatch_ui_action( return js_clear_errors(&form_error_target(error_target.as_ref())).into_response(); } match rpc_check_with_bearer(state, &session.bearer, room, text.clone()).await { - Ok(RpcResult::CheckOk { .. }) => js_clear_errors(&form_error_target(error_target.as_ref())).into_response(), - Ok(_) => form_js_error(error_target.as_ref(), "unexpected response", "Check did not return CheckOk.").into_response(), - Err((msg, hint)) => form_js_error(error_target.as_ref(), &msg, hint.as_deref().unwrap_or("")).into_response(), + Ok(RpcResult::CheckOk { .. }) => { + js_clear_errors(&form_error_target(error_target.as_ref())).into_response() + } + Ok(_) => form_js_error( + error_target.as_ref(), + "unexpected response", + "Check did not return CheckOk.", + ) + .into_response(), + Err((msg, hint)) => { + form_js_error(error_target.as_ref(), &msg, hint.as_deref().unwrap_or("")) + .into_response() + } } } HtmlUiAction::VoteComparePost { @@ -167,7 +187,11 @@ async fn dispatch_ui_action( form_action, } => { if form_action != "/ui" { - return (StatusCode::BAD_REQUEST, "invalid vote_compare_post form_action").into_response(); + return ( + StatusCode::BAD_REQUEST, + "invalid vote_compare_post form_action", + ) + .into_response(); } let Some(session) = session else { return js_redirect("/login").into_response(); @@ -195,23 +219,15 @@ async fn dispatch_ui_action( let left_id = match crate::path_types::ItemId::parse(left_item.trim()) { Some(i) => i.normalized_storage(), None => { - return form_js_error( - err_tgt.as_ref(), - "bad item", - "Invalid left item path.", - ) - .into_response(); + return form_js_error(err_tgt.as_ref(), "bad item", "Invalid left item path.") + .into_response(); } }; let right_id = match crate::path_types::ItemId::parse(right_item.trim()) { Some(i) => i.normalized_storage(), None => { - return form_js_error( - err_tgt.as_ref(), - "bad item", - "Invalid right item path.", - ) - .into_response(); + return form_js_error(err_tgt.as_ref(), "bad item", "Invalid right item path.") + .into_response(); } }; let mut rl = ratio_left.trim().parse::().unwrap_or(0).max(0); @@ -231,7 +247,15 @@ async fn dispatch_ui_action( right_id.as_str() ); - match rpc_post_with_bearer(state, &session.bearer, room.clone(), thread_tag.clone(), text).await { + match rpc_post_with_bearer( + state, + &session.bearer, + room.clone(), + thread_tag.clone(), + text, + ) + .await + { Ok(RpcResult::PostOk { post_id, post_index, @@ -277,7 +301,10 @@ async fn dispatch_ui_action( "Post did not return PostOk.", ) .into_response(), - Err((msg, hint)) => form_js_error(err_tgt.as_ref(), &msg, hint.as_deref().unwrap_or("")).into_response(), + Err((msg, hint)) => { + form_js_error(err_tgt.as_ref(), &msg, hint.as_deref().unwrap_or("")) + .into_response() + } } } HtmlUiAction::SetGardenPin { @@ -288,7 +315,11 @@ async fn dispatch_ui_action( form_action, } => { if form_action != "/ui" { - return (StatusCode::BAD_REQUEST, "invalid set_garden_pin form_action").into_response(); + return ( + StatusCode::BAD_REQUEST, + "invalid set_garden_pin form_action", + ) + .into_response(); } let next_path = sanitize_garden_pin_next(&next); use crate::html::{encode_pin_cookie_value, GARDEN_PIN_COOKIE}; @@ -301,7 +332,11 @@ async fn dispatch_ui_action( if room.is_empty() { return (StatusCode::BAD_REQUEST, "missing room").into_response(); } - let Some(raw) = item_storage.as_ref().map(|s| s.trim().to_string()).filter(|s| !s.is_empty()) else { + let Some(raw) = item_storage + .as_ref() + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + else { return (StatusCode::BAD_REQUEST, "missing item").into_response(); }; let Some(item) = ItemId::parse(&raw) else { @@ -309,16 +344,65 @@ async fn dispatch_ui_action( }; let item = item.normalized_storage(); let val = encode_pin_cookie_value(&room, item.as_str()); - let cookie = format!("{GARDEN_PIN_COOKIE}={val}; Path=/; SameSite=Lax; Max-Age=7776000"); + let cookie = + format!("{GARDEN_PIN_COOKIE}={val}; Path=/; SameSite=Lax; Max-Age=7776000"); redirect_with_pin_cookie(&cookie, &next_path) } + HtmlUiAction::ResolveExternal { + room_wire, + item_storage, + mode, + next, + form_action, + } => { + if form_action != "/ui" { + return ( + StatusCode::BAD_REQUEST, + "invalid resolve_external form_action", + ) + .into_response(); + } + let Some(session) = session else { + return js_redirect("/login").into_response(); + }; + let room = room_wire.trim(); + if room.is_empty() { + return ui_js_warn("missing room").into_response(); + } + let reduced = state.reduced.read().await; + if matches!(scope_from_room_wire(room), ScopeId::Room(_)) { + if !user_can_post_room(&reduced, room, &session.username) { + drop(reduced); + return ui_js_warn("forbidden").into_response(); + } + } + drop(reduced); + + let Some(item) = crate::path_types::ItemId::parse(item_storage.trim()) else { + return ui_js_warn("bad item").into_response(); + }; + let target = if mode.trim() == "siblings" { + match item.parent() { + Some(parent) => parent.normalized_storage(), + None => return ui_js_warn("no parent to resolve siblings").into_response(), + } + } else { + item.normalized_storage() + }; + match resolve_github_children(state, room, &target).await { + Ok(_) => js_redirect(&sanitize_garden_pin_next(&next)).into_response(), + Err(msg) => ui_js_warn(&msg).into_response(), + } + } HtmlUiAction::RedactPost { post_id } => { let Some(session) = session else { return js_redirect("/login").into_response(); }; let h = headers_from_bearer(&session.bearer); match rpc_post_redact(state, &h, post_id).await { - Ok(RpcResult::RedactPostOk {}) => redact_success_response(state).await.into_response(), + Ok(RpcResult::RedactPostOk {}) => { + redact_success_response(state).await.into_response() + } Ok(_) => (StatusCode::BAD_REQUEST, "unexpected response").into_response(), Err((msg, hint)) => { let detail = hint.as_deref().unwrap_or(""); @@ -326,7 +410,10 @@ async fn dispatch_ui_action( } } } - HtmlUiAction::SetRoomMembersExpanded { room_wire, expanded } => { + HtmlUiAction::SetRoomMembersExpanded { + room_wire, + expanded, + } => { let room_wire = room_wire.trim().to_string(); if room_wire.is_empty() { return ui_js_warn("missing room").into_response(); @@ -365,7 +452,10 @@ async fn dispatch_ui_action( } } } - HtmlUiAction::SetNewThreadComposeExpanded { room_wire, expanded } => { + HtmlUiAction::SetNewThreadComposeExpanded { + room_wire, + expanded, + } => { let room_wire = room_wire.trim().to_string(); if room_wire.is_empty() { return ui_js_warn("missing room").into_response(); @@ -614,7 +704,9 @@ async fn post_success_response( }; builder }) - .if_current_path_not_matches(&thread_location, |builder| builder.redirect(&thread_location)); + .if_current_path_not_matches(&thread_location, |builder| { + builder.redirect(&thread_location) + }); builder.into_response() } @@ -625,4 +717,3 @@ async fn redact_success_response(state: &AppState) -> Response { .morph_selector("#thread-feed", feed_markup) .into_response() } - diff --git a/server/src/api/write_actor.rs b/server/src/api/write_actor.rs index 700c3174b0e980d46ebf4aa94e8ffe2f8dedd723..3168956e7dce1cadf50c275904b550bf81500687 100644 --- a/server/src/api/write_actor.rs +++ b/server/src/api/write_actor.rs @@ -54,7 +54,11 @@ fn content_for_room<'a>(reduced: &'a ReducerState, room: &str) -> &'a crate::red } async fn broadcast_web_refresh(state: &AppState, room_key: &str, thread_id: &str) { - let feed_id = if room_key == "public" { "thread-feed" } else { "room-thread-feed" }; + let feed_id = if room_key == "public" { + "thread-feed" + } else { + "room-thread-feed" + }; let thread_url = if room_key == "public" { format!("/t/{thread_id}") } else if let Some(seg) = room_route_segment(room_key) { @@ -86,7 +90,10 @@ async fn broadcast_web_refresh(state: &AppState, room_key: &str, thread_id: &str "/".to_string() }]; path_prefixes.push(thread_url.clone()); - let _ = state.js_tx.send(crate::state::JsSnippet { code: js, path_prefixes }); + let _ = state.js_tx.send(crate::state::JsSnippet { + code: js, + path_prefixes, + }); } fn compute_scope_rank_changes( @@ -104,7 +111,13 @@ fn compute_scope_rank_changes( for comp in &rankings.component_rankings { let total = comp.ranked.len(); for (i, item) in comp.ranked.iter().enumerate() { - map.insert(item.item.clone(), Some(RankPosition { rank: i + 1, of: total })); + map.insert( + item.item.clone(), + Some(RankPosition { + rank: i + 1, + of: total, + }), + ); } } for item in &rankings.unranked_items { @@ -158,7 +171,11 @@ fn compute_scope_rank_changes( }) } -async fn redeem_invite_grant(state: &AppState, invite_token: &str, grantee_username: &str) -> Result<(), String> { +async fn redeem_invite_grant( + state: &AppState, + invite_token: &str, + grantee_username: &str, +) -> Result<(), String> { let now = now_ms(); let ga = { let mut invites = state.invites.write().await; @@ -220,14 +237,14 @@ pub async fn writer_actor(mut rx: mpsc::Receiver, state: AppState) { let out = async { let mut reduced = state.reduced.write().await; - let principal = - verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; + let principal = verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; let delegate: Option = match delegate_opt { None => None, Some(ref s) if s.trim().is_empty() => None, Some(s) => Some( - parse_agent(&s).map_err(|msg| (format!("invalid delegate format"), Some(msg)))?, + parse_agent(&s) + .map_err(|msg| (format!("invalid delegate format"), Some(msg)))?, ), }; @@ -241,11 +258,12 @@ pub async fn writer_actor(mut rx: mpsc::Receiver, state: AppState) { )); } - let v = validate_ingest_document(&reduced, &text, &scope) - .map_err(|(st, m, h)| { + let v = validate_ingest_document(&reduced, &text, &scope).map_err( + |(st, m, h)| { let _ = st; (m, h) - })?; + }, + )?; if is_private { use crate::events::ThreadCapability; @@ -276,7 +294,10 @@ pub async fn writer_actor(mut rx: mpsc::Receiver, state: AppState) { if let Some(ref d) = delegate { match reduced.agent_bindings.get(d) { Some(u) if u != &principal => { - return Err(("delegate already bound to another user".into(), None)); + return Err(( + "delegate already bound to another user".into(), + None, + )); } _ => {} } @@ -427,6 +448,80 @@ pub async fn writer_actor(mut rx: mpsc::Receiver, state: AppState) { let _ = reply.send(out); } + WriteCmd::SystemIngest { + room, + thread_tag, + text, + principal, + reply, + } => { + let out = async { + let (room_key, thread_id) = normalize_room_and_thread(&room, &thread_tag); + let scope = scope_from_room_wire(&room_key); + let is_private = !matches!(scope, ScopeId::Public); + let mut reduced = state.reduced.write().await; + if is_private && !reduced.rooms.contains(&room_key) { + return Err(( + "unknown room".into(), + Some(format!("room `{}` does not exist", room_key)), + )); + } + + let v = validate_ingest_document(&reduced, &text, &scope).map_err( + |(st, m, h)| { + let _ = st; + (m, h) + }, + )?; + + let new_post_id = uuid::Uuid::new_v4().to_string(); + let ingest_event = Event::Ingest(Ingest { + ts: v.ts, + id: new_post_id.clone(), + raw: v.raw_text.clone(), + principal: principal.clone(), + delegate: None, + room_id: room_key.clone(), + thread_tag: thread_id.clone(), + }); + + state + .event_log + .append(&ingest_event) + .await + .map_err(|e| (format!("{e}"), None))?; + reduced.apply_event(ingest_event); + drop(reduced); + + use crate::canonical_path::canonicalize_tag; + let thread_tag_canon = canonicalize_tag(&thread_id); + broadcast_web_refresh(&state, &room_key, &thread_tag_canon).await; + let post_index = { + let reduced = state.reduced.read().await; + reduced.try_thread_post_index_chronological( + &scope_from_room_wire(&room_key), + &thread_tag_canon, + &new_post_id, + ) + }; + + Ok(RpcResult::PostOk { + events_appended: 1, + post_id: Some(new_post_id), + post_index, + ranking_changes: None, + threads: vec![format!("#{}", thread_id)], + next: slug_types::NextMoves { + pair: "npx slugsocial public garden pair".to_string(), + rank: "npx slugsocial public garden rank".to_string(), + web: ForumThreadUrl::from_room_tag(&room_key, &thread_id), + }, + }) + } + .await; + let _ = reply.send(out); + } + WriteCmd::Redact { post_id, bearer, @@ -434,8 +529,7 @@ pub async fn writer_actor(mut rx: mpsc::Receiver, state: AppState) { } => { let out = async { let mut reduced = state.reduced.write().await; - let principal = - verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; + let principal = verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; let post_id = post_id.trim().to_string(); let Some(ing) = reduced.ingests_by_id.get(&post_id).cloned() else { return Err(("post not found".into(), None)); @@ -450,7 +544,11 @@ pub async fn writer_actor(mut rx: mpsc::Receiver, state: AppState) { let scope = scope_from_room_wire(&room_key); if matches!(scope, ScopeId::Room(_)) && (!reduced.rooms.contains(&room_key) - || !reduced.user_has_cap(&room_key, &principal, crate::events::ThreadCapability::View)) + || !reduced.user_has_cap( + &room_key, + &principal, + crate::events::ThreadCapability::View, + )) { return Err(("room not found".into(), None)); } @@ -484,14 +582,16 @@ pub async fn writer_actor(mut rx: mpsc::Receiver, state: AppState) { } => { let out = async { let mut reduced = state.reduced.write().await; - let principal = - verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; + let principal = verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; let slug = slug.trim().to_lowercase(); if slug.is_empty() || slug.len() > 64 { return Err(("slug must be 1-64 characters".into(), None)); } if !slug.chars().all(|c| c.is_ascii_alphanumeric() || c == '-') { - return Err(("slug must be lowercase alphanumeric with hyphens".into(), None)); + return Err(( + "slug must be lowercase alphanumeric with hyphens".into(), + None, + )); } let short_id = loop { let id = gen_short_id(); @@ -539,13 +639,16 @@ pub async fn writer_actor(mut rx: mpsc::Receiver, state: AppState) { let _ = reply.send(out); } - WriteCmd::RoomDelete { room, bearer, reply } => { + WriteCmd::RoomDelete { + room, + bearer, + reply, + } => { let out = async { use crate::events::ThreadCapability; let mut reduced = state.reduced.write().await; - let principal = - verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; + let principal = verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; let room = room.trim().to_string(); if !reduced.rooms.contains(&room) { return Err(("unknown room".into(), None)); @@ -589,15 +692,15 @@ pub async fn writer_actor(mut rx: mpsc::Receiver, state: AppState) { use crate::identity::parse_username; let mut reduced = state.reduced.write().await; - let principal = - verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; + let principal = verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; if !reduced.user_has_cap(&room, &principal, ThreadCapability::Manage) { return Err(("requires Manage capability".into(), None)); } if capabilities.is_empty() { return Err(("capabilities must not be empty".into(), None)); } - let target = parse_username(&username).map_err(|msg| ("invalid username".into(), Some(msg)))?; + let target = parse_username(&username) + .map_err(|msg| ("invalid username".into(), Some(msg)))?; if !reduced.users_by_provider.values().any(|u| u == &target) { return Err((format!("user @{target} not found"), None)); } @@ -637,12 +740,12 @@ pub async fn writer_actor(mut rx: mpsc::Receiver, state: AppState) { use crate::identity::parse_username; let mut reduced = state.reduced.write().await; - let principal = - verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; + let principal = verify_token(&reduced, &bearer).map_err(|(_, m)| (m, None))?; if !reduced.user_has_cap(&room, &principal, ThreadCapability::Manage) { return Err(("requires Manage capability".into(), None)); } - let target = parse_username(&username).map_err(|msg| ("invalid username".into(), Some(msg)))?; + let target = parse_username(&username) + .map_err(|msg| ("invalid username".into(), Some(msg)))?; if !reduced.users_by_provider.values().any(|u| u == &target) { return Err((format!("user @{target} not found"), None)); } diff --git a/server/src/dsl.rs b/server/src/dsl.rs index def8b497c71567016a522648b9630d7038163e41..4203c2f59dd8825d7a91513c4023f3ccd37f1efc 100644 --- a/server/src/dsl.rs +++ b/server/src/dsl.rs @@ -450,8 +450,8 @@ fn parse_block_prefixed_statement( } fn parse_item_definition_statement(stripped: &str, masker: &BlockMasker) -> Result { - let (item1, j) = - parse_item_name_at(stripped, 0).ok_or_else(|| DslError::Parse("invalid item name".to_string()))?; + let (item1, j) = parse_item_name_at(stripped, 0) + .ok_or_else(|| DslError::Parse("invalid item name".to_string()))?; let i = skip_ws(stripped, j); if i >= stripped.len() { @@ -632,6 +632,19 @@ mod tests { ); } + #[test] + fn parse_item_with_fenced_json_body_preserves_braces() { + let input = "~/item/in/url ```json\n{\"test\": true}\n```"; + let doc = parse_full(input).unwrap(); + assert_eq!( + doc.statements, + vec![Stmt::Item { + title: "~/item/in/url".to_string(), + body: Some("```json\n{\"test\": true}\n```".to_string()), + }] + ); + } + #[test] fn parse_vote_ratio_and_symbols() { let d1 = parse_full("{because}\n~/a 3:1 ~/b").unwrap(); diff --git a/server/src/external_resolver.rs b/server/src/external_resolver.rs index 13f06b922a685f110d269f7397da807cf8600fe1..8481f1fbff1c4265da182a225f539f97d6829002 100644 --- a/server/src/external_resolver.rs +++ b/server/src/external_resolver.rs @@ -1,6 +1,26 @@ use async_trait::async_trait; +use serde_json::Value; +use tokio::sync::oneshot; -use crate::path_types::ItemId; +use crate::{path_types::ItemId, state::AppState, write_cmd::WriteCmd}; + +const GITHUB_SYSTEM_PRINCIPAL: &str = "system:github-resolver"; +const GITHUB_RESOLVER_COOLDOWN_MS: i64 = 15_000; + +fn now_ms() -> i64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ResolvedChild { + pub url: String, + pub title: String, + pub body: Option, +} #[async_trait] pub trait ExternalResolver: Send + Sync { @@ -14,7 +34,290 @@ pub trait ExternalResolver: Send + Sync { async fn fetch_body(&self, item: &ItemId) -> Result; } -/// Placeholder until domain-specific resolvers exist. +#[derive(Clone)] +pub struct GitHubResolver { + client: reqwest::Client, + api_base_url: String, + token: Option, +} + +impl GitHubResolver { + pub fn from_env() -> Self { + let api_base_url = std::env::var("SLUG_GITHUB_API_BASE_URL") + .ok() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| "https://api.github.com".to_string()); + let token = std::env::var("SLUG_GITHUB_TOKEN") + .ok() + .filter(|s| !s.trim().is_empty()); + Self { + client: reqwest::Client::new(), + api_base_url: api_base_url.trim_end_matches('/').to_string(), + token, + } + } + + pub fn can_resolve_children(&self, item: &ItemId) -> bool { + github_segments(item).is_some() + } + + pub async fn list_children(&self, item: &ItemId) -> Result, String> { + let segments = github_segments(item).ok_or_else(|| "not a GitHub URL".to_string())?; + match segments.as_slice() { + [] => Ok(vec![]), + [owner] => self.list_repos(owner).await, + [owner, repo] => Ok(github_repo_sections(owner, repo)), + [owner, repo, section] if section == "issues" => self.list_issues(owner, repo).await, + [owner, repo, section] if section == "pulls" => self.list_pulls(owner, repo).await, + _ => Ok(vec![]), + } + } + + async fn get_json(&self, path: &str) -> Result { + let url = format!("{}/{}", self.api_base_url, path.trim_start_matches('/')); + let mut req = self + .client + .get(url) + .header(reqwest::header::USER_AGENT, "slugsocial-github-resolver"); + if let Some(token) = &self.token { + req = req.bearer_auth(token); + } + let resp = req + .send() + .await + .map_err(|e| format!("GitHub request failed: {e}"))?; + let status = resp.status(); + if !status.is_success() { + return Err(format!("GitHub request returned {status}")); + } + resp.json::() + .await + .map_err(|e| format!("GitHub response JSON failed: {e}")) + } + + async fn list_repos(&self, owner: &str) -> Result, String> { + let value = self + .get_json(&format!( + "/users/{owner}/repos?per_page=100&sort=updated&type=owner" + )) + .await?; + let arr = value + .as_array() + .ok_or_else(|| "GitHub repos response was not an array".to_string())?; + let mut out = Vec::new(); + for repo in arr { + let name = repo + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + if name.is_empty() { + continue; + } + let full_name = repo + .get("full_name") + .and_then(|v| v.as_str()) + .map(|s| s.to_ascii_lowercase()) + .unwrap_or_else(|| format!("{owner}/{name}").to_ascii_lowercase()); + out.push(ResolvedChild { + url: format!("https://github.com/{full_name}"), + title: full_name.clone(), + body: Some(github_json_body(repo)), + }); + } + out.sort_by(|a, b| a.url.cmp(&b.url)); + Ok(out) + } + + async fn list_issues(&self, owner: &str, repo: &str) -> Result, String> { + let value = self + .get_json(&format!( + "/repos/{owner}/{repo}/issues?state=open&per_page=100" + )) + .await?; + let arr = value + .as_array() + .ok_or_else(|| "GitHub issues response was not an array".to_string())?; + let mut out = Vec::new(); + for issue in arr { + if issue.get("pull_request").is_some() { + continue; + } + let Some(number) = issue.get("number").and_then(|v| v.as_i64()) else { + continue; + }; + let title = issue + .get("title") + .and_then(|v| v.as_str()) + .unwrap_or("Untitled issue"); + out.push(ResolvedChild { + url: format!("https://github.com/{owner}/{repo}/issues/{number}"), + title: format!("#{number} {title}"), + body: Some(github_json_body(issue)), + }); + } + out.sort_by(|a, b| a.url.cmp(&b.url)); + Ok(out) + } + + async fn list_pulls(&self, owner: &str, repo: &str) -> Result, String> { + let value = self + .get_json(&format!( + "/repos/{owner}/{repo}/pulls?state=open&per_page=100" + )) + .await?; + let arr = value + .as_array() + .ok_or_else(|| "GitHub pulls response was not an array".to_string())?; + let mut out = Vec::new(); + for pull in arr { + let Some(number) = pull.get("number").and_then(|v| v.as_i64()) else { + continue; + }; + let title = pull + .get("title") + .and_then(|v| v.as_str()) + .unwrap_or("Untitled pull request"); + out.push(ResolvedChild { + url: format!("https://github.com/{owner}/{repo}/pulls/{number}"), + title: format!("#{number} {title}"), + body: Some(github_json_body(pull)), + }); + } + out.sort_by(|a, b| a.url.cmp(&b.url)); + Ok(out) + } +} + +fn github_segments(item: &ItemId) -> Option> { + let url = url::Url::parse(item.as_str()).ok()?; + if url.host_str()?.eq_ignore_ascii_case("github.com") { + Some( + url.path_segments() + .map(|segments| { + segments + .filter(|s| !s.is_empty()) + .map(|s| s.to_ascii_lowercase()) + .collect::>() + }) + .unwrap_or_default(), + ) + } else { + None + } +} + +fn github_repo_sections(owner: &str, repo: &str) -> Vec { + [ + ("issues", "GitHub issues for this repository."), + ("pulls", "GitHub pull requests for this repository."), + ("commits", "GitHub commits for this repository."), + ("releases", "GitHub releases for this repository."), + ] + .into_iter() + .map(|(section, body)| ResolvedChild { + url: format!("https://github.com/{owner}/{repo}/{section}"), + title: section.to_string(), + body: Some(body.to_string()), + }) + .collect() +} + +fn resolver_thread_tag(item: &ItemId) -> String { + let tail = item + .display_path() + .trim_start_matches("-/") + .replace('/', ":") + .replace('?', ":"); + format!("import:{tail}") +} + +fn sanitize_body(s: &str) -> String { + s.replace('{', "(") + .replace('}', ")") + .replace("```", "` ` `") + .chars() + .take(4_000) + .collect() +} + +fn github_json_body(value: &Value) -> String { + let json = serde_json::to_string_pretty(value) + .unwrap_or_else(|_| value.to_string()) + .replace("```", "` ` `"); + format!("```json\n{json}\n```") +} + +fn children_to_dsl(children: &[ResolvedChild]) -> String { + let mut out = String::new(); + for child in children { + let body = child + .body + .as_deref() + .filter(|s| !s.trim().is_empty()) + .unwrap_or(child.title.as_str()); + if body.trim_start().starts_with("```") { + out.push_str(&format!("{} {}\n\n", child.url, body.trim())); + } else { + out.push_str(&format!( + "{} {{\n{}\n}}\n\n", + child.url, + sanitize_body(body) + )); + } + } + out +} + +pub async fn resolve_github_children( + state: &AppState, + room: &str, + item: &ItemId, +) -> Result { + if !state.github_resolver.can_resolve_children(item) { + return Err("no GitHub resolver for this item".to_string()); + } + + let key = format!("github:{}:{}", room.trim(), item.as_str()); + let now = now_ms(); + { + let mut runs = state.resolver_runs.write().await; + if let Some(last) = runs.get(&key) { + let remaining = GITHUB_RESOLVER_COOLDOWN_MS - (now - *last); + if remaining > 0 { + return Err(format!( + "GitHub resolver cooldown: try again in {}s", + (remaining + 999) / 1000 + )); + } + } + runs.insert(key, now); + } + + let children = state.github_resolver.list_children(item).await?; + if children.is_empty() { + return Ok(0); + } + let text = children_to_dsl(&children); + let thread_tag = resolver_thread_tag(item); + let (tx, rx) = oneshot::channel(); + state + .write_tx + .send(WriteCmd::SystemIngest { + room: room.to_string(), + thread_tag, + text, + principal: GITHUB_SYSTEM_PRINCIPAL.to_string(), + reply: tx, + }) + .await + .map_err(|_| "writer unavailable".to_string())?; + rx.await + .map_err(|_| "writer dropped".to_string())? + .map_err(|(msg, hint)| hint.map_or(msg.clone(), |h| format!("{msg}: {h}")))?; + Ok(children.len()) +} + +/// Placeholder until other domain-specific resolvers exist. pub struct DefaultExternalResolver; #[async_trait] @@ -31,3 +334,51 @@ impl ExternalResolver for DefaultExternalResolver { Err("external fetch not implemented".to_string()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn github_segments_parse_normalized_url() { + let item = ItemId::parse("https://github.com/Sortersocial/Slug/issues").unwrap(); + assert_eq!( + github_segments(&item), + Some(vec![ + "sortersocial".to_string(), + "slug".to_string(), + "issues".to_string() + ]) + ); + } + + #[test] + fn repo_sections_are_direct_children() { + let sections = github_repo_sections("sortersocial", "slug"); + let urls: Vec = sections.into_iter().map(|c| c.url).collect(); + assert!(urls.contains(&"https://github.com/sortersocial/slug/issues".to_string())); + assert!(urls.contains(&"https://github.com/sortersocial/slug/pulls".to_string())); + } + + #[test] + fn children_to_dsl_contains_item_bodies() { + let dsl = children_to_dsl(&[ResolvedChild { + url: "https://github.com/o/r/issues/1".into(), + title: "#1 title".into(), + body: Some("body with {braces}".into()), + }]); + assert!(dsl.contains("https://github.com/o/r/issues/1")); + assert!(dsl.contains("body with (braces)")); + } + + #[test] + fn children_to_dsl_preserves_fenced_json_bodies() { + let dsl = children_to_dsl(&[ResolvedChild { + url: "https://github.com/o/r/issues/1".into(), + title: "#1 title".into(), + body: Some("```json\n{\"test\": true}\n```".into()), + }]); + assert!(dsl.contains("https://github.com/o/r/issues/1 ```json")); + assert!(dsl.contains("{\"test\": true}")); + } +} diff --git a/server/src/html/garden.rs b/server/src/html/garden.rs index 4c03640334a524babe39c02289989739b1ad0ae9..b28271e0316738eae245855b24aeb76486860554 100644 --- a/server/src/html/garden.rs +++ b/server/src/html/garden.rs @@ -9,18 +9,21 @@ use serde::Deserialize; use serde_json::json; use std::collections::HashSet; -use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD as B64_ENGINE}; +use base64::{engine::general_purpose::URL_SAFE_NO_PAD as B64_ENGINE, Engine as _}; use crate::{ api::optional_principal, canonical_path::{canonicalize_item, canonicalize_tag}, - middleware::canonical_view_url, events::ThreadCapability, form_template::template_json_compact, - html::{JsBuilder, ui_action::UI_RPC_FIELD, user_can_post_room}, + html::{ui_action::UI_RPC_FIELD, user_can_post_room, HtmlUiAction, JsBuilder}, + middleware::canonical_view_url, path_types::ItemId, reducer::{ContentState, ReducerState, ScopeId}, - scope_rank::{ChildrenRankings, build_children_rankings}, + scope_rank::{ + build_children_rankings, build_rankings_for_item_set, resolve_scope_recursive, + ChildrenRankings, + }, state::AppState, timeago, }; @@ -128,7 +131,11 @@ fn left_share_normalized(ratio_left: i32, ratio_right: i32) -> f64 { let l = ratio_left.max(0) as f64; let r = ratio_right.max(0) as f64; let sum = l + r; - if sum <= 0.0 { 0.5 } else { l / sum } + if sum <= 0.0 { + 0.5 + } else { + l / sum + } } /// Stronger preference for **`page_left` first**; ties **newer first**. @@ -864,6 +871,7 @@ struct ItemPageViewModel { /// False at the tilde ontology root (`~/`): sibling-rank footnote does not apply. item_has_parent: bool, child_rankings: ChildrenRankings, + child_depth: usize, rank_history: Vec, /// Forum threads that mention or vote on this item. threads: Vec, @@ -1018,13 +1026,20 @@ fn build_item_page_view_model( reduced: &crate::reducer::ReducerState, scope: &ScopeId, item: &str, + child_depth: usize, ) -> ItemPageViewModel { let content = content_for_garden_view(reduced, scope); let item_key = ItemId::parse(item) .unwrap_or_else(|| ItemId::parse("~/").unwrap()) .normalized_storage(); let item_has_parent = item_key.parent().is_some(); - let child_rankings = build_children_rankings(content, &item_key); + let child_depth = child_depth.clamp(1, 5); + let child_rankings = if child_depth > 1 { + let items = resolve_scope_recursive(content, &[item_key.as_str().to_string()], child_depth); + build_rankings_for_item_set(content, &items) + } else { + build_children_rankings(content, &item_key) + }; let sibling_nav = build_sibling_nav(reduced, scope, &item_key); let rank_history = build_rank_history(reduced, scope, item_key.as_str()); @@ -1046,11 +1061,111 @@ fn build_item_page_view_model( sibling_nav, item_has_parent, child_rankings, + child_depth, rank_history, threads, } } +fn child_depth_from_uri(uri: &Uri) -> usize { + uri.query() + .into_iter() + .flat_map(|q| q.split('&')) + .filter_map(|pair| pair.split_once('=')) + .find_map(|(k, v)| { + if k == "depth" { + v.parse::().ok() + } else { + None + } + }) + .unwrap_or(1) + .clamp(1, 5) +} + +fn external_source_href(item: &str) -> String { + let Ok(mut url) = url::Url::parse(item) else { + return item.to_string(); + }; + let is_youtube = url + .host_str() + .map(|h| h.eq_ignore_ascii_case("www.youtube.com")) + .unwrap_or(false); + if is_youtube { + let segments: Vec = url + .path_segments() + .map(|s| s.map(|seg| seg.to_string()).collect()) + .unwrap_or_default(); + if segments.len() == 3 && segments[0] == "watch" && segments[1] == "v" { + let id = segments[2].clone(); + url.set_path("/watch"); + url.set_query(None); + url.query_pairs_mut().append_pair("v", &id); + return url.to_string(); + } + } + item.to_string() +} + +fn external_frame_allowed(item: &str) -> bool { + let Ok(url) = url::Url::parse(item) else { + return false; + }; + let host = url.host_str().unwrap_or_default(); + !matches!(host, "github.com" | "www.github.com") +} + +fn github_resolver_controls(item: &str, nav: &ThreadNav, next: &str) -> Option { + let item_id = ItemId::parse(item)?.normalized_storage(); + let url = url::Url::parse(item_id.as_str()).ok()?; + if !url + .host_str() + .map(|h| h.eq_ignore_ascii_case("github.com")) + .unwrap_or(false) + { + return None; + } + let children_rpc = template_json_compact(&HtmlUiAction::ResolveExternal { + room_wire: nav.room_wire.clone(), + item_storage: item_id.as_str().to_string(), + mode: "children".to_string(), + next: next.to_string(), + form_action: "/ui".to_string(), + }) + .ok()?; + let siblings_rpc = item_id.parent().and_then(|_| { + template_json_compact(&HtmlUiAction::ResolveExternal { + room_wire: nav.room_wire.clone(), + item_storage: item_id.as_str().to_string(), + mode: "siblings".to_string(), + next: next.to_string(), + form_action: "/ui".to_string(), + }) + .ok() + }); + + Some(html! { + section id="external-resolver-panel" class="ont-tab-panel ont-external-resolver" { + h3 { "GitHub resolver" } + p class="muted" { + "Import GitHub neighbors on demand. Results are saved as system ingests." + } + div class="resolver-actions" { + form method="POST" action="/ui" { + input type="hidden" name=(UI_RPC_FIELD) value=(children_rpc); + button type="submit" data-testid="github-resolve-children" { "Load children from GitHub" } + } + @if let Some(rpc) = siblings_rpc { + form method="POST" action="/ui" { + input type="hidden" name=(UI_RPC_FIELD) value=(rpc); + button type="submit" data-testid="github-resolve-siblings" { "Load siblings from GitHub" } + } + } + } + } + }) +} + async fn render_scope_view( state: AppState, browse: GardenBrowsePath, @@ -1061,11 +1176,14 @@ async fn render_scope_view( let scope = nav.scope(); let pin_ref = pinned_item_from_jar(&jar); let reduced = state.reduced.read().await; - let model = build_item_page_view_model(&reduced, &scope, browse.item()); + let child_depth = child_depth_from_uri(&uri); + let model = build_item_page_view_model(&reduced, &scope, browse.item(), child_depth); let scope_content = content_for_garden_view(&reduced, &scope); let thread_href = |tag: &str| nav.thread_url(tag); let external_empty_body = browse.is_external() && model.body.is_none(); let cli_path_arg = item_display_path(&model.item); + let external_href = external_source_href(&model.item); + let external_can_embed = external_frame_allowed(&model.item); let (garden_room, garden_prefix) = garden_layout_meta(&nav); let next_for_pin = uri .path_and_query() @@ -1104,7 +1222,24 @@ async fn render_scope_view( div class="ont-item-content ont-external-empty" { p { "This is an external scope." } p class="muted" { - button type="button" disabled { "Kick off an Agent Run to import and rank items" } + "This scope has no imported body yet. Open the source page directly." + } + p { + a href=(external_href.as_str()) target="_blank" rel="noopener noreferrer" { + "Open external URL" + } + } + @if external_can_embed { + p class="muted" { "Best-effort embedded preview:" } + iframe + class="ont-external-frame" + src=(external_href.as_str()) + sandbox="" + loading="lazy" + referrerpolicy="no-referrer" + style="width:100%;height:360px;border:1px solid currentColor;" {} + } @else { + p class="muted" { "Embedded preview is unavailable for this host." } } } } @else { @@ -1112,6 +1247,10 @@ async fn render_scope_view( } } + @if let Some(markup) = github_resolver_controls(&model.item, &nav, &next_for_pin) { + (markup) + } + @if !model.rank_history.is_empty() { details class="ont-rank-history" { summary { @@ -1182,7 +1321,13 @@ async fn render_scope_view( } section class="ont-tab-panel ont-tab-panel-children" { - h3 { "ranked child groups" } + h3 { + "ranked child groups" + @if model.child_depth > 1 { + " " + span class="muted" { (format!("(depth {})", model.child_depth)) } + } + } @if model.child_rankings.component_rankings.is_empty() { p class="muted" { "no voted pairs yet in this scope" } } @else { @@ -1415,7 +1560,7 @@ async fn vote_compare_inner( #[cfg(test)] mod tests { - use super::build_item_page_view_model; + use super::{build_item_page_view_model, external_source_href}; use crate::{ events::{Event, Ingest}, reducer::{ReducerState, ScopeId}, @@ -1516,7 +1661,7 @@ mod tests { ~/topic/b {beta}\n", ); - let model = build_item_page_view_model(&reduced, &ScopeId::Public, "~/topic/a"); + let model = build_item_page_view_model(&reduced, &ScopeId::Public, "~/topic/a", 1); assert_eq!(model.body.as_deref(), Some("alpha")); assert!(model.sibling_nav.is_some()); assert!(model.child_rankings.component_rankings.is_empty()); @@ -1536,7 +1681,7 @@ mod tests { {a beats b}\n ~/topic/a 2:1 ~/topic/b\n", ); - let model = build_item_page_view_model(&reduced, &ScopeId::Public, "~/topic/a"); + let model = build_item_page_view_model(&reduced, &ScopeId::Public, "~/topic/a", 1); let nav = model.sibling_nav.expect("expected sibling nav"); assert_eq!(nav.groups.len(), 2); assert_eq!(nav.groups[0].links.len(), 2); @@ -1558,7 +1703,7 @@ mod tests { {a beats b}\n ~/topic/a 2:1 ~/topic/b\n", ); - let model = build_item_page_view_model(&reduced, &ScopeId::Public, "~/topic/a"); + let model = build_item_page_view_model(&reduced, &ScopeId::Public, "~/topic/a", 1); let nav = model.sibling_nav.expect("expected sibling nav"); assert_eq!(nav.groups.len(), 3); assert_eq!(nav.groups[0].links.len(), 2); @@ -1582,7 +1727,7 @@ mod tests { ~/topic/kid1/leaf {leaf}\n", ); - let model = build_item_page_view_model(&reduced, &ScopeId::Public, "~/topic"); + let model = build_item_page_view_model(&reduced, &ScopeId::Public, "~/topic", 1); assert_eq!(model.child_rankings.component_rankings.len(), 1); assert_eq!(model.child_rankings.component_rankings[0].pairs, 1); let names: Vec<&str> = model.child_rankings.component_rankings[0] @@ -1625,6 +1770,7 @@ mod tests { &reduced, &ScopeId::Room("9ab12cd/my-room".to_string()), root.as_str(), + 1, ); assert!(!model.item_has_parent); assert_eq!(model.child_rankings.unranked_items.len(), 2); @@ -1655,6 +1801,7 @@ mod tests { &reduced, &ScopeId::Room("9ab12cd/my-room".to_string()), root.as_str(), + 1, ); assert_eq!(model.child_rankings.component_rankings.len(), 1); assert_eq!(model.child_rankings.component_rankings[0].pairs, 1); @@ -1680,11 +1827,56 @@ mod tests { "@00000000-0000-0000-0000-000000000000:test:local/test\n~/x {x}\n", ); let model = - build_item_page_view_model(&reduced, &ScopeId::Public, "https://slug.social/~/"); + build_item_page_view_model(&reduced, &ScopeId::Public, "https://slug.social/~/", 1); assert_eq!(model.child_rankings.unranked_items.len(), 1); assert_eq!( model.child_rankings.unranked_items[0].as_str(), "https://slug.social/~/x" ); } + + #[test] + fn item_page_model_depth_includes_descendants() { + let mut reduced = ReducerState::default(); + apply_ingest( + &mut reduced, + 1, + "@00000000-0000-0000-0000-000000000000:test:local/test\n\ + ~/topic {root}\n\ + ~/topic/a {alpha}\n\ + ~/topic/a/leaf {leaf}\n\ + ~/topic/b {beta}\n", + ); + + let model = build_item_page_view_model(&reduced, &ScopeId::Public, "~/topic", 2); + let items: std::collections::HashSet<&str> = model + .child_rankings + .unranked_items + .iter() + .map(|u| u.as_str()) + .collect(); + assert!(items.contains("https://slug.social/~/topic/a")); + assert!(items.contains("https://slug.social/~/topic/a/leaf")); + assert!(items.contains("https://slug.social/~/topic/b")); + } + + #[test] + fn external_source_href_maps_youtube_path_identity_back_to_watch_url() { + assert_eq!( + external_source_href("https://www.youtube.com/watch/v/dQw4w9WgXcQ"), + "https://www.youtube.com/watch?v=dQw4w9WgXcQ" + ); + assert_eq!( + external_source_href("https://github.com/sortersocial/slug"), + "https://github.com/sortersocial/slug" + ); + } + + #[test] + fn external_frame_allowed_skips_known_blocked_hosts() { + assert!(!super::external_frame_allowed( + "https://github.com/sortersocial/slug" + )); + assert!(super::external_frame_allowed("https://example.com/path")); + } } diff --git a/server/src/html/ui_action.rs b/server/src/html/ui_action.rs index 874fe729865c8b6179c94745221d8b79d625f642..8ab6f2924c3c66f826f38c487c48d6140eee4671 100644 --- a/server/src/html/ui_action.rs +++ b/server/src/html/ui_action.rs @@ -66,10 +66,17 @@ pub enum HtmlUiAction { #[serde(default = "default_ui_form_action")] form_action: String, }, - /// Author redacts own post via `POST /ui`. - RedactPost { - post_id: String, + /// Resolve children or siblings for an external garden item. + ResolveExternal { + room_wire: String, + item_storage: String, + mode: String, + next: String, + #[serde(default = "default_ui_form_action")] + form_action: String, }, + /// Author redacts own post via `POST /ui`. + RedactPost { post_id: String }, /// Morph `#room-members-section` — members list open or collapsed (server-rendered). SetRoomMembersExpanded { room_wire: String, @@ -77,9 +84,7 @@ pub enum HtmlUiAction { expanded: bool, }, /// Delete the private room (Manage only); redirects to `/` on success. - DeleteRoom { - room: String, - }, + DeleteRoom { room: String }, /// Morph `#new-thread-ui-slot` inner — compose open or collapsed (`room_wire: "public"` for home). SetNewThreadComposeExpanded { room_wire: String, @@ -117,13 +122,14 @@ pub enum HtmlUiParseError { } /// Parse `__rpc__` JSON, apply `$form` holes from the rest of the form map, deserialize. -pub fn parse_html_ui_from_form(form: &HashMap) -> Result { - let template = form - .get(UI_RPC_FIELD) - .ok_or(HtmlUiParseError::MissingRpc)?; +pub fn parse_html_ui_from_form( + form: &HashMap, +) -> Result { + let template = form.get(UI_RPC_FIELD).ok_or(HtmlUiParseError::MissingRpc)?; let mut hole_map = form.clone(); hole_map.remove(UI_RPC_FIELD); - let v: Value = fill_template_from_form(template, &hole_map).map_err(HtmlUiParseError::Template)?; + let v: Value = + fill_template_from_form(template, &hole_map).map_err(HtmlUiParseError::Template)?; serde_json::from_value(v).map_err(HtmlUiParseError::Action) } diff --git a/server/src/lib.rs b/server/src/lib.rs index b902094653cf9319efd14e67b91436e269362441..84e94bbec144eae77de68482385941cd2c5845eb 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -3,11 +3,11 @@ pub mod paths; pub mod api; pub mod canonical_path; pub mod dsl; +pub mod event_log; +pub mod events; pub mod external_resolver; pub mod form_template; pub mod html; -pub mod event_log; -pub mod events; pub mod identity; pub mod middleware; pub mod path_types; @@ -50,16 +50,18 @@ pub fn create_app_state(cfg: AppConfig) -> AppState { js_tx, write_tx, views, + resolver_runs: Arc::new(RwLock::new(HashMap::new())), + github_resolver: Arc::new(crate::external_resolver::GitHubResolver::from_env()), }; - tokio::spawn(crate::api::write_actor::writer_actor(write_rx, state.clone())); + tokio::spawn(crate::api::write_actor::writer_actor( + write_rx, + state.clone(), + )); state } /// Spawn the serialized writer. Used by integration tests after seeding reducer state in-process. -pub fn spawn_writer_actor_for_test( - state: AppState, - rx: tokio::sync::mpsc::Receiver, -) { +pub fn spawn_writer_actor_for_test(state: AppState, rx: tokio::sync::mpsc::Receiver) { tokio::spawn(crate::api::write_actor::writer_actor(rx, state)); } @@ -89,10 +91,7 @@ pub fn create_app(state: AppState) -> Router { .route("/-", get(crate::html::external_garden_index)) .route("/-/*path", get(crate::html::external_ontology_path)) .route("/r/:room_key/~", get(crate::html::room_garden_index)) - .route( - "/r/:room_key/~/*path", - get(crate::html::room_ontology_path), - ) + .route("/r/:room_key/~/*path", get(crate::html::room_ontology_path)) .route( "/r/:room_key/-", get(crate::html::room_external_garden_index), diff --git a/server/src/state.rs b/server/src/state.rs index 13831e868031edc19fda0f347376e0b17e8b99b7..8a80fed0a6bc7ae3516c76ab9d1e8592d7038ad1 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -3,7 +3,10 @@ use std::sync::Arc; use tokio::sync::{broadcast, mpsc, RwLock}; -use crate::{event_log::EventLog, events::ThreadCapability, reducer::ReducerState, write_cmd::WriteCmd}; +use crate::{ + event_log::EventLog, events::ThreadCapability, external_resolver::GitHubResolver, + reducer::ReducerState, write_cmd::WriteCmd, +}; /// Ephemeral invite link (24h TTL, in-memory only; not written to the event log). #[derive(Debug, Clone)] @@ -24,7 +27,7 @@ pub struct PendingSession { 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*/ )>, + pub complete: Option<(String /*username*/, String /*bearer*/)>, } /// An SSE event broadcast to all live stream subscribers when an ingest occurs. @@ -68,6 +71,9 @@ pub struct AppState { /// All durable writes and reducer mutations are serialized through this channel. pub write_tx: mpsc::Sender, pub views: crate::views::ViewStore, + /// Ephemeral resolver cooldowns keyed by `room + item`: not persisted; durable results are JSONL ingests. + pub resolver_runs: Arc>>, + pub github_resolver: Arc, } impl AppState { @@ -94,7 +100,8 @@ impl AppState { js_tx, write_tx, views, + resolver_runs: Arc::new(RwLock::new(HashMap::new())), + github_resolver: Arc::new(GitHubResolver::from_env()), } } } - diff --git a/server/src/write_cmd.rs b/server/src/write_cmd.rs index 494aaeadf7bb919768788e101c3c1ffa12da8503..094a1b80b64caa327af523dc36c618ea04551cdc 100644 --- a/server/src/write_cmd.rs +++ b/server/src/write_cmd.rs @@ -14,6 +14,13 @@ pub enum WriteCmd { bearer: String, reply: oneshot::Sender, }, + SystemIngest { + room: String, + thread_tag: String, + text: String, + principal: String, + reply: oneshot::Sender, + }, Redact { post_id: String, bearer: String, diff --git a/test/browser_github_resolver.clj b/test/browser_github_resolver.clj new file mode 100644 index 0000000000000000000000000000000000000000..1d5b6f1c4575e67ec430e4ce7f02c1bca5a35775 --- /dev/null +++ b/test/browser_github_resolver.clj @@ -0,0 +1,133 @@ +(ns test.browser-github-resolver + "Playwright coverage for on-demand GitHub external resolver imports." + (:require [babashka.fs :as fs] + [cheshire.core :as json] + [clojure.string :as str] + [clojure.test :refer [deftest is]] + [com.blockether.spel.core :as core] + [com.blockether.spel.locator :as locator] + [com.blockether.spel.page :as page] + [org.httpkit.server :as http] + [test.common :as common] + [test.oauth :as oauth])) + +(defn- wait-for-text [pg selector expected timeout-ms] + (let [deadline (+ (System/currentTimeMillis) timeout-ms)] + (loop [] + (let [text (locator/text-content (page/locator pg selector))] + (if (and (string? text) (str/includes? text expected)) + true + (if (< (System/currentTimeMillis) deadline) + (do (Thread/sleep 200) (recur)) + false)))))) + +(defn- start-mock-github [port] + (let [!paths (atom []) + handler (fn [req] + (swap! !paths conj (:uri req)) + (case (:uri req) + "/users/octo/repos" + {:status 200 + :headers {"Content-Type" "application/json"} + :body (json/generate-string + [{:name "hello" + :full_name "octo/hello" + :description "Hello repo"} + {:name "other" + :full_name "octo/other" + :description "Other repo"}])} + + "/repos/octo/hello/issues" + {:status 200 + :headers {"Content-Type" "application/json"} + :body (json/generate-string + [{:number 42 + :title "Seeded issue" + :body "Already pasted"} + {:number 43 + :title "Imported sibling" + :body "Loaded from mock GitHub"} + {:number 99 + :title "PR should be filtered" + :pull_request {}}])} + + {:status 404 :body "not found"})) + stop-fn (http/run-server handler {:port port})] + {:stop-fn stop-fn :paths !paths})) + +(defn github-resolver-flow! [] + (println "\n━━━ browser GitHub resolver: on-demand children and siblings ━━━\n") + + (common/letlocals + (bind build (common/run-cargo-build-release! ["slugsocial-server"])) + (is (zero? (:exit build)) "cargo build succeeds") + (bind server-bin "target/release/slugsocial-server") + + (bind tmp-dir (str (fs/create-temp-dir {:prefix "slug-browser-github-resolver-"}))) + (bind slug-port (common/pick-port)) + (bind google-port (common/pick-port)) + (bind github-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 github-url (str "http://127.0.0.1:" github-port)) + + (bind !server (atom nil)) + (bind !google (atom nil)) + (bind !github (atom nil)) + (bind server-env (assoc (common/slug-server-env tmp-dir base-url google-url slug-port) + "SLUG_GITHUB_API_BASE_URL" github-url)) + (try + (reset! !google (oauth/start-mock-google google-port + :google-users ["google-user-alice"])) + (reset! !github (start-mock-github github-port)) + (reset! !server (common/start-server server-bin server-env)) + (is (common/wait-for-server base-url 10000) "server responds to /healthz") + + (let [alice-token (oauth/fetch-bearer-token! base-url :username "alice") + seed-resp (oauth/http-post-json + (str base-url "/api/v0/rpc") + [{"Post" {"room" "public" + "thread_tag" "github-resolver-seed" + "text" "https://github.com/octo/hello/issues/42 {Seed pasted issue}" + "return_rank_diff" false}}] + :headers {"Authorization" (str "Bearer " alice-token)}) + seed-json (json/parse-string (:body seed-resp) false)] + (is (true? (get-in seed-json ["results" 0 "ok"])) "seed pasted GitHub issue") + + (core/with-playwright [pw] + (core/with-browser [browser (core/launch-chromium pw {:headless true :channel "chrome"})] + (core/with-context [ctx (core/new-context browser)] + (core/with-page [pg (core/new-page-from-context ctx)] + (page/navigate pg (str base-url "/login")) + (is (wait-for-text pg "body" "@alice" 15000) "alice session after login") + + (page/navigate pg (str base-url "/-/github.com/octo")) + (is (wait-for-text pg "#external-resolver-panel" "GitHub resolver" 15000) + "user page shows resolver panel") + (locator/click (page/locator pg "[data-testid=\"github-resolve-children\"]")) + (is (wait-for-text pg "body" "-/github.com/octo/other" 15000) + "user resolver imports repos") + + (page/navigate pg (str base-url "/-/github.com/octo/hello")) + (locator/click (page/locator pg "[data-testid=\"github-resolve-children\"]")) + (is (wait-for-text pg "body" "-/github.com/octo/hello/pulls" 15000) + "repo resolver imports structural children") + + (page/navigate pg (str base-url "/-/github.com/octo/hello/issues/42")) + (locator/click (page/locator pg "[data-testid=\"github-resolve-siblings\"]")) + (page/navigate pg (str base-url "/-/github.com/octo/hello/issues")) + (is (wait-for-text pg "body" "-/github.com/octo/hello/issues/43" 15000) + "issue sibling resolver imports issue siblings"))))) + + (is (some #{"/users/octo/repos"} @(:paths @!github)) + "mock GitHub saw user repos request") + (is (some #{"/repos/octo/hello/issues"} @(:paths @!github)) + "mock GitHub saw repo issues request")) + + (finally + (when-some [s @!server] (common/kill-server s)) + (when-some [g @!google] ((:stop-fn g))) + (when-some [g @!github] ((:stop-fn g))))))) + +(deftest browser-github-resolver-test + (github-resolver-flow!)) diff --git a/test/browser_ui_morph.clj b/test/browser_ui_morph.clj index 1dad7eb142728e14f0fe1a77747c6f443192cd03..7740a44546de12731833f048a19c35af8c6f2d1e 100644 --- a/test/browser_ui_morph.clj +++ b/test/browser_ui_morph.clj @@ -73,8 +73,8 @@ (is (wait-for-text pg "nav.breadcrumb" "repo" 5000) "breadcrumb includes path segment repo") (is (wait-for-text pg ".ont-external-empty" "This is an external scope." 10000) "external item shows empty-state copy") - (is (wait-for-text pg ".ont-external-empty" "Kick off an Agent Run to import and rank items" 5000) - "external empty state shows disabled agent-run button label") + (is (wait-for-text pg ".ont-external-empty" "Open external URL" 5000) + "external empty state links to source URL") (page/navigate pg (str base-url "/t/" thread-tag)) (is (wait-for-text pg "#thread-feed-region" "[show full post]" 15000) "truncated card shows expand link") diff --git a/tests.edn b/tests.edn index 103d38a1769a8f63754f08fb1e42a24c80ee37dc..5e2b0433a66c51806d08ef38d691259cfccd1fe0 100644 --- a/tests.edn +++ b/tests.edn @@ -20,7 +20,8 @@ "^test\\.browser-public-garden$" "^test\\.browser-redact-thread-index$" "^test\\.browser-garden-pin$" - "^test\\.browser-vote-compare$"] + "^test\\.browser-vote-compare$" + "^test\\.browser-github-resolver$"] :kaocha.filter/skip-meta [:skip] :parallel? false}] :plugins [:kaocha.plugin/junit-xml] diff --git a/types/src/paths.rs b/types/src/paths.rs index 8669f6872b9fa71d72a3f5c346945427bc00ac5a..791a44a0850b54b2e6baf16b47726bde507194c6 100644 --- a/types/src/paths.rs +++ b/types/src/paths.rs @@ -361,13 +361,14 @@ mod tests { SLUG_TILDE_ONTOLOGY_ROOT.to_string() ); assert_eq!( - ItemId::parse("https://slug.social/~/") - .unwrap() - .as_str(), + ItemId::parse("https://slug.social/~/").unwrap().as_str(), SLUG_TILDE_ONTOLOGY_ROOT ); let legacy = ItemId::opaque("https://slug.social/~/".to_string()); - assert_eq!(legacy.normalized_storage().as_str(), SLUG_TILDE_ONTOLOGY_ROOT); + assert_eq!( + legacy.normalized_storage().as_str(), + SLUG_TILDE_ONTOLOGY_ROOT + ); } #[test] @@ -423,7 +424,8 @@ mod tests { #[test] fn garden_private_room_prefixes_ontology() { assert_eq!( - GardenItemUrl::from_storage_str("https://slug.social/~/topic/x", "9ab12cd/my-room").as_str(), + GardenItemUrl::from_storage_str("https://slug.social/~/topic/x", "9ab12cd/my-room") + .as_str(), "https://slug.social/r/9ab12cdmy-room/~/topic/x" ); } @@ -465,11 +467,11 @@ mod tests { fn canonicalize_youtube_short_links() { assert_eq!( canonicalize_item("https://youtu.be/dQw4w9WgXcQ"), - "https://www.youtube.com/watch?v=dQw4w9WgXcQ" + "https://youtu.be/dQw4w9WgXcQ" ); assert_eq!( canonicalize_item("-/youtu.be/dQw4w9WgXcQ"), - "https://www.youtube.com/watch?v=dQw4w9WgXcQ" + "https://youtu.be/dQw4w9WgXcQ" ); } @@ -487,12 +489,29 @@ mod tests { } #[test] - fn item_id_parent_external_strips_last_segment() { - let c = ItemId::parse("https://spotify.com/track/1").unwrap(); + fn canonicalize_external_url_strips_fragment_and_query_params() { + assert_eq!( + canonicalize_item("https://example.com/path?utm_source=newsletter&b=2#section"), + "https://example.com/path" + ); assert_eq!( - c.parent().unwrap().as_str(), - "https://spotify.com/track" + canonicalize_item("-/example.com/path?FbClId=abc&a=1"), + "https://example.com/path" ); + } + + #[test] + fn canonicalize_github_external_url_normalizes_repo_identity() { + assert_eq!( + canonicalize_item("https://github.com/ORG/REPO.git?tab=issues&q=is%3Aopen"), + "https://github.com/org/repo" + ); + } + + #[test] + fn item_id_parent_external_strips_last_segment() { + let c = ItemId::parse("https://spotify.com/track/1").unwrap(); + assert_eq!(c.parent().unwrap().as_str(), "https://spotify.com/track"); assert_eq!( ItemId::parse("https://github.com/iss/1") .unwrap() @@ -501,7 +520,10 @@ mod tests { .as_str(), "https://github.com/iss" ); - assert!(ItemId::parse("https://github.com").unwrap().parent().is_none()); + assert!(ItemId::parse("https://github.com") + .unwrap() + .parent() + .is_none()); } #[test] diff --git a/types/src/url_normalize.rs b/types/src/url_normalize.rs index 999baa4f3367b632763b44ddc77b894451207c50..b655f4d834dd5a482aa1b5adad25b33461dbb76e 100644 --- a/types/src/url_normalize.rs +++ b/types/src/url_normalize.rs @@ -1,7 +1,9 @@ //! Normalization for external `http(s)://` item identity (not slug tilde ontology). //! //! Policy (intentional, extend here as new domains need treatment): -//! - Query pairs sorted lexicographically by **lowercased** key, then value. +//! - Query pairs are navigation/filter state by default and are stripped from identity. +//! - Known query-primary resources may canonicalize to path-based short links (YouTube `youtu.be/:id`). +//! - Fragments are stripped from identity. //! - YouTube family → stable `www.youtube.com` shapes where possible. use url::Url; @@ -23,8 +25,10 @@ pub fn normalize_http_identity_url(s: &str) -> Option { if !matches!(u.scheme(), "http" | "https") { return None; } + u.set_fragment(None); rewrite_youtube(&mut u); - sort_query_pairs(&mut u); + normalize_github(&mut u); + u.set_query(None); Some(u.to_string()) } @@ -45,23 +49,18 @@ fn rewrite_youtube(u: &mut Url) { match base.as_str() { "youtu.be" => { - let id = path.trim_start_matches('/').split('/').next().unwrap_or("").to_string(); + let id = path + .trim_start_matches('/') + .split('/') + .next() + .unwrap_or("") + .to_string(); if id.is_empty() { return; } - let saved: Vec<(String, String)> = u.query_pairs().into_owned().collect(); - let Ok(mut out) = Url::parse(&format!("https://www.youtube.com/watch?v={id}")) else { + let Ok(out) = Url::parse(&format!("https://youtu.be/{id}")) else { return; }; - { - let mut q = out.query_pairs_mut(); - for (k, v) in saved { - if k.eq_ignore_ascii_case("v") { - continue; - } - q.append_pair(&k, &v); - } - } *u = out; } "youtube.com" | "m.youtube.com" => { @@ -80,20 +79,9 @@ fn rewrite_youtube(u: &mut Url) { if id.is_empty() { return; } - let saved: Vec<(String, String)> = u.query_pairs().into_owned().collect(); - let Ok(mut out) = Url::parse(&format!("https://www.youtube.com/watch?v={id}")) - else { + let Ok(out) = Url::parse(&format!("https://youtu.be/{id}")) else { return; }; - { - let mut q = out.query_pairs_mut(); - for (k, v) in saved { - if k.eq_ignore_ascii_case("v") { - continue; - } - q.append_pair(&k, &v); - } - } *u = out; return; } @@ -109,25 +97,21 @@ fn rewrite_youtube(u: &mut Url) { if id.is_empty() { return; } - let saved: Vec<(String, String)> = u.query_pairs().into_owned().collect(); - let Ok(mut out) = Url::parse(&format!("https://www.youtube.com/watch?v={id}")) - else { + let Ok(out) = Url::parse(&format!("https://youtu.be/{id}")) else { return; }; - { - let mut q = out.query_pairs_mut(); - for (k, v) in saved { - if k.eq_ignore_ascii_case("v") { - continue; - } - q.append_pair(&k, &v); - } - } *u = out; return; } if path.starts_with("/watch") { let _ = u.set_host(Some("www.youtube.com")); + if let Some((_, id)) = u.query_pairs().find(|(k, _)| k.eq_ignore_ascii_case("v")) { + if !id.is_empty() { + if let Ok(out) = Url::parse(&format!("https://youtu.be/{id}")) { + *u = out; + } + } + } return; } if path.starts_with("/shorts/") { @@ -142,17 +126,9 @@ fn rewrite_youtube(u: &mut Url) { if id.is_empty() { return; } - let saved: Vec<(String, String)> = u.query_pairs().into_owned().collect(); - let Ok(mut out) = Url::parse(&format!("https://www.youtube.com/shorts/{id}")) - else { + let Ok(out) = Url::parse(&format!("https://www.youtube.com/shorts/{id}")) else { return; }; - { - let mut q = out.query_pairs_mut(); - for (k, v) in saved { - q.append_pair(&k, &v); - } - } *u = out; return; } @@ -163,25 +139,32 @@ fn rewrite_youtube(u: &mut Url) { } } -fn sort_query_pairs(u: &mut Url) { - let pairs: Vec<(String, String)> = u.query_pairs().into_owned().collect(); - if pairs.is_empty() { - u.set_query(None); +fn normalize_github(u: &mut Url) { + let Some(host_raw) = u.host_str() else { + return; + }; + if host_raw.to_ascii_lowercase() != "github.com" { return; } - let mut pairs = pairs; - pairs.sort_by(|a, b| { - a.0.to_ascii_lowercase() - .cmp(&b.0.to_ascii_lowercase()) - .then_with(|| a.1.cmp(&b.1)) - }); - u.set_query(None); - { - let mut q = u.query_pairs_mut(); - for (k, v) in pairs { - q.append_pair(&k, &v); + + let Some(segments) = u.path_segments() else { + return; + }; + let mut segments: Vec = segments + .filter(|s| !s.is_empty()) + .map(|s| s.to_ascii_lowercase()) + .collect(); + if let Some(repo) = segments.get_mut(1) { + if let Some(stripped) = repo.strip_suffix(".git") { + *repo = stripped.to_string(); } } + + if segments.is_empty() { + u.set_path("/"); + } else { + u.set_path(&format!("/{}", segments.join("/"))); + } } #[cfg(test)] @@ -192,19 +175,19 @@ mod tests { fn youtube_youtu_be_to_watch() { assert_eq!( normalize_http_identity_url("https://youtu.be/dQw4w9WgXcQ").as_deref(), - Some("https://www.youtube.com/watch?v=dQw4w9WgXcQ") + Some("https://youtu.be/dQw4w9WgXcQ") ); } #[test] - fn youtube_watch_query_sorted() { + fn youtube_watch_v_query_becomes_path_child() { assert_eq!( normalize_http_identity_url("https://youtube.com/watch?v=Z&a=1&b=2").as_deref(), - Some("https://www.youtube.com/watch?a=1&b=2&v=Z") + Some("https://youtu.be/Z") ); assert_eq!( normalize_http_identity_url("https://youtube.com/watch?b=2&a=1&v=Z").as_deref(), - Some("https://www.youtube.com/watch?a=1&b=2&v=Z") + Some("https://youtu.be/Z") ); } @@ -212,7 +195,7 @@ mod tests { fn youtube_embed_to_watch() { assert_eq!( normalize_http_identity_url("https://www.youtube.com/embed/dQw4w9WgXcQ").as_deref(), - Some("https://www.youtube.com/watch?v=dQw4w9WgXcQ") + Some("https://youtu.be/dQw4w9WgXcQ") ); } @@ -225,10 +208,56 @@ mod tests { } #[test] - fn arbitrary_query_sorted() { + fn arbitrary_query_is_stripped() { assert_eq!( normalize_http_identity_url("https://example.com/x?z=1&a=2").as_deref(), - Some("https://example.com/x?a=2&z=1") + Some("https://example.com/x") + ); + } + + #[test] + fn fragment_is_stripped_from_identity() { + assert_eq!( + normalize_http_identity_url("https://example.com/path?b=2#a-section").as_deref(), + Some("https://example.com/path") + ); + } + + #[test] + fn all_generic_query_params_are_stripped() { + assert_eq!( + normalize_http_identity_url( + "https://example.com/x?z=1&utm_source=newsletter&FbClId=abc&ref=share&a=2" + ) + .as_deref(), + Some("https://example.com/x") + ); + } + + #[test] + fn stripping_tracking_params_removes_empty_query() { + assert_eq!( + normalize_http_identity_url("https://example.com/x?utm_medium=email&si=share") + .as_deref(), + Some("https://example.com/x") + ); + } + + #[test] + fn youtube_identity_query_is_promoted_before_query_cleanup() { + assert_eq!( + normalize_http_identity_url("https://youtu.be/dQw4w9WgXcQ?si=share&v=ignored&t=12") + .as_deref(), + Some("https://youtu.be/dQw4w9WgXcQ") + ); + } + + #[test] + fn github_repo_suffix_and_noise_query_are_normalized() { + assert_eq!( + normalize_http_identity_url("https://github.com/ORG/REPO.git?tab=readme&q=is%3Aopen") + .as_deref(), + Some("https://github.com/org/repo") ); } } @@ -324,7 +353,10 @@ mod url_identity_tests { fn empty_path_vs_slash_only_path_may_differ() { let root = Url::parse("https://example.com").unwrap(); let slash = Url::parse("https://example.com/").unwrap(); - assert_eq!(root, slash, "root and trailing-slash-only merge for this parser"); + assert_eq!( + root, slash, + "root and trailing-slash-only merge for this parser" + ); } #[test] @@ -353,6 +385,9 @@ mod url_identity_tests { let plus = Url::parse("https://example.com/?q=a+b").unwrap(); let encoded = Url::parse("https://example.com/?q=a%20b").unwrap(); - assert_ne!(plus, encoded, "space as + vs %20 — different keys unless normalized"); + assert_ne!( + plus, encoded, + "space as + vs %20 — different keys unless normalized" + ); } } Side B — contributor: tommy-mor Side B — commit message: [1d14ff09] Ephemeral Reddit content; log structure only (#46) * Keep Reddit content ephemeral; log structure only Remove EntityImported and EntityStore. Reddit fetches write display content directly to the projection with a fetched_at timestamp, while the event log records NodeEnsured for discovered identities only. A background task evicts cached display content after 48 hours. Votes, tree structure, and ItemIds remain in the log and projection. Co-authored-by: tommy * Fix reddit import test assertions and Clojure syntax Co-authored-by: tommy --------- Co-authored-by: Cursor Agent Side B — unified diff (full patch): diff --git a/server/src/bin/storage_bench.rs b/server/src/bin/storage_bench.rs index 4762d0d23bc8e2df439afe09a3235ba7f72bd486..3d988416ad36b27a7d3dc84280cfdbdcafa43e69 100644 --- a/server/src/bin/storage_bench.rs +++ b/server/src/bin/storage_bench.rs @@ -6,8 +6,8 @@ use std::{ }; use sorter2_server::{ - entity_store::EntityStore, event_log::EventLog, events::Event, journal::JournalClient, - projection_apply, projection_store::ProjectionStore, + event_log::EventLog, events::Event, journal::JournalClient, projection_apply, + projection_store::ProjectionStore, }; #[tokio::main] @@ -18,12 +18,10 @@ async fn main() -> Result<(), Box> { let data_dir = opts.data_dir.to_string_lossy().into_owned(); let event_log = Arc::new(EventLog::new(format!("{data_dir}/events.jsonl"))); let db = durable::Db::open(opts.data_dir.join("store"))?; - let entity_store = EntityStore::from_db(&db)?; let projection_store = ProjectionStore::from_db(&db)?; let journal = JournalClient::spawn( event_log.clone(), - entity_store.clone(), projection_store.clone(), event_log.last_sequence().await? + 1, ); @@ -46,12 +44,9 @@ async fn main() -> Result<(), Box> { drop(journal); let rebuild_start = Instant::now(); - entity_store.reset()?; projection_store.reset()?; let rebuild = event_log - .replay(|record| { - projection_apply::apply_records(&projection_store, &entity_store, &[record]) - }) + .replay(|record| projection_apply::apply_records(&projection_store, &[record])) .await?; let rebuild_elapsed = rebuild_start.elapsed(); diff --git a/server/src/entity_store.rs b/server/src/entity_store.rs deleted file mode 100644 index d29f39aecca6f12cdcf263cf77c3654eb4ee6cfa..0000000000000000000000000000000000000000 --- a/server/src/entity_store.rs +++ /dev/null @@ -1,134 +0,0 @@ -//! Off-heap storage for full entity payloads (Reddit API JSON). -//! -//! Derived [`crate::reducer::EntityData`] is stored on the node; the raw JSON -//! lives here, in the shared durable [`Store`] schema. - -use std::path::Path; - -use durable::{Batch, Db, Durability}; -use serde_json::Value; - -use crate::{ - path_types::ItemId, - storage_dto::{decode_entity_payload, encode_entity_payload}, - storage_schema::{Store, StoreFields}, -}; - -const ENTITY_SCHEMA_KEY: &str = "schema_version"; -const ENTITY_SCHEMA_VERSION: u64 = 2; - -#[derive(Debug, thiserror::Error)] -pub enum EntityStoreError { - #[error("durable error: {0}")] - Durable(#[from] durable::Error), - #[error("json error: {0}")] - Json(#[from] serde_json::Error), - #[error("storage decode error: {0}")] - Storage(String), - #[error("io error: {0}")] - Io(#[from] std::io::Error), -} - -/// Disk-backed map of entity id → raw JSON payload. -#[derive(Clone)] -pub struct EntityStore { - db: Db, -} - -impl EntityStore { - /// Open (or create) the entity database under `dir`. - pub fn open(dir: &Path) -> Result { - std::fs::create_dir_all(dir)?; - let db = Db::open(dir)?; - Self::from_db(&db) - } - - /// Create an entity store backed by an already-open database. - pub fn from_db(db: &Db) -> Result { - let store = Self { db: db.clone() }; - let version = Store::root() - .entity_meta() - .key(&ENTITY_SCHEMA_KEY.to_string()) - .get(db)?; - if version != Some(ENTITY_SCHEMA_VERSION) { - store.reset()?; - } - Ok(store) - } - - /// Clear rebuildable entity payloads and reset storage schema metadata. - pub fn reset(&self) -> Result<(), EntityStoreError> { - let root = Store::root(); - self.db.apply( - &[root.entities().clear(), root.entity_meta().clear()], - Durability::SyncWal, - )?; - self.db.run( - root.entity_meta() - .key(&ENTITY_SCHEMA_KEY.to_string()) - .set(&ENTITY_SCHEMA_VERSION), - Durability::SyncWal, - )?; - Ok(()) - } - - /// Persist a payload for `id` (overwrites any existing entry). - pub fn put(&self, id: &ItemId, payload: &Value) -> Result<(), EntityStoreError> { - self.db.run( - Store::root() - .entities() - .key(&id.as_str().to_string()) - .set(&encode_entity_payload(payload)), - Durability::SyncWal, - )?; - Ok(()) - } - - /// Add a payload write to the caller's batch. - pub fn put_in_batch( - &self, - batch: &mut Batch, - id: &ItemId, - payload: &Value, - ) -> Result<(), EntityStoreError> { - batch.write( - Store::root() - .entities() - .key(&id.as_str().to_string()) - .set(&encode_entity_payload(payload)), - ); - Ok(()) - } - - /// Load a stored payload, if present. - pub fn get(&self, id: &ItemId) -> Result, EntityStoreError> { - match Store::root() - .entities() - .key(&id.as_str().to_string()) - .get(&self.db)? - { - Some(record) => decode_entity_payload(record) - .map(Some) - .map_err(EntityStoreError::Storage), - None => Ok(None), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use serde_json::json; - - #[test] - fn round_trip_payload() { - let tmp = tempfile::tempdir().unwrap(); - let store = EntityStore::open(tmp.path()).unwrap(); - let id = ItemId::from_url("https://reddit.com/r/rust").unwrap(); - let payload = json!({"kind": "t5", "data": {"display_name": "rust"}}); - - store.put(&id, &payload).unwrap(); - let loaded = store.get(&id).unwrap().unwrap(); - assert_eq!(loaded, payload); - } -} diff --git a/server/src/events.rs b/server/src/events.rs index a3d88e285c645a96d33e6f0aed1b487b843aceb8..d76c3bb4277216b0d39c9422ba7a50db10a95e05 100644 --- a/server/src/events.rs +++ b/server/src/events.rs @@ -1,5 +1,4 @@ use serde::{Deserialize, Serialize}; -use serde_json::Value; /// Schema version for JSONL log records. Bump when event semantics change. pub const CURRENT_LOG_SCHEMA: u32 = 1; @@ -30,7 +29,7 @@ pub type ViewRecord = LogRecord; /// Wall-clock timestamp carried on the log envelope for domain events. pub fn event_timestamp(event: &Event) -> i64 { match event { - Event::VoteRecorded { ts, .. } | Event::EntityImported { ts, .. } => *ts, + Event::VoteRecorded { ts, .. } => *ts, Event::NodeEnsured { .. } => crate::fetch::now_ms(), } } @@ -57,6 +56,4 @@ pub enum Event { }, /// Register a node path in the fractal tree (no external fetch). NodeEnsured { id: String }, - /// Full upstream API payload for a node (domain-specific view derived at replay/render time). - EntityImported { id: String, ts: i64, payload: Value }, } diff --git a/server/src/journal.rs b/server/src/journal.rs index 50bc89f976edb82b7b0e49e954a8eccbbe82bf87..d50023aca7c3e74068de988449b11baee32d1ecf 100644 --- a/server/src/journal.rs +++ b/server/src/journal.rs @@ -5,7 +5,6 @@ use std::sync::Arc; use tokio::sync::{mpsc, oneshot}; use crate::{ - entity_store::EntityStore, event_log::EventLog, events::{event_timestamp, Event, EventRecord}, projection_apply, @@ -25,7 +24,6 @@ pub struct JournalClient { impl JournalClient { pub fn spawn( event_log: Arc, - entity_store: EntityStore, projection_store: ProjectionStore, next_seq: u64, ) -> Self { @@ -33,7 +31,6 @@ impl JournalClient { tokio::spawn(journal_worker( rx, event_log, - entity_store, projection_store, next_seq, )); @@ -62,7 +59,6 @@ impl JournalClient { async fn journal_worker( mut rx: mpsc::Receiver, event_log: Arc, - entity_store: EntityStore, projection_store: ProjectionStore, mut next_seq: u64, ) { @@ -75,7 +71,6 @@ async fn journal_worker( let result = append_and_project_batch( &event_log, &projection_store, - &entity_store, &mut next_seq, &batch, ) @@ -99,7 +94,6 @@ async fn journal_worker( async fn append_and_project_batch( event_log: &EventLog, projection_store: &ProjectionStore, - entity_store: &EntityStore, next_seq: &mut u64, commands: &[JournalCommand], ) -> Result<(), String> { @@ -117,7 +111,7 @@ async fn append_and_project_batch( .await .map_err(|e| e.to_string())?; *next_seq = seq; - projection_apply::apply_records(projection_store, entity_store, &records) + projection_apply::apply_records(projection_store, &records) .map_err(|e| format!("projection apply failed after durable append: {e}")) } @@ -132,10 +126,9 @@ mod tests { let log_path = tmp.path().join("events.jsonl"); let event_log = Arc::new(EventLog::new(log_path)); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - let journal = JournalClient::spawn(event_log, entity_store, projection_store.clone(), 1); + let journal = JournalClient::spawn(event_log, projection_store.clone(), 1); let j1 = journal.clone(); let j2 = journal.clone(); @@ -177,11 +170,9 @@ mod tests { .unwrap(); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); projection_apply::apply_records( &projection_store, - &entity_store, &[EventRecord::new( 1, 1, @@ -196,7 +187,6 @@ mod tests { let journal = JournalClient::spawn( event_log.clone(), - entity_store, projection_store.clone(), next_seq, ); @@ -219,10 +209,9 @@ mod tests { let log_path = tmp.path().join("events.jsonl"); let event_log = Arc::new(EventLog::new(log_path)); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); let journal = - JournalClient::spawn(event_log.clone(), entity_store, projection_store.clone(), 1); + JournalClient::spawn(event_log.clone(), projection_store.clone(), 1); journal .append_many(vec![ diff --git a/server/src/lib.rs b/server/src/lib.rs index 5c02c8e704e4664453bad75d819df8a067668176..3dfc7c8acb8ed61bb73ade63e72768e402042cc5 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -1,5 +1,4 @@ pub mod api; -pub mod entity_store; pub mod event_log; pub mod events; pub mod fetch; diff --git a/server/src/projection_apply.rs b/server/src/projection_apply.rs index 5644557a41b3e9497c7421b444155ae629fa79f1..9c8990a8af927f35d3344c8d0872a516aba56b86 100644 --- a/server/src/projection_apply.rs +++ b/server/src/projection_apply.rs @@ -1,22 +1,20 @@ //! Apply event-log records to the durable projection as precise point updates. //! //! Each batch of records lowers to reified durable writes (edge merges, child -//! links, voted-pair flags, recent-vote pushes, entity payloads) plus a cursor -//! advance, all committed in one atomic `DisableWal` batch. The cursor moving in -//! the same batch as the (non-idempotent) edge merges guarantees exactly-once -//! application across replay. +//! links, voted-pair flags, recent-vote pushes) plus a cursor advance, all +//! committed in one atomic `DisableWal` batch. The cursor moving in the same +//! batch as the (non-idempotent) edge merges guarantees exactly-once application +//! across replay. use std::collections::BTreeSet; use crate::{ - entity_store::EntityStore, event_log::EventLogError, events::{Event, EventRecord}, path_types::ItemId, projection_store::ProjectionStore, - reddit::entity_view_from_payload, reducer::VoteData, - storage_schema::{ensure_path_writes, entity_view_writes, vote_writes}, + storage_schema::{ensure_path_writes, vote_writes}, }; fn parse_event_id(id: &str) -> Result { @@ -38,7 +36,6 @@ fn parent_from_event_scope(scope: &str) -> ItemId { pub fn apply_records( projection_store: &ProjectionStore, - entity_store: &EntityStore, records: &[EventRecord], ) -> Result<(), EventLogError> { if records.is_empty() { @@ -79,14 +76,6 @@ pub fn apply_records( let parsed = parse_event_id(id)?; ensure_path_writes(&mut batch, &parsed); } - Event::EntityImported { id, payload, .. } => { - let parsed = parse_event_id(id)?; - let view = entity_view_from_payload(&parsed, payload); - entity_view_writes(&mut batch, &parsed, view.as_ref()); - entity_store - .put_in_batch(&mut batch, &parsed, payload) - .map_err(|e| EventLogError::Apply(e.to_string()))?; - } } last_seq = record.seq; } diff --git a/server/src/projection_store.rs b/server/src/projection_store.rs index 30ee478f953e80d7322bdbfa521ae559ff08fa51..8576d671f351004426207894ac35594ddb0f70cf 100644 --- a/server/src/projection_store.rs +++ b/server/src/projection_store.rs @@ -9,13 +9,16 @@ use durable::{Db, Durability, Write}; use crate::{ path_types::ItemId, - reducer::{GlobalTree, NodeState}, - storage_schema::{load_node_state, node, NodeSchemaFields, Store, StoreFields}, + reducer::{EntityData, GlobalTree, NodeState}, + storage_schema::{ + entity_content_clear_writes, entity_content_writes, load_node_state, node, NodeSchemaFields, + Store, StoreFields, + }, }; const PROJECTION_CURSOR_KEY: &str = "cursor"; const PROJECTION_SCHEMA_KEY: &str = "schema_version"; -const PROJECTION_SCHEMA_VERSION: u64 = 2; +const PROJECTION_SCHEMA_VERSION: u64 = 3; #[derive(Debug, thiserror::Error)] pub enum ProjectionStoreError { @@ -148,6 +151,45 @@ impl ProjectionStore { )?; Ok(()) } + + /// Cache Reddit display content outside the event log (must be evicted per policy). + pub fn put_ephemeral_content( + &self, + id: &ItemId, + view: &EntityData, + fetched_at: i64, + ) -> Result<(), ProjectionStoreError> { + let mut batch = self.db.batch(); + entity_content_writes(&mut batch, id, view, fetched_at); + batch + .commit_with(Durability::DisableWal) + .map_err(ProjectionStoreError::from)?; + Ok(()) + } + + /// Drop cached display content older than `cutoff_ms` (votes and tree structure remain). + pub fn evict_content_older_than(&self, cutoff_ms: i64) -> Result { + let keys = Store::root().nodes().keys(&self.db)?; + let mut batch = self.db.batch(); + let mut evicted = 0usize; + for key in keys { + let id = parse_node_key(&key)?; + let np = node(&id); + let Some(fetched_at) = np.fetched_at().get(&self.db)? else { + continue; + }; + if fetched_at > 0 && fetched_at < cutoff_ms { + entity_content_clear_writes(&mut batch, &id); + evicted += 1; + } + } + if evicted > 0 { + batch + .commit_with(Durability::DisableWal) + .map_err(ProjectionStoreError::from)?; + } + Ok(evicted) + } } fn parse_node_key(key: &str) -> Result { @@ -162,7 +204,7 @@ fn parse_node_key(key: &str) -> Result { #[cfg(test)] mod tests { use super::*; - use crate::{entity_store::EntityStore, events::Event, projection_apply}; + use crate::{events::Event, projection_apply, reducer::EntityData}; fn record(seq: u64, event: Event) -> crate::events::EventRecord { crate::events::EventRecord::new(seq, crate::events::event_timestamp(&event), event) @@ -172,7 +214,6 @@ mod tests { fn applies_and_loads_reducer_nodes() { let tmp = tempfile::tempdir().unwrap(); let db = Db::open(tmp.path()).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let store = ProjectionStore::from_db(&db).unwrap(); let event = Event::VoteRecorded { @@ -183,7 +224,7 @@ mod tests { ratio_right: 1, scope: String::new(), }; - projection_apply::apply_records(&store, &entity_store, &[record(1, event)]).unwrap(); + projection_apply::apply_records(&store, &[record(1, event)]).unwrap(); assert_eq!(store.last_applied_event_count().unwrap(), 1); let loaded = store.load_tree().unwrap(); @@ -196,7 +237,6 @@ mod tests { fn hydrates_scope_with_child_nodes() { let tmp = tempfile::tempdir().unwrap(); let db = Db::open(tmp.path()).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let store = ProjectionStore::from_db(&db).unwrap(); let event = Event::VoteRecorded { @@ -207,11 +247,36 @@ mod tests { ratio_right: 1, scope: String::new(), }; - projection_apply::apply_records(&store, &entity_store, &[record(1, event)]).unwrap(); + projection_apply::apply_records(&store, &[record(1, event)]).unwrap(); let scoped = store.scope_tree(&ItemId::root()).unwrap(); let root = scoped.get(&ItemId::root()).unwrap(); assert_eq!(root.children.len(), 2); assert!(scoped.get(&ItemId::opaque("alpha")).is_some()); } + + #[test] + fn evicts_stale_ephemeral_content() { + let tmp = tempfile::tempdir().unwrap(); + let db = Db::open(tmp.path()).unwrap(); + let store = ProjectionStore::from_db(&db).unwrap(); + let id = ItemId::from_url("https://reddit.com/r/rust").unwrap(); + store + .put_ephemeral_content( + &id, + &EntityData { + title: "Rust".into(), + author: None, + body_html: None, + thumb_url: None, + image_url: None, + link_url: None, + }, + 1_000, + ) + .unwrap(); + assert!(store.load_node(&id).unwrap().unwrap().data.is_some()); + assert_eq!(store.evict_content_older_than(2_000).unwrap(), 1); + assert!(store.load_node(&id).unwrap().unwrap().data.is_none()); + } } diff --git a/server/src/reddit.rs b/server/src/reddit.rs index 20b7f9e3f8be39268a1767d09f5cf81eaa6ae0df..a874814f8927192ee62cab2d0db1efd27dcd57b7 100644 --- a/server/src/reddit.rs +++ b/server/src/reddit.rs @@ -9,10 +9,13 @@ use serde_json::Value; use tokio::sync::{mpsc, oneshot}; use crate::{ - entity_store::EntityStore, events::Event, fetch::now_ms, journal::JournalClient, - path_types::ItemId, reducer::GlobalTree, + events::Event, fetch::now_ms, journal::JournalClient, + path_types::ItemId, projection_store::ProjectionStore, }; +/// Reddit display content must not be retained longer than this (API policy). +pub const REDDIT_CONTENT_TTL: std::time::Duration = std::time::Duration::from_secs(48 * 3600); + #[derive(Debug, Clone, PartialEq, Eq)] pub enum FetchJobResult { /// Number of entities written (1 for self, N for children). @@ -70,7 +73,11 @@ struct OAuthToken { } impl RedditBroker { - pub fn spawn(journal: JournalClient, config: RedditApiConfig) -> Self { + pub fn spawn( + journal: JournalClient, + projection_store: ProjectionStore, + config: RedditApiConfig, + ) -> Self { let (tx, rx) = mpsc::channel(100); let mut headers = header::HeaderMap::new(); @@ -93,7 +100,7 @@ impl RedditBroker { "reddit worker started" ); - tokio::spawn(reddit_worker(rx, journal, client, config)); + tokio::spawn(reddit_worker(rx, journal, projection_store, client, config)); Self { tx } } @@ -189,27 +196,50 @@ pub fn entity_view_from_payload( None } -pub fn apply_entity_import( - tree: &mut GlobalTree, - store: &EntityStore, - id: &ItemId, - payload: Value, -) -> Result<(), String> { - let view = entity_view_from_payload(id, &payload); - store.put(id, &payload).map_err(|e| e.to_string())?; - tree.apply_entity(id, view); - Ok(()) -} - fn notify(done: Option>, result: FetchJobResult) { if let Some(tx) = done { let _ = tx.send(result); } } +async fn import_fetched_payload( + kind: FetchKind, + fetch_id: &ItemId, + payload: Value, + projection_store: &ProjectionStore, + journal: &JournalClient, +) -> Result { + let fetched_at = now_ms(); + let imports: Vec<(ItemId, Value)> = match kind { + FetchKind::SelfEntity => vec![(fetch_id.clone(), payload)], + FetchKind::Children => parse_children(fetch_id, &payload), + }; + + for (id, child_payload) in &imports { + if let Some(view) = entity_view_from_payload(id, child_payload) { + projection_store + .put_ephemeral_content(id, &view, fetched_at) + .map_err(|e| e.to_string())?; + } + } + + let events: Vec = imports + .iter() + .map(|(id, _)| Event::NodeEnsured { + id: id.as_str().to_string(), + }) + .collect(); + let written = events.len(); + if !events.is_empty() { + journal.append_many(events).await?; + } + Ok(written) +} + async fn reddit_worker( mut rx: mpsc::Receiver, journal: JournalClient, + projection_store: ProjectionStore, client: Client, config: RedditApiConfig, ) { @@ -276,33 +306,20 @@ async fn reddit_worker( match outcome { Ok(FetchOutcome::Payload(payload)) => { - let imports: Vec<(ItemId, Value)> = match kind { - FetchKind::SelfEntity => vec![(fetch_id.clone(), payload)], - FetchKind::Children => parse_children(&fetch_id, &payload), - }; tracing::debug!( item = %fetch_id, ?kind, - count = imports.len(), - "reddit fetch got payload, importing" + "reddit fetch got payload, caching ephemerally" ); - let events: Vec = imports - .into_iter() - .map(|(child_id, child_payload)| Event::EntityImported { - id: child_id.as_str().to_string(), - ts: now_ms(), - payload: child_payload, - }) - .collect(); - let written = events.len(); - - match journal.append_many(events).await { + match import_fetched_payload(kind, &fetch_id, payload, &projection_store, &journal) + .await + { Err(e) => { - tracing::warn!(item = %fetch_id, err = %e, "reddit import journal failed"); + tracing::warn!(item = %fetch_id, err = %e, "reddit import failed"); notify(done, FetchJobResult::Failed(e)); } - Ok(()) => { + Ok(written) => { recently_fetched.insert(key.clone(), Instant::now()); current_delay = Duration::from_millis(600); tracing::info!(item = %fetch_id, ?kind, written, "reddit import complete"); diff --git a/server/src/reducer.rs b/server/src/reducer.rs index 1352a8f0771add3d868a1b30109b087a2a6dba6f..0c75c85150bb9e5f578bbadf58b3e43f8a80be4b 100644 --- a/server/src/reducer.rs +++ b/server/src/reducer.rs @@ -136,8 +136,7 @@ pub struct EntityData { #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct NodeState { pub id: ItemId, - /// Domain-specific view derived from imported payload (e.g. Reddit title/author). - /// Raw JSON lives in [`crate::entity_store::EntityStore`]. + /// Ephemeral display view (Reddit title/author/etc.; not event-logged). pub data: Option, pub children: HashSet, pub local_ranking: GroupState, diff --git a/server/src/state.rs b/server/src/state.rs index e44849eec46123072b238afd40a1fdd51ce19bd9..247b9047a57956f76c4b6bef691662101e62a8f9 100644 --- a/server/src/state.rs +++ b/server/src/state.rs @@ -1,14 +1,14 @@ use std::{error::Error, sync::Arc}; use crate::{ - entity_store::EntityStore, event_log::EventLog, events::Event, + fetch::now_ms, journal::JournalClient, path_types::ItemId, projection_apply, projection_store::ProjectionStore, - reddit::{RedditApiConfig, RedditBroker}, + reddit::{RedditApiConfig, RedditBroker, REDDIT_CONTENT_TTL}, reducer::{GlobalTree, VoteData}, view_log::ViewLog, views::ViewStore, @@ -45,7 +45,6 @@ pub fn normalize_scope(raw: &str) -> String { async fn catch_up_projection( event_log: &EventLog, - entity_store: &EntityStore, projection_store: &ProjectionStore, ) -> Result<(), crate::event_log::EventLogError> { let after_seq = projection_store @@ -54,7 +53,7 @@ async fn catch_up_projection( let stats = event_log .replay_from(after_seq, |record| { - projection_apply::apply_records(projection_store, entity_store, &[record]) + projection_apply::apply_records(projection_store, &[record]) }) .await?; if after_seq > stats.last_seq { @@ -67,22 +66,34 @@ async fn catch_up_projection( Ok(()) } +fn spawn_content_evictor(projection_store: ProjectionStore) { + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(15 * 60)); + interval.tick().await; + loop { + interval.tick().await; + let cutoff = now_ms() - REDDIT_CONTENT_TTL.as_millis() as i64; + match projection_store.evict_content_older_than(cutoff) { + Ok(0) => {} + Ok(n) => tracing::info!(evicted = n, "reddit display content TTL eviction"), + Err(e) => tracing::warn!(err = %e, "reddit content TTL eviction failed"), + } + } + }); +} + pub async fn rebuild_projection( cfg: &AppConfig, ) -> Result> { let event_log = EventLog::new(cfg.event_log_path.clone()); let store_path = format!("{}/store", cfg.data_dir); let db = durable::Db::open(std::path::Path::new(&store_path))?; - let entity_store = EntityStore::from_db(&db)?; let projection_store = ProjectionStore::from_db(&db)?; - entity_store.reset()?; projection_store.reset()?; let stats = event_log - .replay(|record| { - projection_apply::apply_records(&projection_store, &entity_store, &[record]) - }) + .replay(|record| projection_apply::apply_records(&projection_store, &[record])) .await?; let cursor = projection_store.last_applied_event_count()?; if cursor != stats.last_seq { @@ -132,7 +143,6 @@ pub struct AppState { pub cfg: Arc, pub event_log: Arc, pub view_log: Arc, - pub entity_store: EntityStore, pub projection_store: ProjectionStore, pub views: ViewStore, journal: JournalClient, @@ -145,7 +155,6 @@ impl AppState { let view_log = Arc::new(ViewLog::new(cfg.views_log_path.clone())); let store_path = format!("{}/store", cfg.data_dir); let db = durable::Db::open(std::path::Path::new(&store_path))?; - let entity_store = EntityStore::from_db(&db)?; let projection_store = ProjectionStore::from_db(&db)?; let views = ViewStore::from_db(&db)?; @@ -157,22 +166,25 @@ impl AppState { } views.spawn_worker(view_log.clone()); - catch_up_projection(&event_log, &entity_store, &projection_store).await?; + catch_up_projection(&event_log, &projection_store).await?; let next_seq = event_log.last_sequence().await? + 1; let journal = JournalClient::spawn( event_log.clone(), - entity_store.clone(), projection_store.clone(), next_seq, ); - let reddit = RedditBroker::spawn(journal.clone(), RedditApiConfig::from_env()); + let reddit = RedditBroker::spawn( + journal.clone(), + projection_store.clone(), + RedditApiConfig::from_env(), + ); + spawn_content_evictor(projection_store.clone()); Ok(Self { cfg: Arc::new(cfg), event_log, view_log, - entity_store, projection_store, views, journal, @@ -248,54 +260,72 @@ impl AppState { mod tests { use super::{normalize_scope, parse_item_param, AppConfig, AppState}; use crate::{ - entity_store::EntityStore, event_log::EventLog, events::Event, path_types::ItemId, - projection_apply, projection_store::ProjectionStore, + event_log::EventLog, events::Event, path_types::ItemId, projection_apply, + projection_store::ProjectionStore, reducer::EntityData, }; - use serde_json::json; fn event_record(seq: u64, event: Event) -> crate::events::EventRecord { crate::events::EventRecord::new(seq, crate::events::event_timestamp(&event), event) } #[tokio::test] - async fn replay_entity_imported_restores_view() { + async fn rebuild_projection_drops_ephemeral_content() { let tmp = tempfile::tempdir().unwrap(); - let log_path = tmp.path().join("events.jsonl"); - let log = EventLog::new(log_path.to_string_lossy().into_owned()); - let payload = json!({"kind":"t5","data":{"title":"Rust","display_name":"rust"}}); - let event = Event::EntityImported { - id: "https://reddit.com/r/rust".into(), - ts: 1, - payload: payload.clone(), - }; - log.append(&event_record(1, event)).await.unwrap(); + let data_dir = tmp.path().to_string_lossy().into_owned(); + let log = EventLog::new(format!("{data_dir}/events.jsonl")); + log.append(&event_record( + 1, + Event::NodeEnsured { + id: "https://reddit.com/r/rust".into(), + }, + )) + .await + .unwrap(); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - super::catch_up_projection(&log, &entity_store, &projection_store) + super::catch_up_projection(&log, &projection_store) .await .unwrap(); - let tree = projection_store - .scope_tree(&ItemId::parse("https://reddit.com/r/rust").unwrap()) - .unwrap(); - let node = tree - .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) - .unwrap(); - assert_eq!(node.data.as_ref().unwrap().title, "Rust"); - let stored = entity_store - .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) - .unwrap() + let id = ItemId::parse("https://reddit.com/r/rust").unwrap(); + projection_store + .put_ephemeral_content( + &id, + &EntityData { + title: "Rust".into(), + author: None, + body_html: None, + thumb_url: None, + image_url: None, + link_url: None, + }, + 1, + ) .unwrap(); - assert_eq!(stored["data"]["display_name"], "rust"); + assert!(projection_store.load_node(&id).unwrap().unwrap().data.is_some()); + drop(projection_store); + drop(db); + + super::rebuild_projection(&AppConfig { + data_dir: data_dir.clone(), + event_log_path: format!("{data_dir}/events.jsonl"), + views_log_path: format!("{data_dir}/views.jsonl"), + port: 0, + }) + .await + .unwrap(); + + let db = durable::Db::open(tmp.path().join("store")).unwrap(); + let projection_store = ProjectionStore::from_db(&db).unwrap(); + let node = projection_store.load_node(&id).unwrap().unwrap(); + assert!(node.data.is_none()); } #[tokio::test] - async fn rebuild_projection_restores_nodes_payloads_and_cursor_from_jsonl() { + async fn rebuild_projection_restores_structure_and_cursor_from_jsonl() { let tmp = tempfile::tempdir().unwrap(); let data_dir = tmp.path().to_string_lossy().into_owned(); let log = EventLog::new(format!("{data_dir}/events.jsonl")); - let payload = json!({"kind":"t5","data":{"title":"Rust","display_name":"rust"}}); log.append_batch(&[ event_record( 1, @@ -305,16 +335,8 @@ mod tests { ), event_record( 2, - Event::EntityImported { - id: "https://reddit.com/r/rust".into(), - ts: 2, - payload: payload.clone(), - }, - ), - event_record( - 3, Event::VoteRecorded { - ts: 3, + ts: 2, a: "alpha".into(), b: "beta".into(), ratio_left: 2, @@ -328,11 +350,9 @@ mod tests { { let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); projection_apply::apply_records( &projection_store, - &entity_store, &[event_record( 1, Event::NodeEnsured { @@ -352,13 +372,12 @@ mod tests { }) .await .unwrap(); - assert_eq!(stats.applied, 3); - assert_eq!(stats.last_seq, 3); + assert_eq!(stats.applied, 2); + assert_eq!(stats.last_seq, 2); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - assert_eq!(projection_store.last_applied_event_count().unwrap(), 3); + assert_eq!(projection_store.last_applied_event_count().unwrap(), 2); let tree = projection_store.scope_tree(&ItemId::root()).unwrap(); let root = tree.get(&ItemId::root()).unwrap(); assert!(root.children.contains(&ItemId::parse("alpha").unwrap())); @@ -366,11 +385,6 @@ mod tests { .load_node(&ItemId::parse("https://reddit.com/r/stale").unwrap()) .unwrap() .is_none()); - let stored = entity_store - .get(&ItemId::parse("https://reddit.com/r/rust").unwrap()) - .unwrap() - .unwrap(); - assert_eq!(stored["data"]["display_name"], "rust"); } #[tokio::test] @@ -389,12 +403,9 @@ mod tests { { let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - // Advance the projection cursor to 2 while the log tail is only 1. projection_apply::apply_records( &projection_store, - &entity_store, &[event_record( 2, Event::NodeEnsured { @@ -403,7 +414,7 @@ mod tests { )], ) .unwrap(); - let err = super::catch_up_projection(&log, &entity_store, &projection_store) + let err = super::catch_up_projection(&log, &projection_store) .await .unwrap_err(); assert!(err @@ -432,10 +443,9 @@ mod tests { .unwrap(); let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - super::catch_up_projection(&log, &entity_store, &projection_store) + super::catch_up_projection(&log, &projection_store) .await .unwrap(); assert_eq!(projection_store.last_applied_event_count().unwrap(), 1); @@ -444,7 +454,7 @@ mod tests { let first_edge_total: f64 = first_root.local_ranking.edges.values().sum(); assert_eq!(first_edge_total, 3.0); - super::catch_up_projection(&log, &entity_store, &projection_store) + super::catch_up_projection(&log, &projection_store) .await .unwrap(); assert_eq!(projection_store.last_applied_event_count().unwrap(), 1); @@ -556,9 +566,8 @@ mod tests { .unwrap(); { let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - super::catch_up_projection(&log, &entity_store, &projection_store) + super::catch_up_projection(&log, &projection_store) .await .unwrap(); } @@ -605,9 +614,8 @@ mod tests { .unwrap(); { let db = durable::Db::open(tmp.path().join("store")).unwrap(); - let entity_store = EntityStore::from_db(&db).unwrap(); let projection_store = ProjectionStore::from_db(&db).unwrap(); - super::catch_up_projection(&log, &entity_store, &projection_store) + super::catch_up_projection(&log, &projection_store) .await .unwrap(); } diff --git a/server/src/storage_dto.rs b/server/src/storage_dto.rs index de5ed995797ae306d3a71394a4d9d245ffd5771d..9dfb13c53efe4389277625a6ab3bfc18f566a453 100644 --- a/server/src/storage_dto.rs +++ b/server/src/storage_dto.rs @@ -3,17 +3,15 @@ //! Node structure (children, edges, voted pairs, recent votes) is no longer a //! single blob — it lives as point-addressable durable collections (see //! [`crate::storage_schema`]). This module only defines the small leaf values: -//! the derived entity view, raw entity payloads, and individual votes. +//! ephemeral entity views and individual votes. use serde::{Deserialize, Serialize}; -use serde_json::Value; use crate::{ path_types::ItemId, reducer::{EntityData, VoteData}, }; -pub const ENTITY_RECORD_VERSION: u32 = 1; pub const VOTE_RECORD_VERSION: u32 = 1; pub const ENTITY_DATA_VERSION: u32 = 1; @@ -29,14 +27,7 @@ impl Versioned { } } -pub type StoredEntityRecord = Versioned; - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct StoredEntityV1 { - pub json: Value, -} - -/// Derived entity view stored at a node's `data` leaf. +/// Derived entity view stored at a node's `data` leaf (ephemeral; not logged). #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StoredEntityDataV1 { pub version: u32, @@ -63,25 +54,6 @@ pub struct StoredVoteV1 { pub thread_tag: String, } -pub fn encode_entity_payload(payload: &Value) -> StoredEntityRecord { - Versioned::new( - ENTITY_RECORD_VERSION, - StoredEntityV1 { - json: payload.clone(), - }, - ) -} - -pub fn decode_entity_payload(record: StoredEntityRecord) -> Result { - if record.version != ENTITY_RECORD_VERSION { - return Err(format!( - "unsupported entity record version: {}", - record.version - )); - } - Ok(record.payload.json) -} - pub fn encode_entity_data(data: &EntityData) -> StoredEntityDataV1 { StoredEntityDataV1 { version: ENTITY_DATA_VERSION, diff --git a/server/src/storage_schema.rs b/server/src/storage_schema.rs index 76bf7bc74a2c5ef4f78678a333b7661778f5b835..bd26e665e084b95b10fdfff091c31e8dc84d07b8 100644 --- a/server/src/storage_schema.rs +++ b/server/src/storage_schema.rs @@ -15,7 +15,7 @@ use crate::{ reducer::{EntityData, GroupState, NodeState, VoteData}, storage_dto::{ decode_entity_data, decode_vote, encode_entity_data, encode_vote, parse_stored_id, - StoredEntityDataV1, StoredEntityRecord, StoredVoteV1, + StoredEntityDataV1, StoredVoteV1, }, }; @@ -40,17 +40,17 @@ pub struct NodeSchema { pub voted_pairs: Map>, /// Recent votes, newest at the front (capped on write). pub recent_votes: Deque>, + /// When ephemeral Reddit display content was last fetched (ms); absent after eviction. + pub fetched_at: Leaf, } -/// The single database root: nodes, raw payloads, view counts, and per-concern -/// metadata maps (cursors and schema versions). +/// The single database root: nodes, view counts, and per-concern metadata maps +/// (cursors and schema versions). #[derive(Durable)] #[allow(dead_code)] pub struct Store { pub nodes: Map, pub proj_meta: Map>, - pub entities: Map>, - pub entity_meta: Map>, pub view_counts: Map>, pub view_meta: Map>, } @@ -264,12 +264,17 @@ pub fn vote_writes( Ok(()) } -/// Reified writes for an imported entity view (node data + path wiring). -pub fn entity_view_writes(batch: &mut Batch, id: &ItemId, view: Option<&EntityData>) { +/// Reified writes for ephemeral Reddit display content (not event-logged). +pub fn entity_content_writes(batch: &mut Batch, id: &ItemId, view: &EntityData, fetched_at: i64) { ensure_path_writes(batch, id); - if let Some(view) = view { - batch.write(node(id).data().set(&encode_entity_data(view))); - } + batch.write(node(id).data().set(&encode_entity_data(view))); + batch.write(node(id).fetched_at().set(&fetched_at)); +} + +/// Clear cached display content for one node (structure/votes are untouched). +pub fn entity_content_clear_writes(batch: &mut Batch, id: &ItemId) { + batch.write(node(id).data().delete()); + batch.write(node(id).fetched_at().delete()); } #[cfg(test)] diff --git a/test/reddit_import.clj b/test/reddit_import.clj index b476488526252c13fd73bdda76e5201678e4a714..6d8c5bca7ebad0b5abfddecd4ea48cc09d738ebe 100644 --- a/test/reddit_import.clj +++ b/test/reddit_import.clj @@ -59,9 +59,10 @@ "curl" "-sf" browse-url)) log (slurp (io/file log-path))] (is (str/includes? after "The Rust Programming Language")) - (is (str/includes? log "\"type\":\"entity_imported\"")) - (is (str/includes? log "\"subscribers\":350000")) - (is (str/includes? log "\"display_name\":\"rust\""))) + (is (str/includes? log "\"type\":\"node_ensured\"")) + (is (not (str/includes? log "\"subscribers\""))) + (is (not (str/includes? log "\"display_name\""))) + (is (not (str/includes? log "entity_imported")))) (let [children-sse (curl-fetch-ui-sse app-base "reddit.com/r/rust" "children")] (is (zero? (:exit children-sse)) "POST /ui fetch_entity (children) SSE succeeds") (is (str/includes? (:out children-sse) "Idiomorph.morph")) @@ -71,10 +72,12 @@ log2 (slurp (io/file log-path))] (is (str/includes? after-children "Announcing Rust 1.99")) (is (str/includes? after-children "Unranked")) - (is (str/includes? log2 "announcing_rust_199"))))) + (is (str/includes? log2 "\"type\":\"node_ensured\"")) + (is (str/includes? log2 "/comments/")) + (is (not (str/includes? log2 "\"selftext\"")))))) (deftest reddit-fetch-via-mock-api - (testing "Fetch more queues import; event log stores full payload; page shows title" + (testing "Fetch caches display content ephemerally; log records structure only" (let [root (repo-root) fixtures (mock-reddit/fixtures-dir root) data-dir (.getAbsolutePath