diff --git a/server/src/api/ui_html.rs b/server/src/api/ui_html.rs
index 7af6527d03c483f33f3469ce6766c01a554c5fe3..d1defd28242fd2ca3b886adc91a7070bef75e653 100644
--- a/server/src/api/ui_html.rs
+++ b/server/src/api/ui_html.rs
@@ -56,12 +56,9 @@ pub async fn post_ui_html(
return ui_js_warn(&e).into_response();
}
let tree = state.tree.read().await;
- let empty = crate::reducer::GroupState::new();
- let group = tree
- .get(&parent)
- .map(|n| &n.local_ranking)
- .unwrap_or(&empty);
- let panel = ranking_panel(&parent, group);
+ let empty = crate::reducer::NodeState::default();
+ let node = tree.get(&parent).unwrap_or(&empty);
+ let panel = ranking_panel(&parent, node);
JsBuilder::new()
.morph_selector("#ranking-panel", panel)
.into_response()
@@ -88,9 +85,9 @@ pub async fn post_ui_html(
.into_response()
}
},
- HtmlUiAction::FetchEntity { item } => {
+ HtmlUiAction::FetchEntity { item, kind } => {
let id = parse_item_param(&item);
- fetch::fetch_entity_stream(state, id).into_response()
+ fetch::fetch_entity_stream(state, id, kind).into_response()
}
}
}
diff --git a/server/src/fetch/html.rs b/server/src/fetch/html.rs
index 63634508496e224c38b9ec0308b7a6086462f925..9b6c2d0964640a157bca2a6a64cb0820238d8e4f 100644
--- a/server/src/fetch/html.rs
+++ b/server/src/fetch/html.rs
@@ -5,7 +5,7 @@ use maud::{html, Markup};
use crate::{
form_template::template_json_compact,
path_types::ItemId,
- reddit::is_fetchable,
+ reddit::{is_children_fetchable, is_fetchable},
reducer::NodeState,
ui_action::UI_RPC_FIELD,
};
@@ -26,25 +26,16 @@ fn entity_panel(node: &NodeState) -> Markup {
}
}
-/// Reddit/API import — `POST /ui` with `fetch_entity` returns an SSE stream.
-pub fn fetch_entity_panel(item: &ItemId, has_data: bool, fetching: bool) -> Markup {
- if !is_fetchable(item) {
- return html! {};
- }
- let label = if fetching {
- "Fetching…"
- } else if has_data {
- "Fetch more"
- } else {
- "Fetch from Reddit"
- };
+/// One `fetch_entity` form/button targeting `kind` ("self" or "children").
+fn fetch_button(item: &ItemId, kind: &str, label: &str, fetching: bool) -> Markup {
let rpc = template_json_compact(&serde_json::json!({
"action": "fetch_entity",
"item": item.as_str(),
+ "kind": kind,
}))
.expect("fetch_entity rpc template");
html! {
- form method="post" action="/ui" id="fetch-entity-form" class="fetch-entity-form" {
+ form method="post" action="/ui" class="fetch-entity-form" {
input type="hidden" name=(UI_RPC_FIELD) value=(rpc);
@if fetching {
button type="submit" class="btn-secondary" disabled { (label) }
@@ -55,6 +46,33 @@ pub fn fetch_entity_panel(item: &ItemId, has_data: bool, fetching: bool) -> Mark
}
}
+/// Reddit/API import controls — `POST /ui` with `fetch_entity` returns an SSE
+/// stream whose events are JS snippets to `eval`.
+pub fn fetch_entity_panel(item: &ItemId, has_data: bool, fetching: bool) -> Markup {
+ let self_ok = is_fetchable(item);
+ let children_ok = is_children_fetchable(item);
+ if !self_ok && !children_ok {
+ return html! {};
+ }
+ let self_label = if fetching {
+ "Fetching…"
+ } else if has_data {
+ "Refresh this"
+ } else {
+ "Fetch from Reddit"
+ };
+ html! {
+ div id="fetch-controls" class="fetch-controls" {
+ @if self_ok {
+ (fetch_button(item, "self", self_label, fetching))
+ }
+ @if children_ok {
+ (fetch_button(item, "children", if fetching { "Fetching…" } else { "Fetch posts" }, fetching))
+ }
+ }
+ }
+}
+
/// Entity card + fetch control (target `#entity-section` for Idiomorph / SSE).
pub fn entity_section(item: &ItemId, node: &NodeState, fetching: bool) -> Markup {
let has_data = node.data.is_some();
diff --git a/server/src/fetch/mod.rs b/server/src/fetch/mod.rs
index 2290f9d3a0f1cbf1806c6339f82a4515c11cc3d3..0177bb161cea1b72a100b52efdfc5e710271c9eb 100644
--- a/server/src/fetch/mod.rs
+++ b/server/src/fetch/mod.rs
@@ -1,4 +1,8 @@
//! Entity import over `POST /ui` as SSE (Reddit worker in [`crate::reddit`]).
+//!
+//! Each SSE event's `data` is a JS snippet that the browser `eval`s — the same
+//! Idiomorph-morph snippets the non-streaming `/ui` responses use. There is no
+//! bespoke JSON envelope; the client just evals whatever each event carries.
pub mod html;
@@ -8,14 +12,15 @@ use std::time::Duration;
use async_stream::stream;
use axum::response::sse::{Event, KeepAlive, Sse};
use futures_util::Stream;
-use serde::Serialize;
use tokio::sync::oneshot;
use crate::{
+ html::{ranking_panel, JsBuilder},
path_types::ItemId,
- reddit::FetchJobResult,
+ reddit::{FetchJobResult, FetchKind},
reducer::NodeState,
state::AppState,
+ ui_action::FetchTarget,
};
pub fn now_ms() -> i64 {
@@ -25,55 +30,62 @@ pub fn now_ms() -> i64 {
t.as_millis() as i64
}
-#[derive(Serialize)]
-struct SseMorphPayload {
- selector: &'static str,
- html: String,
+fn js_event(js: String) -> Event {
+ Event::default().data(js)
}
-fn morph_complete_event(html: maud::Markup) -> Event {
- let payload = SseMorphPayload {
- selector: "#entity-section",
- html: html.into_string(),
- };
- let data = serde_json::to_string(&payload).unwrap_or_else(|_| "{}".into());
- Event::default().event("complete").data(data)
+/// JS that surfaces a transient message in the page's `#errors` region.
+fn error_js(message: &str) -> String {
+ JsBuilder::new()
+ .morph_selector(
+ "#errors",
+ maud::html! { div id="errors" { p class="muted" { (message) } } },
+ )
+ .build()
}
-/// Stream `fetching` → `complete` / `error` for [`crate::ui_action::HtmlUiAction::FetchEntity`].
+/// Stream Idiomorph-morph JS snippets for [`crate::ui_action::HtmlUiAction::FetchEntity`].
pub fn fetch_entity_stream(
state: AppState,
id: ItemId,
+ target: FetchTarget,
) -> Sse>> {
- tracing::debug!(item = %id, "fetch entity stream opened");
+ let kind = match target {
+ FetchTarget::SelfEntity => FetchKind::SelfEntity,
+ FetchTarget::Children => FetchKind::Children,
+ };
+ tracing::debug!(item = %id, ?kind, "fetch entity stream opened");
let stream = stream! {
if id.is_root() {
- yield Ok(Event::default().event("error").data("{\"message\":\"nothing to fetch for the root\"}"));
+ yield Ok(js_event(error_js("Nothing to fetch for the root.")));
return;
}
- if !crate::reddit::is_fetchable(&id) {
- tracing::debug!(item = %id, "fetch stream: not fetchable");
- yield Ok(Event::default().event("error").data("{\"message\":\"this page cannot be fetched from Reddit\"}"));
+ let fetchable = match kind {
+ FetchKind::SelfEntity => crate::reddit::is_fetchable(&id),
+ FetchKind::Children => crate::reddit::is_children_fetchable(&id),
+ };
+ if !fetchable {
+ tracing::debug!(item = %id, ?kind, "fetch stream: not fetchable");
+ yield Ok(js_event(error_js("This page cannot be fetched from Reddit.")));
return;
}
- let fetching_html = {
+ // Optimistic "Fetching…" morph of the entity section.
+ let fetching_js = {
let tree = state.tree.read().await;
let empty = NodeState::default();
let node = tree.get(&id).unwrap_or(&empty);
- html::entity_section(&id, node, true).into_string()
+ JsBuilder::new()
+ .morph_selector("#entity-section", html::entity_section(&id, node, true))
+ .build()
};
- let fetching_payload = serde_json::json!({
- "selector": "#entity-section",
- "html": fetching_html,
- });
- yield Ok(Event::default().event("fetching").data(fetching_payload.to_string()));
+ yield Ok(js_event(fetching_js));
let (tx, rx) = oneshot::channel();
- state.reddit.request_fetch(id.clone(), true, Some(tx));
- tracing::debug!(item = %id, "fetch stream: queued reddit job");
+ state.queue_entity_fetch(id.clone(), kind, Some(tx));
+ tracing::debug!(item = %id, ?kind, "fetch stream: queued reddit job");
let result = match rx.await {
Ok(r) => r,
@@ -82,31 +94,42 @@ pub fn fetch_entity_stream(
FetchJobResult::Failed("reddit worker stopped".into())
}
};
-
tracing::debug!(item = %id, ?result, "fetch stream: job finished");
match result {
- FetchJobResult::Imported | FetchJobResult::NotFound => {
+ FetchJobResult::Imported(_)
+ | FetchJobResult::NotFound
+ | FetchJobResult::SkippedCached
+ | FetchJobResult::SkippedDuplicate => {
let tree = state.tree.read().await;
let empty = NodeState::default();
let node = tree.get(&id).unwrap_or(&empty);
- yield Ok(morph_complete_event(html::entity_section(&id, node, false)));
+ let mut b = JsBuilder::new()
+ .morph_selector("#entity-section", html::entity_section(&id, node, false));
+ if kind == FetchKind::Children {
+ b = b.morph_selector("#ranking-panel", ranking_panel(&id, node));
+ }
+ yield Ok(js_event(b.build()));
}
- FetchJobResult::SkippedCached | FetchJobResult::SkippedDuplicate => {
+ FetchJobResult::RateLimited { reset_secs } => {
let tree = state.tree.read().await;
let empty = NodeState::default();
let node = tree.get(&id).unwrap_or(&empty);
- yield Ok(morph_complete_event(html::entity_section(&id, node, false)));
- }
- FetchJobResult::RateLimited { reset_secs } => {
- yield Ok(Event::default().event("error").data(
- serde_json::json!({"message": format!("Reddit rate limit — retry in {reset_secs}s")}).to_string(),
- ));
+ let js = JsBuilder::new()
+ .morph_selector("#entity-section", html::entity_section(&id, node, false))
+ .raw(&error_js(&format!("Reddit rate limit — retry in {reset_secs}s.")))
+ .build();
+ yield Ok(js_event(js));
}
FetchJobResult::Failed(msg) => {
- yield Ok(Event::default().event("error").data(
- serde_json::json!({"message": msg}).to_string(),
- ));
+ let tree = state.tree.read().await;
+ let empty = NodeState::default();
+ let node = tree.get(&id).unwrap_or(&empty);
+ let js = JsBuilder::new()
+ .morph_selector("#entity-section", html::entity_section(&id, node, false))
+ .raw(&error_js(&format!("Fetch failed: {msg}")))
+ .build();
+ yield Ok(js_event(js));
}
}
};
diff --git a/server/src/html/mod.rs b/server/src/html/mod.rs
index 9314a7556306ddab969b896dbf4126b542a46722..27ce9118c73ec5643e04363ab1a36cf5da6101bf 100644
--- a/server/src/html/mod.rs
+++ b/server/src/html/mod.rs
@@ -6,12 +6,16 @@ use axum::{
};
use maud::{html, Markup, DOCTYPE};
+use std::collections::HashSet;
+
use crate::{
fetch::html::entity_section,
form_template::template_json_compact,
path_types::ItemId,
- ranking::{top_bottom, RankedItem},
- reducer::{GroupState, NodeState},
+ ranking::{
+ connected_components_from_voted_pairs, ranked_items_subset, RankedItem, MAX_ITERS, TOL,
+ },
+ reducer::NodeState,
state::AppState,
ui_action::UI_RPC_FIELD,
};
@@ -178,12 +182,60 @@ fn display_label(id: &ItemId) -> String {
.to_string()
}
-pub fn ranking_panel(item: &ItemId, group: &GroupState) -> Markup {
- let total = group.idx_to_item.len();
- let (top, bottom) = top_bottom(group, 8);
+/// Plain (unscored) list of children that have no votes yet.
+fn unranked_list(label: &str, items: &[ItemId]) -> Markup {
+ html! {
+ @if !items.is_empty() {
+ h3 class="rank-heading muted small" { (label) }
+ ul class="rank-list unranked" {
+ @for it in items {
+ li {
+ a href=(item_href(it)) {
+ strong { (display_label(it)) }
+ }
+ }
+ }
+ }
+ }
+ }
+}
+
+pub fn ranking_panel(item: &ItemId, node: &NodeState) -> Markup {
+ let group = &node.local_ranking;
+ let n = group.idx_to_item.len();
+ let (comps, _isolates) =
+ connected_components_from_voted_pairs(n, group.voted_pairs.iter().copied());
+
+ // Each connected component of voted items is its own ranking; isolated and
+ // never-voted children fall into the "unranked" bucket below.
+ let mut ranked_ids: HashSet = HashSet::new();
+ let mut ranked_groups: Vec> = Vec::new();
+ for comp in &comps {
+ if comp.len() < 2 {
+ continue;
+ }
+ let ranked = ranked_items_subset(group, comp, MAX_ITERS, TOL);
+ for r in &ranked {
+ ranked_ids.insert(r.item.clone());
+ }
+ ranked_groups.push(ranked);
+ }
+ ranked_groups.sort_by(|a, b| b.len().cmp(&a.len()));
+
+ let mut unranked: Vec = node
+ .children
+ .iter()
+ .filter(|c| !ranked_ids.contains(*c))
+ .cloned()
+ .collect();
+ unranked.sort_by(|a, b| a.as_str().cmp(b.as_str()));
+
+ let has_ranked = !ranked_groups.is_empty();
+ let multi = ranked_groups.len() > 1;
+
html! {
section id="ranking-panel" class="demo-panel" {
- @if total == 0 {
+ @if !has_ranked && unranked.is_empty() {
p class="muted" {
@if item.is_root() {
"No votes yet — compare two items below."
@@ -192,11 +244,11 @@ pub fn ranking_panel(item: &ItemId, group: &GroupState) -> Markup {
}
}
} @else {
- (rank_list(if bottom.is_empty() { "" } else { "Top" }, &top, 1))
- @if !bottom.is_empty() {
- p class="rank-gap muted small" { "⋯" }
- (rank_list("Bottom", &bottom, total - bottom.len() + 1))
+ @for (gi, ranked) in ranked_groups.iter().enumerate() {
+ @let label = if multi { format!("Ranking group {}", gi + 1) } else { "Ranking".to_string() };
+ (rank_list(&label, ranked, 1))
}
+ (unranked_list("Unranked", &unranked))
}
}
}
@@ -239,14 +291,13 @@ async fn item_page(state: AppState, uri: Uri, item: ItemId) -> Markup {
let tree = state.tree.read().await;
let empty_node = NodeState::default();
let node = tree.get(&item).unwrap_or(&empty_node);
- let group = &node.local_ranking;
let body = html! {
h1 { "sorter" }
(input_panel("", None))
(breadcrumb_path(&item))
(entity_section(&item, node, false))
- (ranking_panel(&item, group))
+ (ranking_panel(&item, node))
};
layout("sorter2", body, views)
}
diff --git a/server/src/reddit.rs b/server/src/reddit.rs
index 0e6ce32720951a58456852b67fb76a85dd9f4aad..a0eb688709478ee0185b953b41a5d26cc354764d 100644
--- a/server/src/reddit.rs
+++ b/server/src/reddit.rs
@@ -24,7 +24,8 @@ pub fn ensure_partial_tree(tree: &mut GlobalTree, id: &ItemId) {
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FetchJobResult {
- Imported,
+ /// Number of entities written (1 for self, N for children).
+ Imported(usize),
NotFound,
SkippedDuplicate,
SkippedCached,
@@ -32,8 +33,16 @@ pub enum FetchJobResult {
Failed(String),
}
+/// What to import for a node: the node's own entity, or its child listing.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum FetchKind {
+ SelfEntity,
+ Children,
+}
+
pub struct RedditCommand {
pub id: ItemId,
+ pub kind: FetchKind,
/// User-initiated fetch bypasses the in-memory "recently fetched" cache.
pub force: bool,
pub done: Option>,
@@ -52,8 +61,12 @@ struct RedditCredentials {
#[derive(Clone)]
pub struct RedditApiConfig {
+ /// Unauthenticated `.json` requests (public).
pub api_base: String,
- pub oauth_base: String,
+ /// `POST …/api/v1/access_token` (always www.reddit.com in production).
+ pub oauth_token_base: String,
+ /// Bearer-authenticated API (`GET` subreddit about, etc.).
+ pub oauth_api_base: String,
pub user_agent: String,
creds: Option,
}
@@ -85,7 +98,8 @@ impl RedditBroker {
tracing::debug!(
api_base = %config.api_base,
- oauth_base = %config.oauth_base,
+ oauth_token_base = %config.oauth_token_base,
+ oauth_api_base = %config.oauth_api_base,
oauth = config.creds.is_some(),
"reddit worker started"
);
@@ -99,11 +113,12 @@ impl RedditBroker {
pub fn request_fetch(
&self,
id: ItemId,
+ kind: FetchKind,
force: bool,
done: Option>,
) {
- match self.tx.try_send(RedditCommand { id: id.clone(), force, done }) {
- Ok(()) => tracing::debug!(item = %id, force, "reddit fetch queued"),
+ match self.tx.try_send(RedditCommand { id: id.clone(), kind, force, done }) {
+ Ok(()) => tracing::debug!(item = %id, ?kind, force, "reddit fetch queued"),
Err(_) => tracing::warn!(item = %id, "reddit fetch queue full, dropped"),
}
}
@@ -113,7 +128,8 @@ impl RedditApiConfig {
pub fn from_env() -> Self {
Self {
api_base: reddit_api_base(),
- oauth_base: reddit_oauth_base(),
+ oauth_token_base: reddit_oauth_token_base(),
+ oauth_api_base: reddit_oauth_api_base(),
user_agent: default_user_agent(),
creds: RedditCredentials::from_env(),
}
@@ -142,8 +158,16 @@ pub fn reddit_api_base() -> String {
std::env::var("REDDIT_API_BASE").unwrap_or_else(|_| "https://www.reddit.com".into())
}
-pub fn reddit_oauth_base() -> String {
- std::env::var("REDDIT_OAUTH_BASE").unwrap_or_else(|_| "https://www.reddit.com".into())
+/// Where to `POST /api/v1/access_token` (not the bearer API host).
+pub fn reddit_oauth_token_base() -> String {
+ std::env::var("REDDIT_OAUTH_TOKEN_BASE")
+ .or_else(|_| std::env::var("REDDIT_OAUTH_BASE"))
+ .unwrap_or_else(|_| "https://www.reddit.com".into())
+}
+
+/// Host for OAuth-authenticated API `GET`s (must not be www.reddit.com).
+pub fn reddit_oauth_api_base() -> String {
+ std::env::var("REDDIT_OAUTH_API_BASE").unwrap_or_else(|_| "https://oauth.reddit.com".into())
}
pub fn default_user_agent() -> String {
@@ -152,10 +176,16 @@ pub fn default_user_agent() -> String {
})
}
+/// Can the node's own entity be imported (subreddit `about` or a post)?
pub fn is_fetchable(id: &ItemId) -> bool {
!map_item_to_reddit_api(id, "https://example.com").is_empty()
}
+/// Can we import this node's children (currently: a subreddit's posts)?
+pub fn is_children_fetchable(id: &ItemId) -> bool {
+ !map_children_url(id, "https://example.com").is_empty()
+}
+
pub fn entity_view_from_payload(id: &ItemId, payload: &Value) -> Option {
if id.as_str().starts_with("reddit.com") {
return parse_reddit_view(id, payload);
@@ -181,92 +211,116 @@ async fn reddit_worker(
client: Client,
config: RedditApiConfig,
) {
- let mut in_flight = HashSet::new();
- let mut recently_fetched: HashMap = HashMap::new();
+ let mut in_flight: HashSet<(ItemId, FetchKind)> = HashSet::new();
+ let mut recently_fetched: HashMap<(ItemId, FetchKind), Instant> = HashMap::new();
let mut current_delay = Duration::from_secs(1);
let mut oauth: Option = None;
let cache_ttl = Duration::from_secs(300);
let creds = config.creds.clone();
let api_base = config.api_base.clone();
- let oauth_base = config.oauth_base.clone();
+ let oauth_token_base = config.oauth_token_base.clone();
+ let oauth_api_base = config.oauth_api_base.clone();
while let Some(cmd) = rx.recv().await {
let now = Instant::now();
recently_fetched.retain(|_, t| now.duration_since(*t) < cache_ttl);
- if in_flight.contains(&cmd.id) {
- tracing::debug!(item = %cmd.id, "reddit fetch skipped: already in flight");
+ let kind = cmd.kind;
+ let key = (cmd.id.clone(), kind);
+
+ if in_flight.contains(&key) {
+ tracing::debug!(item = %cmd.id, ?kind, "reddit fetch skipped: already in flight");
notify(cmd.done, FetchJobResult::SkippedDuplicate);
continue;
}
- if !cmd.force && recently_fetched.contains_key(&cmd.id) {
- tracing::debug!(item = %cmd.id, "reddit fetch skipped: recently fetched cache");
+ if !cmd.force && recently_fetched.contains_key(&key) {
+ tracing::debug!(item = %cmd.id, ?kind, "reddit fetch skipped: recently fetched cache");
notify(cmd.done, FetchJobResult::SkippedCached);
continue;
}
- in_flight.insert(cmd.id.clone());
+ in_flight.insert(key.clone());
let fetch_id = cmd.id.clone();
let done = cmd.done;
tracing::debug!(
item = %fetch_id,
+ ?kind,
delay_ms = current_delay.as_millis(),
"reddit fetch starting after delay"
);
tokio::time::sleep(current_delay).await;
if let Some(c) = &creds {
- oauth = ensure_oauth_token(&client, &oauth_base, c, oauth.take()).await;
+ oauth = ensure_oauth_token(&client, &oauth_token_base, c, oauth.take()).await;
}
let token = oauth.as_ref().map(|t| t.access_token.as_str());
- if token.is_some() {
- tracing::debug!(item = %fetch_id, "reddit fetch using OAuth bearer");
- }
-
let fetch_base = if token.is_some() {
- &oauth_base
+ tracing::debug!(
+ item = %fetch_id,
+ base = %oauth_api_base,
+ "reddit fetch using OAuth bearer"
+ );
+ &oauth_api_base
} else {
&api_base
};
- let outcome = do_fetch(&client, fetch_base, &fetch_id, token).await;
+ let url = match kind {
+ FetchKind::SelfEntity => map_item_to_reddit_api(&fetch_id, fetch_base),
+ FetchKind::Children => map_children_url(&fetch_id, fetch_base),
+ };
+ let outcome = do_fetch(&client, &url, &fetch_id, token).await;
match outcome {
Ok(FetchOutcome::Payload(payload)) => {
- let ts = now_ms();
- let event = Event::EntityImported {
- id: fetch_id.as_str().to_string(),
- ts,
- payload: payload.clone(),
+ 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,
- ts,
- payload_keys = ?payload.as_object().map(|o| o.len()),
- "reddit fetch got JSON payload, appending event"
+ ?kind,
+ count = imports.len(),
+ "reddit fetch got payload, importing"
);
- match event_log.append(&event).await {
- Err(e) => {
- tracing::warn!(
- item = %fetch_id,
- err = %e,
- "reddit event log append failed"
- );
- notify(done, FetchJobResult::Failed(e.to_string()));
+
+ let mut write_err: Option = None;
+ let mut written = 0usize;
+ for (child_id, child_payload) in imports {
+ let event = Event::EntityImported {
+ id: child_id.as_str().to_string(),
+ ts: now_ms(),
+ payload: child_payload.clone(),
+ };
+ if let Err(e) = event_log.append(&event).await {
+ tracing::warn!(item = %child_id, err = %e, "reddit event log append failed");
+ write_err = Some(e.to_string());
+ break;
+ }
+ {
+ let mut tree = tree.write().await;
+ apply_entity_import(&mut tree, &child_id, child_payload);
+ if kind == FetchKind::Children {
+ tree.link_child(&fetch_id, &child_id);
+ }
}
- Ok(()) => {
- apply_entity_import(&mut *tree.write().await, &fetch_id, payload);
- recently_fetched.insert(fetch_id.clone(), Instant::now());
+ written += 1;
+ }
+
+ match write_err {
+ Some(e) => notify(done, FetchJobResult::Failed(e)),
+ None => {
+ recently_fetched.insert(key.clone(), Instant::now());
current_delay = Duration::from_millis(600);
- tracing::info!(item = %fetch_id, "reddit entity imported");
- notify(done, FetchJobResult::Imported);
+ tracing::info!(item = %fetch_id, ?kind, written, "reddit import complete");
+ notify(done, FetchJobResult::Imported(written));
}
}
}
Ok(FetchOutcome::NotFound) => {
tracing::debug!(item = %fetch_id, "reddit fetch: not found (no event written)");
- recently_fetched.insert(fetch_id.clone(), Instant::now());
+ recently_fetched.insert(key.clone(), Instant::now());
notify(done, FetchJobResult::NotFound);
}
Ok(FetchOutcome::RateLimited { reset_secs }) => {
@@ -286,7 +340,7 @@ async fn reddit_worker(
}
}
- in_flight.remove(&fetch_id);
+ in_flight.remove(&key);
}
}
@@ -358,11 +412,10 @@ async fn ensure_oauth_token(
async fn do_fetch(
client: &Client,
- api_base: &str,
+ url: &str,
id: &ItemId,
bearer: Option<&str>,
) -> Result {
- let url = map_item_to_reddit_api(id, api_base);
if url.is_empty() {
tracing::debug!(item = %id, "reddit do_fetch: no API URL for item");
return Ok(FetchOutcome::NotFound);
@@ -370,7 +423,7 @@ async fn do_fetch(
tracing::debug!(item = %id, %url, bearer = bearer.is_some(), "reddit HTTP GET");
- let mut req = client.get(&url);
+ let mut req = client.get(url);
if let Some(token) = bearer {
req = req.bearer_auth(token);
}
@@ -408,6 +461,9 @@ async fn do_fetch(
body_prefix = %body.chars().take(240).collect::(),
"reddit non-success body"
);
+ if status == StatusCode::FORBIDDEN || status == StatusCode::UNAUTHORIZED {
+ return Err(format!("Reddit API {status}: {body}"));
+ }
return Ok(FetchOutcome::NotFound);
}
@@ -474,6 +530,43 @@ pub fn map_item_to_reddit_api(id: &ItemId, api_base: &str) -> String {
String::new()
}
+/// Listing URL for a node's children. Currently only subreddits
+/// (`reddit.com/r/` → `/r/.json`) expose a child listing.
+pub fn map_children_url(id: &ItemId, api_base: &str) -> String {
+ let path = id.as_str();
+ if !path.starts_with("reddit.com/") {
+ return String::new();
+ }
+ let base = api_base.trim_end_matches('/');
+ let segments: Vec<&str> = path.split('/').collect();
+ if segments.len() == 3 && segments[1] == "r" {
+ return format!("{base}/r/{}.json?raw_json=1&limit=25", segments[2]);
+ }
+ String::new()
+}
+
+/// Parse a subreddit listing payload into `(child_id, child_payload)` entries.
+/// Each child id is the post's permalink under `reddit.com/…`, and the payload
+/// is the raw `{kind, data}` listing element (persisted per child).
+fn parse_children(_parent: &ItemId, payload: &Value) -> Vec<(ItemId, Value)> {
+ let mut out = Vec::new();
+ let children = match payload.pointer("/data/children").and_then(|c| c.as_array()) {
+ Some(c) => c,
+ None => return out,
+ };
+ for child in children {
+ let permalink = match child.pointer("/data/permalink").and_then(|p| p.as_str()) {
+ Some(p) if !p.is_empty() => p,
+ _ => continue,
+ };
+ let path = format!("reddit.com{}", permalink.trim_end_matches('/'));
+ if let Some(id) = ItemId::parse(&path) {
+ out.push((id, child.clone()));
+ }
+ }
+ out
+}
+
fn parse_reddit_view(id: &ItemId, v: &Value) -> Option {
let segments: Vec<&str> = id.as_str().split('/').collect();
@@ -512,8 +605,13 @@ fn parse_subreddit_about(v: &Value) -> Option {
}
fn parse_post_listing(v: &Value) -> Option {
- let listing = v.as_array()?.first()?;
- let child = listing.pointer("/data/children/0/data")?;
+ // Two shapes: a comments-page array `[listing, comments]`, or a single
+ // listing element `{kind, data}` (from a subreddit children import).
+ let child = if let Some(arr) = v.as_array() {
+ arr.first()?.pointer("/data/children/0/data")?
+ } else {
+ v.get("data")?
+ };
let title = child.get("title")?.as_str()?.to_string();
let author = child
.get("author")
@@ -550,6 +648,10 @@ mod tests {
map_item_to_reddit_api(&id, "https://www.reddit.com"),
"https://www.reddit.com/r/rust/about.json?raw_json=1"
);
+ assert_eq!(
+ map_item_to_reddit_api(&id, "https://oauth.reddit.com"),
+ "https://oauth.reddit.com/r/rust/about.json?raw_json=1"
+ );
}
#[test]
diff --git a/server/src/reducer.rs b/server/src/reducer.rs
index 60db81a562e3590775012573225a76ba82a14563..a36cd5c9287d61536f9f4a3f6b0df2342388857a 100644
--- a/server/src/reducer.rs
+++ b/server/src/reducer.rs
@@ -208,6 +208,17 @@ impl GlobalTree {
node.data = view;
}
}
+
+ /// Directly attach `child` under `parent`, bypassing path-based nesting.
+ /// Used for imported listings (e.g. a subreddit's posts) so they show up
+ /// as children of the subreddit rather than a deep `…/comments/` path.
+ pub fn link_child(&mut self, parent: &ItemId, child: &ItemId) {
+ self.ensure_path(parent);
+ self.ensure_path(child);
+ if let Some(p) = self.nodes.get_mut(parent) {
+ p.children.insert(child.clone());
+ }
+ }
}
#[cfg(test)]
diff --git a/server/src/state.rs b/server/src/state.rs
index e7ff9f663e45b5e168d6a3869948e3bd890966d4..d238701208a1c708b94a778a6a2e1891a678ecbc 100644
--- a/server/src/state.rs
+++ b/server/src/state.rs
@@ -144,8 +144,13 @@ impl AppState {
}
/// User-initiated Reddit/API import (SSE / fetch module only).
- pub fn queue_entity_fetch(&self, id: ItemId, done: Option>) {
- self.reddit.request_fetch(id, true, done);
+ pub fn queue_entity_fetch(
+ &self,
+ id: ItemId,
+ kind: crate::reddit::FetchKind,
+ done: Option>,
+ ) {
+ self.reddit.request_fetch(id, kind, true, done);
}
pub async fn record_vote(
diff --git a/server/src/ui_action.rs b/server/src/ui_action.rs
index ef9ac873fc0442e0b960f144309c8e91124d7884..2047713762c932ac9bc325fe15624f2aecb3364d 100644
--- a/server/src/ui_action.rs
+++ b/server/src/ui_action.rs
@@ -8,6 +8,16 @@ use thiserror::Error;
pub const UI_RPC_FIELD: &str = "__rpc__";
+/// What a `fetch_entity` action targets: the node itself, or its children.
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
+pub enum FetchTarget {
+ #[default]
+ #[serde(rename = "self")]
+ SelfEntity,
+ #[serde(rename = "children")]
+ Children,
+}
+
/// HTML form / fetch `POST /ui` payload after template fill and deserialization.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "action", rename_all = "snake_case")]
@@ -26,9 +36,12 @@ pub enum HtmlUiAction {
ParseQuery {
query: String,
},
- /// Import entity data; `POST /ui` responds with `text/event-stream` (not JS).
+ /// Import entity data; `POST /ui` responds with `text/event-stream` whose
+ /// events carry JS snippets to `eval` (Idiomorph morphs), not JSON.
FetchEntity {
item: String,
+ #[serde(default)]
+ kind: FetchTarget,
},
}
diff --git a/server/static/sorter_ui.js b/server/static/sorter_ui.js
index fe7bfc657c0fccae5c596d005b0e807abefec677..62c9158ecb3c2140b83efd2b27ea706a9e0ae127 100644
--- a/server/static/sorter_ui.js
+++ b/server/static/sorter_ui.js
@@ -8,48 +8,24 @@
}
}
- function morphSelector(selector, html) {
- var el = document.querySelector(selector);
- if (el && typeof Idiomorph !== 'undefined') {
- Idiomorph.morph(el, html);
- }
- }
-
- function handleSseEvent(eventType, data, form) {
- if (eventType === 'fetching' || eventType === 'complete') {
- try {
- var msg = JSON.parse(data);
- morphSelector(msg.selector || '#entity-section', msg.html);
- } catch (err) {
- console.warn('fetch morph parse', err);
- }
- }
- if (eventType === 'complete' || eventType === 'error') {
- var btn = form && form.querySelector('button[type="submit"]');
- if (btn) btn.disabled = false;
- }
- if (eventType === 'error') {
- try {
- var err = JSON.parse(data);
- console.warn('fetch error:', err.message || data);
- } catch (_e) {
- console.warn('fetch error:', data);
- }
- }
- }
-
- function consumeSseStream(response, form) {
+ // Each SSE event's `data` is a JS snippet to eval (same as the non-stream
+ // /ui responses). Parse the raw event stream, joining multi-line `data:`
+ // fields, and eval each event as it arrives.
+ function consumeSseStream(response) {
var reader = response.body.getReader();
var decoder = new TextDecoder();
var buffer = '';
- var eventType = '';
var dataLines = [];
function dispatch() {
- if (!eventType && dataLines.length === 0) return;
- handleSseEvent(eventType || 'message', dataLines.join('\n'), form);
- eventType = '';
+ if (dataLines.length === 0) return;
+ var js = dataLines.join('\n');
dataLines = [];
+ try {
+ evalJs(js);
+ } catch (err) {
+ console.warn('fetch eval failed', err);
+ }
}
function pump() {
@@ -65,11 +41,10 @@
var line = parts[i].replace(/\r$/, '');
if (line === '') {
dispatch();
- } else if (line.indexOf('event:') === 0) {
- eventType = line.slice(6).trim();
} else if (line.indexOf('data:') === 0) {
- dataLines.push(line.slice(5).trim());
+ dataLines.push(line.slice(5).replace(/^ /, ''));
}
+ // `event:`/`id:`/`:` comment lines are ignored — data carries the JS.
}
return pump();
});
@@ -78,9 +53,13 @@
return pump();
}
+ function isFetchForm(form) {
+ return form.classList && form.classList.contains('fetch-entity-form');
+ }
+
function postUiForm(form) {
var btn = form.querySelector('button[type="submit"]');
- if (form.id === 'fetch-entity-form' && btn) {
+ if (isFetchForm(form) && btn) {
btn.disabled = true;
}
return fetch(form.action, {
@@ -91,11 +70,11 @@
}).then(function (resp) {
var ct = resp.headers.get('content-type') || '';
if (ct.indexOf('text/event-stream') !== -1) {
- return consumeSseStream(resp, form);
+ return consumeSseStream(resp);
}
return resp.text().then(evalJs);
}).catch(function (err) {
- if (form.id === 'fetch-entity-form' && btn) {
+ if (isFetchForm(form) && btn) {
btn.disabled = false;
}
console.warn('POST /ui failed', err);
diff --git a/test/fixtures/reddit/r_rust_listing.json b/test/fixtures/reddit/r_rust_listing.json
new file mode 100644
index 0000000000000000000000000000000000000000..c478abe0331123f1f8922dd7269addcf0c302dc6
--- /dev/null
+++ b/test/fixtures/reddit/r_rust_listing.json
@@ -0,0 +1,29 @@
+{
+ "kind": "Listing",
+ "data": {
+ "after": "t3_bbb",
+ "children": [
+ {
+ "kind": "t3",
+ "data": {
+ "title": "Announcing Rust 1.99",
+ "permalink": "/r/rust/comments/aaa/announcing_rust_199/",
+ "author": "alice",
+ "selftext_html": "release notes
",
+ "thumbnail": "self",
+ "subreddit": "rust"
+ }
+ },
+ {
+ "kind": "t3",
+ "data": {
+ "title": "What are you working on this week?",
+ "permalink": "/r/rust/comments/bbb/what_are_you_working_on/",
+ "author": "bob",
+ "thumbnail": "default",
+ "subreddit": "rust"
+ }
+ }
+ ]
+ }
+}
diff --git a/test/reddit_import.clj b/test/reddit_import.clj
index 17f2ee39e1498b0b6bb6a9d8a7e35608b0bd1b32..84cbdcf7965e50290313cbce2243a16b583d2097 100644
--- a/test/reddit_import.clj
+++ b/test/reddit_import.clj
@@ -14,16 +14,22 @@
(.getLocalPort s)))
(defn- start-mock-reddit [port fixtures-dir]
- (let [fixture (io/file fixtures-dir "r_rust_about.json")
- body (.getBytes (slurp fixture) "UTF-8")
+ (let [about (.getBytes (slurp (io/file fixtures-dir "r_rust_about.json")) "UTF-8")
+ listing (.getBytes (slurp (io/file fixtures-dir "r_rust_listing.json")) "UTF-8")
server (HttpServer/create (InetSocketAddress. "127.0.0.1" port) 0)
handler
(proxy [HttpHandler] []
(handle [^HttpExchange exchange]
- (.sendResponseHeaders exchange 200 (alength body))
- (let [out (.getResponseBody exchange)]
- (.write out body)
- (.close out))))]
+ ;; Route by path: `/r//about.json` is the subreddit entity,
+ ;; `/r/.json` is the children listing.
+ (let [path (.getPath (.getRequestURI exchange))
+ body (if (str/includes? path "/about")
+ about
+ listing)]
+ (.sendResponseHeaders exchange 200 (alength body))
+ (let [out (.getResponseBody exchange)]
+ (.write out body)
+ (.close out)))))]
(.createContext server "/" handler)
(.setExecutor server nil)
(.start server)
@@ -44,12 +50,13 @@
(do (Thread/sleep 200) (recur))
false))))))
-(defn- curl-fetch-ui-sse [base item]
+(defn- curl-fetch-ui-sse [base item kind]
(process/shell {:out :string :err :string}
"curl" "-sfN" "--max-time" "20"
"-X" "POST" (str base "/ui")
"--data-urlencode"
- (str "__rpc__={\"action\":\"fetch_entity\",\"item\":\"" item "\"}")))
+ (str "__rpc__={\"action\":\"fetch_entity\",\"item\":\"" item
+ "\",\"kind\":\"" kind "\"}")))
(defn- wait-event-log [path ms]
(let [deadline (+ (System/currentTimeMillis) ms)]