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: [1c914c6e] stage set Side A — unified diff (full patch): diff --git a/plan.md b/plan.md new file mode 100644 index 0000000000000000000000000000000000000000..00d6867a1e0ed144a16a020ea037f685ce646c73 --- /dev/null +++ b/plan.md @@ -0,0 +1,155 @@ +# Plan: `ItemId` + `RouteContext` (identity vs hrefs) + +This document is for **the next agent** to continue the refactor without re-deriving context from chat. It supersedes ad-hoc notes: treat it as the checklist of record until the work lands and this file is deleted or trimmed. + +## Goal + +- **Identity** (what lives in the reducer graph, votes, indexes) becomes a **structural `ItemId` enum** in `slug-types`, not a canonical `String` / `CanonicalItemUrl` newtype. +- **Presentation** (tilde / dash display, breadcrumbs) derives from `ItemId` via explicit methods, not string stripping. +- **Routing** (browser `href`s for public vs room) goes through **`RouteContext`** (started in `server/src/html/routing.rs`) so Maud/handlers do not stitch `/r/…` vs `/~` ad hoc. + +**Non-goals for v1 of the migration:** backward-compatible JSONL or dual-read of old canonical strings in the event log (project has accepted breaking changes). If you reintroduce compat, document it here. + +## Current state (as of this plan) + +- **`CanonicalItemUrl`** (`types/src/paths.rs`): newtype around `String`; `parse` / `parent` / `display_path` / `tilde_tail` / etc. Reducer `ContentState`, `VoteData`, ranking, RPC, search, garden, breadcrumbs all use it or `String` keys derived from it. +- **`ThreadNav`** (`server/src/html/forum/nav.rs`): encodes scope prefixes for threads and garden URLs; **`RouteContext`** now wraps `ThreadNav` (`server/src/html/routing.rs`, re-exported from `server/src/html/mod.rs`) but **most HTML still takes `&ThreadNav` directly** — migration incomplete. +- **URL normalization** lives in `types/src/url_normalize.rs` + `canonicalize_item` / `finalize_external_identity_url` in `paths.rs` (YouTube, sorted query params, room path `room_route_segment` in `paths.rs`). +- **Room HTTP paths** are `/r/{short}{slug}` (fused segment); wire **`room_id`** remains `short/slug` for RPC/events. + +## Target architecture + +### `ItemId` (types) + +Suggested shape (adjust after profiling `Ord` / `Hash` / serde size): + +```text +ItemId::Root — tilde ontology root (today `SLUG_TILDE_ONTOLOGY_ROOT`) +ItemId::Local { segments } — slug.social ~/… path as Vec (lowercase segments, non-empty for non-root) +ItemId::External { url: Url } — normalized `url::Url` (crate `url` already in `slug-types`) +``` + +**API surface (minimum):** + +- `ItemId::parse(&str) -> Option` — single entry from DSL / user input / legacy wire (internally may call `canonicalize_item` + structured split). +- `ItemId::to_wire_url(&self) -> String` — only for **external** boundaries if needed (HTTP fetch, rare assertions); avoid using as the primary key once maps use `ItemId`. +- `parent`, `display_path`, `tilde_tail` / `tilde_http_tail`, `tilde_segments`, `last_segment`, `normalized_storage` — port from `CanonicalItemUrl`. +- **`Ord` + `Hash` + `Eq`** stable for `BTreeSet` / `HashMap` (see `write_actor` scope-rank snapshots). +- **`Serialize` / `Deserialize`** — decide **tagged JSON** for any persisted or API-carried structs (e.g. `VoteData` in tests). If RPC must stay stringy for clients, use a **DTO layer** that converts `ItemId` ↔ wire at the boundary only. + +**Remove:** `CanonicalItemUrl` type and all `path_types::CanonicalItemUrl` / `slug_types::paths::CanonicalItemUrl` exports once call sites are migrated. **`Borrow`** on the old newtype goes away; update `nav!` / any code that assumed map keys borrowed as `str`. + +### `RouteContext` (server HTML) + +- **File:** `server/src/html/routing.rs` — **`RouteContext(ThreadNav)`** with `item_href`, `item_href_raw`, `thread_url`, `garden_root_url`, `room_url`, `From`/`Into` `ThreadNav`. +- **Direction:** new code and refactored Maud should take **`&RouteContext`** (or owned where appropriate) instead of `&ThreadNav` when building links. Long term, **`item_href(&ItemId)`** should not parse strings — it should pattern-match `ItemId` and append tilde tail or `/-/…` external tail using the same rules as today’s `ThreadNav::garden_item_url`. + +### Axum / garden routes + +- **No** single catch-all route (explicit decision): keep the existing router layout in `server/src/lib.rs`. +- Room routes stay **`/r/:room_key/...`** with `room_key` fused; parsing via `slug_types::room_id_from_route_segment` / `room_route_segment` in `paths.rs`. + +## Phased execution (recommended order) + +### Phase 0 — Preconditions (quick) + +1. Read **`AGENTS.md`** (UI contract, durability matrix, `RpcCommand` vs `HtmlUiAction`). +2. Run **`cargo test --workspace`** and **`./scripts/clj-test.sh`** on clean `main` before large diffs; repeat after each phase. + +### Phase 1 — `ItemId` in `slug-types` (no server yet) + +1. Add **`ItemId`** (new file e.g. `types/src/item_id.rs` **or** inline at bottom of `paths.rs` — see **Module cycle** below). +2. Implement **`ItemId::parse`** using existing **`canonicalize_item`** + normalization; port **`CanonicalItemUrl`** methods to **`ItemId`** with tests ported from `paths.rs` `#[cfg(test)] mod tests`. +3. **`GardenItemUrl::from_stored(&ItemId, room_wire)`** (and thread helpers) — build absolute hrefs from structure, not from re-parsing a canonical string. +4. **`TildeHttpPathTail::to_item_id`** (rename from `to_canonical`) / **`tilde_http_path_to_item_id`**. +5. **`TildeOntologyPath::from_stored(&ItemId)`**. +6. Export **`ItemId`** from **`types/src/lib.rs`**; update **`server/src/path_types.rs`** re-exports. +7. **Delete `CanonicalItemUrl`** and fix all **in-crate** references in `types` only until `cargo test` passes for `slug-types`. + +**Module cycle trap:** `item_id.rs` must not `use crate::paths::{...}` if `paths.rs` also imports `ItemId` for `GardenItemUrl` in the same module. **Fix one of:** + +- **A)** Put `ItemId` **inside `paths.rs`** below `canonicalize_item` / helpers (simplest, large file), or +- **B)** Split **`canonicalize_item`** (+ dash host helpers + `finalize_external_identity_url`) into **`types/src/item_wire.rs`**, then `paths.rs` + `item_id.rs` both depend on `item_wire` only (cleaner, more files). + +### Phase 2 — Reducer + ranking (server core) + +1. **`server/src/reducer.rs`**: `ContentState` / `GroupState` / **`VoteData`** — replace **`CanonicalItemUrl`** with **`ItemId`** on all maps, sets, deques, vectors. +2. **`apply_vote`**: normalize `a`/`b` via **`ItemId::parse`** or **`ItemId`**-aware logic (remove string round-trip). +3. **`apply_ingest_to_content`**: **`dsl`** still yields strings for item titles in statements; normalize to **`ItemId`** at ingest boundary via **`ItemId::parse`** once per item. +4. **`server/src/ranking.rs`**, **`server/src/scope_rank.rs`**, **`server/src/api/write_actor.rs`** (including **`BTreeSet`** ordering), **`server/src/api/validate.rs`**, **`server/src/api/helpers.rs`** — propagate **`ItemId`**. +5. **`server/tests/basic.rs`** and any reducer tests constructing **`VoteData`** — use **`ItemId::parse(...).unwrap()`** or helpers. + +### Phase 3 — RPC + search + external resolver + +1. **`server/src/api/rpc.rs`**: rank/pair/matchup/search payloads; today many paths use **`GardenItemUrl::from_storage_str(item.as_str(), …)`** — switch to **`ItemId`** + **`GardenItemUrl::from_stored(&item_id, …)`** (or equivalent). +2. **`server/src/html/search.rs`**: scoring uses item path strings — derive from **`ItemId::display_path`** / **`to_wire_url`** only at the scoring boundary if needed. +3. **`server/src/external_resolver.rs`**: take **`&ItemId`** or **`ItemId::external_url()`** instead of **`&CanonicalItemUrl`**. + +### Phase 4 — HTML / Maud + +1. **`ThreadNav::garden_item_url`**: overload or replace with **`garden_item_href(&self, item: &ItemId)`** (no `CanonicalItemUrl::parse` inside). +2. **`RouteContext`**: extend **`item_href(&ItemId)`**; migrate call sites from **`ThreadNav`** to **`RouteContext`** where only link-building is needed (keep **`ThreadNav`** where scope / auth helpers need the full struct). +3. **`server/src/html/garden.rs`**, **`breadcrumb_path.rs`**, **`forum/*`**, **`editor.rs`**: replace **`CanonicalItemUrl`** with **`ItemId`**; breadcrumbs should walk **`ItemId::parent`** without string `rsplit`. +4. **`types` JSON types** (`RankRow`, etc.): decide whether **`GardenItemUrl`** stays string for JSON or becomes a structured field; keep **one** wire format for the public API. + +### Phase 5 — Cleanup + docs + +1. Remove dead **`canonical_path`** / **`breadcrumb_path`** string logic if fully superseded. +2. Update **`AGENTS.md`** if durability, `POST /ui`, or command surfaces change. +3. Delete or shrink **`plan.md`** when done. + +## File / symbol checklist (non-exhaustive — grep-driven) + +Run periodically: + +```bash +rg "CanonicalItemUrl" -g'*.rs' +rg "path_types::CanonicalItemUrl" -g'*.rs' +rg "tilde_http_path_to_canonical" -g'*.rs' +``` + +**High-touch files (from prior exploration):** + +| Area | Files | +|------|--------| +| Types | `types/src/paths.rs`, `types/src/lib.rs`, `types/src/url_normalize.rs`, (optional) `types/src/item_id.rs`, `types/src/item_wire.rs` | +| Server re-exports | `server/src/path_types.rs`, `server/src/canonical_path.rs` | +| Reducer / ingest | `server/src/reducer.rs`, `server/src/dsl.rs` (parse output types if changed) | +| Ranking | `server/src/ranking.rs`, `server/src/scope_rank.rs` | +| Writer / RPC | `server/src/api/write_actor.rs`, `server/src/api/rpc.rs`, `server/src/api/helpers.rs`, `server/src/api/validate.rs` | +| HTML | `server/src/html/garden.rs`, `server/src/html/breadcrumb_path.rs`, `server/src/html/forum/nav.rs`, `server/src/html/routing.rs`, `server/src/html/search.rs`, `server/src/html/editor.rs`, `server/src/html/forum/ingest.rs`, … | +| Tests | `server/tests/basic.rs`, `server/tests/integration.rs`, `types/src/paths.rs` tests, Clojure under `test/` if URLs/assertions mention canonical shapes | + +## Events / JSONL + +- **`Ingest`** events store **`raw` DSL** only — no change required for item identity inside the event. +- If any future event type stores item ids as strings, migrate to **structured `ItemId` serde** or accept string only at the event boundary with immediate parse into **`ItemId`** on `apply_event`. + +## `nav!` macro (`server/src/paths.rs`) + +- Macros use **`keypath($key)`** with **`.clone()`** — **`ItemId`** must be **`Clone`** (already for enums). Remove any reliance on **`Borrow`** for map keys. + +## Testing gate + +After each phase: + +```bash +cargo test --workspace +./scripts/clj-test.sh +``` + +## Risks / gotchas + +1. **`Ord` on `ItemId`**: must match prior **`CanonicalItemUrl`** / `String` ordering wherever **`BTreeSet`** is used (e.g. deterministic scope-rank snapshots in **`write_actor`**). +2. **External `ItemId`**: **`Url`** equality / hashing — normalization is already centralized in **`url_normalize`**; ensure **`ItemId::parse`** always inserts normalized **`Url`** into **`External`**. +3. **Fake parent URLs** in garden (e.g. **`https://.`** for external root ranking): find all **`parse("https://.")`** style hacks and express as **`ItemId`** or a dedicated sentinel. +4. **Serde**: tests and any RPC clients that snapshot JSON may need expectation updates if **`VoteData`** shape changes. + +## Optional follow-ups (not blocking `ItemId`) + +- More **domain normalizers** in **`url_normalize.rs`** (e.g. `music.youtube.com`, Spotify, etc.). +- **Room wire** vs **HTTP segment** helpers already in **`paths.rs`** (`ROOM_SHORT_ID_LEN`, `room_route_segment`, `room_id_from_route_segment`). + +--- + +**End state criteria:** `rg CanonicalItemUrl` returns nothing; reducer maps use **`ItemId`**; HTML link generation for items goes through **`RouteContext` + `ItemId`**; tests and Kaocha green. diff --git a/server/src/html/mod.rs b/server/src/html/mod.rs index c3dccd03dc6da2a2f6f6fa828657e33a951884b1..668403a35c415e6f091362b28c63ac294057fb6d 100644 --- a/server/src/html/mod.rs +++ b/server/src/html/mod.rs @@ -15,6 +15,7 @@ mod breadcrumb_path; mod editor; mod forum; mod garden; +pub mod routing; mod search; pub mod ui_action; use breadcrumb_path::{ExternalOntologyPath, OntologyPath}; @@ -35,6 +36,7 @@ pub use garden::{ external_garden_index, external_ontology_path, garden_index, ontology_path, room_external_garden_index, room_external_ontology_path, room_garden_index, room_ontology_path, }; +pub use routing::RouteContext; pub use search::{search_page, search_results_fragment}; pub use forum::user_profile_page; pub use ui_action::{parse_html_ui_from_form, HtmlUiAction, HtmlUiParseError, UI_RPC_FIELD}; diff --git a/server/src/html/routing.rs b/server/src/html/routing.rs new file mode 100644 index 0000000000000000000000000000000000000000..13e9ae0cbe32d454e34a7737edf8234d2a558502 --- /dev/null +++ b/server/src/html/routing.rs @@ -0,0 +1,71 @@ +//! Scoped browser paths for HTML. **RouteContext** is the intended single place to build `href`s +//! given public vs room scope (the original blueprint name); today it wraps [`ThreadNav`]. +//! +//! Prefer `RouteContext::item_href` / [`RouteContext::thread_url`] in new Maud over stitching +//! `/r/…` vs `/~` manually. Call sites can migrate incrementally from passing `&ThreadNav`. + +use crate::path_types::CanonicalItemUrl; + +use super::forum::ThreadNav; + +#[derive(Clone)] +pub struct RouteContext(ThreadNav); + +impl RouteContext { + #[inline] + pub fn public() -> Self { + Self(ThreadNav::public()) + } + + #[inline] + pub fn from_room_id(room_id: &str) -> Option { + ThreadNav::from_room_id(room_id).map(Self) + } + + #[inline] + pub fn thread_nav(&self) -> &ThreadNav { + &self.0 + } + + #[inline] + pub fn into_thread_nav(self) -> ThreadNav { + self.0 + } + + /// Relative path for a stored canonical item in this scope’s garden. + pub fn item_href(&self, item: &CanonicalItemUrl) -> String { + self.0.garden_item_url(item.as_str()) + } + + /// Same as [`Self::item_href`] but parses `item` first (raw DSL / user paste). + pub fn item_href_raw(&self, item: &str) -> String { + self.0.garden_item_url(item) + } + + #[inline] + pub fn thread_url(&self, tag: &str) -> String { + self.0.thread_url(tag) + } + + #[inline] + pub fn garden_root_url(&self) -> &str { + self.0.garden_root_url() + } + + #[inline] + pub fn room_url(&self) -> &str { + self.0.room_url() + } +} + +impl From for RouteContext { + fn from(nav: ThreadNav) -> Self { + Self(nav) + } +} + +impl From for ThreadNav { + fn from(ctx: RouteContext) -> Self { + ctx.0 + } +} 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