diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..32d88e2 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,33 @@ +name: CI + +on: + push: + pull_request: + +jobs: + check: + name: fmt + clippy + tests (macOS) + runs-on: macos-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + with: + components: rustfmt, clippy + + - name: Cache cargo + uses: Swatinem/rust-cache@v2 + + - name: Install nextest + uses: taiki-e/install-action@nextest + + - name: Format check + run: cargo fmt --all --check + + - name: Clippy + run: cargo clippy --workspace --all-targets --all-features -- -D warnings + + - name: Tests + run: cargo nextest run --workspace diff --git a/Cargo.lock b/Cargo.lock index 1070e06..cf685c5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3034,6 +3034,8 @@ dependencies = [ [[package]] name = "nodedb-client" version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e4c775d84d70535bfa1e4f70fcd185b7f555385ef2122f5ec588c418e0a63ed5" dependencies = [ "async-trait", "nodedb-types", @@ -3046,6 +3048,8 @@ dependencies = [ [[package]] name = "nodedb-codec" version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a21422d579c2ac16e5237e4cd1144262dd4641d072345b4a891844fc1cd557a8" dependencies = [ "lz4_flex", "nalgebra", @@ -3078,6 +3082,8 @@ dependencies = [ [[package]] name = "nodedb-types" version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77145a5667abbe3726f46b50ff60ea5cb771fc4fcc782904ee2b03f9b46c2c00" dependencies = [ "aes-gcm", "bytemuck", diff --git a/Cargo.toml b/Cargo.toml index f033695..01498a0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,7 +21,7 @@ nodedb-client = "0.3.0" nodedb-types = "0.3.0" # Async async-trait = "0.1" -tokio = { version = "1", features = ["rt-multi-thread", "macros"] } +tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] } # UI. The `router` feature re-exports the router through dioxus::prelude. # Desktop-only for the skeleton. dioxus = { version = "0.7", features = ["desktop", "router"] } diff --git a/nodedb-studio/src/app.rs b/nodedb-studio/src/app.rs index 1a06c34..e0c01fd 100644 --- a/nodedb-studio/src/app.rs +++ b/nodedb-studio/src/app.rs @@ -5,7 +5,7 @@ //! - Connected -> `Studio` //! //! Global state is exposed as fine-grained signals via context. The backend -//! seam is a `dyn ConnectionService` behind an `Rc`, so swapping the mock for a +//! seam is a `dyn Backend` behind an `Rc`, so swapping the mock for a //! real NodeDB client later is a one-line change here. use std::rc::Rc; @@ -15,7 +15,8 @@ use dioxus::prelude::*; use crate::modals::ModalHost; use crate::models::notification::Notification; use crate::services::async_state::AsyncState; -use crate::services::connection_service::{ConnectionService, MockConnectionService}; +use crate::services::backend::Backend; +use crate::services::connection_service::MockConnectionService; use crate::state::connection::ActiveConnection; use crate::state::connections_registry::SavedConnection; use crate::state::preferences::Preferences; @@ -30,7 +31,7 @@ const STYLES: Asset = asset!("/assets/styles.css"); pub fn App() -> Element { // The single seam to the outside world. Provided as a trait object so a // real client can replace the mock without touching consumers. - let service: Rc = Rc::new(MockConnectionService); + let service: Rc = Rc::new(MockConnectionService::ready()); // The real-client stub's instantiability + object-safety behind the seam is // proven by `nodedb_service::tests::stub_is_object_safe_behind_rc`, so it is // not constructed here on every render. diff --git a/nodedb-studio/src/components/command_palette.rs b/nodedb-studio/src/components/command_palette.rs index ebe3e6f..ae493ac 100644 --- a/nodedb-studio/src/components/command_palette.rs +++ b/nodedb-studio/src/components/command_palette.rs @@ -6,7 +6,7 @@ use dioxus::prelude::*; use crate::routes::Route; -use crate::services::connection_service::ConnectionService; +use crate::services::backend::Backend; use crate::state::connection::ActiveConnection; use crate::state::ui::ModalKind; @@ -15,7 +15,7 @@ pub fn CommandPalette() -> Element { let mut open = use_context::>(); let mut active = use_context::>>(); let mut modal = use_context::>>(); - let service = use_context::>(); + let service = use_context::>(); let nav = use_navigator(); if !*open.read() { diff --git a/nodedb-studio/src/components/live_tail.rs b/nodedb-studio/src/components/live_tail.rs new file mode 100644 index 0000000..6a9dea9 --- /dev/null +++ b/nodedb-studio/src/components/live_tail.rs @@ -0,0 +1,57 @@ +//! Reusable live-tail layout shell (frame + toolbar/body/footer slots). Owns the +//! `.live-tail` structure + CSS so every live screen (CDC, Notify, MV, Console) +//! composes the same chrome instead of re-emitting it. Screen-specific content +//! (rows, status pills, footer stats) is passed in as slot Elements. + +use dioxus::prelude::*; + +#[derive(Clone, PartialEq, Props)] +pub struct LiveTailProps { + /// Toolbar content (title, status pill, filter, Pause/Export buttons). + pub toolbar: Element, + /// Scrolling body (the rows). + pub body: Element, + /// Footer content (buffer/lag stats). + pub footer: Element, +} + +#[component] +pub fn LiveTail(props: LiveTailProps) -> Element { + rsx! { + div { class: "live-tail", + div { class: "tail-toolbar", {props.toolbar} } + div { class: "tail-body", {props.body} } + div { class: "tail-footer", {props.footer} } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn render(app: fn() -> Element) -> String { + let mut dom = VirtualDom::new(app); + dom.rebuild_in_place(); + dioxus_ssr::render(&dom) + } + + #[test] + fn renders_three_slots_in_frame() { + fn app() -> Element { + rsx! { + LiveTail { + toolbar: rsx! { span { "TB" } }, + body: rsx! { div { "ROWS" } }, + footer: rsx! { span { "FT" } }, + } + } + } + let html = render(app); + assert!(html.contains("live-tail")); + assert!(html.contains("tail-toolbar")); + assert!(html.contains("tail-body")); + assert!(html.contains("tail-footer")); + assert!(html.contains("TB") && html.contains("ROWS") && html.contains("FT")); + } +} diff --git a/nodedb-studio/src/components/mod.rs b/nodedb-studio/src/components/mod.rs index b224990..7503d1b 100644 --- a/nodedb-studio/src/components/mod.rs +++ b/nodedb-studio/src/components/mod.rs @@ -2,10 +2,12 @@ pub mod async_view; pub mod command_palette; +pub mod live_tail; pub mod modal; pub mod popovers; pub mod rail; pub mod snav; +pub mod sparkline; pub mod statusbar; pub mod subnav; pub mod topbar; diff --git a/nodedb-studio/src/components/popovers/connection_popover.rs b/nodedb-studio/src/components/popovers/connection_popover.rs index 53fd6dd..3af40d1 100644 --- a/nodedb-studio/src/components/popovers/connection_popover.rs +++ b/nodedb-studio/src/components/popovers/connection_popover.rs @@ -5,7 +5,7 @@ use std::rc::Rc; use dioxus::prelude::*; -use crate::services::connection_service::ConnectionService; +use crate::services::backend::Backend; use crate::state::connection::ActiveConnection; use crate::state::connections_registry::{ConnStatus, SavedConnection}; use crate::state::ui::{ModalKind, Popover}; @@ -16,7 +16,7 @@ pub fn ConnectionPopover() -> Element { let mut popover = use_context::>>(); let mut modal = use_context::>>(); let registry = use_context::>>(); - let service = use_context::>(); + let service = use_context::>(); let conn = active.read(); let Some(c) = conn.as_ref() else { diff --git a/nodedb-studio/src/components/popovers/notification_popover.rs b/nodedb-studio/src/components/popovers/notification_popover.rs index eaff8f0..19a64cc 100644 --- a/nodedb-studio/src/components/popovers/notification_popover.rs +++ b/nodedb-studio/src/components/popovers/notification_popover.rs @@ -10,12 +10,15 @@ //! and the list never diverge. The Error-state Retry reloads the feed via the //! shared `Resource` handle, gated on `StudioError::is_retriable()`. +use std::rc::Rc; + use dioxus::prelude::*; use crate::components::async_view::AsyncView; use crate::models::notification::{Notification, NotificationTarget}; use crate::routes::Route; use crate::services::async_state::AsyncState; +use crate::services::backend::Backend; use crate::state::connection::{ActiveConnection, Capabilities}; use crate::state::notifications::{mark_all_read, mark_read, visible}; use crate::state::ui::Popover; @@ -45,6 +48,7 @@ pub fn NotificationPopover() -> Element { let mut reload = use_context::>(); let mut popover = use_context::>>(); let active = use_context::>>(); + let backend = use_context::>(); let nav = use_navigator(); // Capability gate (unchanged): no connection -> render nothing. @@ -86,10 +90,24 @@ pub fn NotificationPopover() -> Element { h4 { "Notifications " span { class: "count", "{count_label}" } } button { onclick: move |_| { - // Mutate the shared store; the badge re-derives from it. + // In-memory update first for snappy UI (write guard dropped + // before the block ends, never held across an await). if let Some(items) = store.write().loaded_mut() { mark_all_read(items); } + // Persist through the seam so the badge stays cleared on + // any subsequent reload (fixes POP-03). The spawn avoids + // holding any signal guard across the await. + // On success, reconcile the shared feed so the real client's + // persisted state is reflected (reload.restart() re-fetches). + let backend = backend.clone(); + let mut reload = reload; + spawn(async move { + match backend.mark_all_read().await { + Ok(()) => reload.restart(), + Err(e) => tracing::warn!("mark_all_read failed: {e}"), + } + }); }, "Mark all read" } diff --git a/nodedb-studio/src/components/sparkline.rs b/nodedb-studio/src/components/sparkline.rs new file mode 100644 index 0000000..d198358 --- /dev/null +++ b/nodedb-studio/src/components/sparkline.rs @@ -0,0 +1,75 @@ +//! SVG-fidelity spike: a minimal sparkline drawn as an inline `` +//! in pure Dioxus (no JS interop), fed from a typed `Vec`. Proves whether +//! rsx!-driven SVG gives the fidelity later chart/topology screens need. + +use dioxus::prelude::*; + +#[derive(Clone, PartialEq, Props)] +pub struct SparklineProps { + /// Y values; X is the index. Normalized to a 100x24 viewBox. + pub points: Vec, +} + +#[component] +pub fn Sparkline(props: SparklineProps) -> Element { + let pts = &props.points; + if pts.len() < 2 { + return rsx! { svg { width: "100", height: "24", "viewBox": "0 0 100 24" } }; + } + let max = pts.iter().cloned().fold(f64::MIN, f64::max); + let min = pts.iter().cloned().fold(f64::MAX, f64::min); + let span = (max - min).max(f64::EPSILON); + let n = (pts.len() - 1) as f64; + let coords: String = pts + .iter() + .enumerate() + .map(|(i, y)| { + let x = (i as f64 / n) * 100.0; + let yy = 24.0 - ((y - min) / span) * 24.0; + format!("{x:.1},{yy:.1}") + }) + .collect::>() + .join(" "); + rsx! { + svg { width: "100", height: "24", "viewBox": "0 0 100 24", + polyline { + points: "{coords}", + fill: "none", + stroke: "currentColor", + "stroke-width": "1.5", + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn render(app: fn() -> Element) -> String { + let mut dom = VirtualDom::new(app); + dom.rebuild_in_place(); + dioxus_ssr::render(&dom) + } + + #[test] + fn renders_polyline_from_points() { + fn app() -> Element { + rsx! { Sparkline { points: vec![1.0, 4.0, 2.0, 8.0, 5.0] } } + } + let html = render(app); + assert!(html.contains(" Vec { - vec![ - SavedConnection { - name: "local-nodedb-dev".into(), - meta: "nodedb · localhost:2480".into(), - sub: "nodedb · 8.4ms · 3 dbs".into(), - status: ConnStatus::Online, - db_count: Some(3), - ping: Some("8.4ms".into()), - server: "dev".into(), - profile: Some(ConnectionProfile { - user: "root".into(), - role: "admin".into(), - capabilities: Capabilities { - graph: true, - vector: true, - streams: true, - timeseries: true, - spatial: true, - fts: true, - sync: false, - cluster: false, - readonly: false, - }, - databases: vec![ - "analytics".into(), - "events_log".into(), - "social_graph".into(), - ], - default_database: "analytics".into(), - }), - }, - SavedConnection { - name: "staging-cluster".into(), - meta: "nodedb · 3-node cluster · vpn".into(), - sub: "nodedb · 22ms · 12 dbs".into(), - status: ConnStatus::Online, - db_count: Some(12), - ping: Some("22ms".into()), - server: "dev".into(), - profile: Some(ConnectionProfile { - user: "hatta_admin".into(), - role: "admin".into(), - capabilities: Capabilities { - graph: true, - vector: true, - streams: true, - timeseries: true, - spatial: true, - fts: true, - sync: true, - cluster: true, - readonly: false, - }, - databases: vec![ - "analytics".into(), - "events_log".into(), - "social_graph".into(), - "iot_telemetry".into(), - "docs_corpus".into(), - "cache_layer".into(), - "staging_a".into(), - "staging_b".into(), - "audit_trail".into(), - "temp_workspace".into(), - "sandbox".into(), - "archive_2025".into(), - ], - default_database: "analytics".into(), - }), - }, - SavedConnection { - name: "prod-replica-eu".into(), - meta: "nodedb · eu-fra-1 · tls".into(), - sub: "nodedb · 98ms · 28 dbs".into(), - status: ConnStatus::ReadOnly, - db_count: Some(28), - ping: Some("98ms".into()), - server: "dev".into(), - profile: Some(ConnectionProfile { - user: "hatta_ro".into(), - role: "analyst (read-only)".into(), - capabilities: Capabilities { - graph: true, - vector: true, - streams: true, - timeseries: true, - spatial: true, - fts: true, - sync: true, - cluster: true, - readonly: true, - }, - databases: vec![ - "analytics_prod".into(), - "events_prod".into(), - "social_prod".into(), - "orders_prod".into(), - "iot_prod".into(), - "docs_prod".into(), - "audit_prod".into(), - "cache_prod".into(), - ], - default_database: "analytics_prod".into(), - }), - }, - SavedConnection { - name: "test-nodedb".into(), - meta: "nodedb · localhost:2480".into(), - sub: String::new(), - status: ConnStatus::Offline, - db_count: None, - ping: None, - server: "dev".into(), - profile: None, - }, - ] -} - -/// Explorer collections, in sidebar display order (grouped by storage mode). -/// One NodeDB instance exposes all eight modes; these are not separate engines. -pub fn explorer_collections() -> Vec { - let c = |name: &str, mode, count: &str| Collection { - name: name.to_string(), - mode, - count: count.to_string(), - }; - vec![ - c("users", StorageMode::Document, "12,481"), - c("events", StorageMode::Document, "2.4M"), - c("sessions", StorageMode::Document, "88,209"), - c("orders", StorageMode::Strict, "442,003"), - c("invoices", StorageMode::Strict, "95,818"), - c("doc_embeddings", StorageMode::Vector, "1.1M"), - c("product_embeds", StorageMode::Vector, "88,400"), - c("social_graph", StorageMode::Graph, "3.2M"), - c("metrics", StorageMode::Timeseries, "48M"), - c("sensor_temps", StorageMode::Timeseries, "5.1M"), - c("sessions_cache", StorageMode::Kv, "18,200"), - c("feature_flags", StorageMode::Kv, "42"), - c("store_locations", StorageMode::Spatial, "2,108"), - c("articles_idx", StorageMode::Fts, "241,005"), - ] -} - -/// The notification feed. Capability gating is applied at render time against -/// the active connection (see `state::notifications`). -pub fn notifications() -> Vec { - vec![ - Notification { - id: "sync-conflict-1".into(), - severity: Severity::Warn, - group: "Sync conflicts".into(), - required_cap: Some(Capability::Sync), - title: "users / u_44182 / email".into(), - desc: "Two writes within 14ms · LWW pending".into(), - when: "2m ago".into(), - target: NotificationTarget::Sync, - unread: true, - }, - Notification { - id: "sync-conflict-2".into(), - severity: Severity::Warn, - group: "Sync conflicts".into(), - required_cap: Some(Capability::Sync), - title: "users / u_77103 / preferences".into(), - desc: "Concurrent edit on edge-sg-1 · merge needed".into(), - when: "4m ago".into(), - target: NotificationTarget::Sync, - unread: true, - }, - Notification { - id: "cron-fail".into(), - severity: Severity::Err, - group: "Scheduled jobs".into(), - required_cap: None, - title: "vector_reindex failed".into(), - desc: "Out of memory at step 4 · last 3 runs failed".into(), - when: "3d ago".into(), - target: NotificationTarget::StreamsCron, - unread: true, - }, - Notification { - id: "stream-stalled".into(), - severity: Severity::Warn, - group: "Streams".into(), - required_cap: Some(Capability::Streams), - title: "payment_failed consumer stalled".into(), - desc: "lag 4.2s · group=fraud-svc · 1 of 2 stalled".into(), - when: "8m ago".into(), - target: NotificationTarget::StreamsTopics, - unread: true, - }, - Notification { - id: "lag-resolved".into(), - severity: Severity::Info, - group: "Cluster".into(), - required_cap: Some(Capability::Cluster), - title: "Replication lag normalized".into(), - desc: "node-5 caught up · was 1,309 behind".into(), - when: "12m ago".into(), - target: NotificationTarget::Admin, - unread: false, - }, - Notification { - id: "query-done".into(), - severity: Severity::Info, - group: "Queries".into(), - required_cap: None, - title: "Long-running query finished".into(), - desc: "\"orders_by_region_2024.sql\" · 18 rows in 2m 41s".into(), - when: "22m ago".into(), - target: NotificationTarget::Query, - unread: false, - }, - ] -} - -// ── Streams payloads ───────────────────────────────────────────────────────── -// -// The CDC and LISTEN/NOTIFY tails show database records. Those records are -// modelled here as native typed documents (`MockDoc`), mirroring how the real -// `ConnectionService` will hand back `nodedb_types::Value` documents from the -// client. The viewers serialize them to JSON via `sonic_rs` purely for display -// — no JSON strings are carried around as data. `MockDoc` preserves field order -// (unlike `Value::Object`'s `HashMap`), so the rendered JSON is deterministic. - -use FieldValue::{Float, Int, Nested, Str}; - -/// A scalar (or nested-document) field value inside a [`MockDoc`]. -pub enum FieldValue { - Str(&'static str), - Int(i64), - Float(f64), - Nested(MockDoc), -} - -/// An ordered document: `(key, value)` pairs in display order. Stands in for a -/// `nodedb_types::Value::Object` returned by the client. -pub struct MockDoc(pub Vec<(&'static str, FieldValue)>); - -impl Serialize for MockDoc { - fn serialize(&self, serializer: S) -> Result { - let mut map = serializer.serialize_map(Some(self.0.len()))?; - for (key, value) in &self.0 { - map.serialize_entry(key, value)?; - } - map.end() - } -} - -impl Serialize for FieldValue { - fn serialize(&self, serializer: S) -> Result { - match self { - FieldValue::Str(s) => serializer.serialize_str(s), - FieldValue::Int(n) => serializer.serialize_i64(*n), - FieldValue::Float(f) => serializer.serialize_f64(*f), - FieldValue::Nested(doc) => doc.serialize(serializer), - } - } -} - -fn doc(fields: Vec<(&'static str, FieldValue)>) -> MockDoc { - MockDoc(fields) -} - -/// The change operation in a CDC event. -#[derive(Clone, Copy)] -pub enum ChangeOp { - Insert, - Update, - Delete, -} - -impl ChangeOp { - /// Uppercase label shown in the op column. - pub fn label(self) -> &'static str { - match self { - ChangeOp::Insert => "INSERT", - ChangeOp::Update => "UPDATE", - ChangeOp::Delete => "DELETE", - } - } - - /// CSS modifier class for the op pill. - pub fn css(self) -> &'static str { - match self { - ChangeOp::Insert => "ins", - ChangeOp::Update => "upd", - ChangeOp::Delete => "del", - } - } -} - -/// One row in the Streams · CDC live tail. -pub struct ChangeEvent { - pub time: &'static str, - pub op: ChangeOp, - pub collection: &'static str, - pub payload: MockDoc, - /// Optional annotation appended after the document (e.g. `⤳ +1 field`). - pub note: Option<&'static str>, -} - -/// The CDC change feed, newest first. -pub fn cdc_events() -> Vec { - vec![ - ChangeEvent { - time: "04:23:18.041", - op: ChangeOp::Insert, - collection: "events", - payload: doc(vec![ - ("_id", Str("evt_01HMNJ…")), - ("type", Str("page_view")), - ("user_id", Str("u_44182")), - ("props", Nested(doc(vec![("path", Str("/dashboard"))]))), - ]), - note: None, - }, - ChangeEvent { - time: "04:23:18.039", - op: ChangeOp::Update, - collection: "sessions", - payload: doc(vec![ - ("_id", Str("s_88209")), - ("last_seen", Str("2026-06-13T04:23:18Z")), - ]), - note: Some("⤳ +1 field"), - }, - ChangeEvent { - time: "04:23:18.037", - op: ChangeOp::Insert, - collection: "events", - payload: doc(vec![ - ("_id", Str("evt_01HMNJ…")), - ("type", Str("click")), - ("user_id", Str("u_77103")), - ("props", Nested(doc(vec![("el", Str("#cta-buy"))]))), - ]), - note: None, - }, - ChangeEvent { - time: "04:23:18.035", - op: ChangeOp::Insert, - collection: "orders", - payload: doc(vec![ - ("id", Int(442004)), - ("user_id", Str("u_77103")), - ("total", Float(89.40)), - ("currency", Str("USD")), - ]), - note: None, - }, - ChangeEvent { - time: "04:23:18.033", - op: ChangeOp::Delete, - collection: "sessions_cache", - payload: doc(vec![("key", Str("session:u_91002"))]), - note: None, - }, - ChangeEvent { - time: "04:23:18.031", - op: ChangeOp::Insert, - collection: "events", - payload: doc(vec![ - ("_id", Str("evt_01HMNJ…")), - ("type", Str("scroll")), - ("user_id", Str("u_12998")), - ]), - note: None, - }, - ChangeEvent { - time: "04:23:18.028", - op: ChangeOp::Update, - collection: "users", - payload: doc(vec![ - ("_id", Str("u_44182")), - ("last_login", Str("2026-06-13T04:23:18Z")), - ]), - note: None, - }, - ChangeEvent { - time: "04:23:18.025", - op: ChangeOp::Insert, - collection: "events", - payload: doc(vec![ - ("_id", Str("evt_01HMNJ…")), - ("type", Str("page_view")), - ("user_id", Str("u_31001")), - ("props", Nested(doc(vec![("path", Str("/pricing"))]))), - ]), - note: None, - }, - ChangeEvent { - time: "04:23:18.022", - op: ChangeOp::Insert, - collection: "events", - payload: doc(vec![ - ("_id", Str("evt_01HMNJ…")), - ("type", Str("form_submit")), - ("user_id", Str("u_44182")), - ("props", Nested(doc(vec![("form", Str("feedback"))]))), - ]), - note: None, - }, - ] -} - -/// A LISTEN/NOTIFY channel in the sidebar. -pub struct NotifyChannel { - pub name: &'static str, - pub listeners: &'static str, - pub active: bool, -} - -/// One row in the LISTEN/NOTIFY live tail. -pub struct NotifyMessage { - pub time: &'static str, - pub source: &'static str, - pub payload: MockDoc, -} - -/// The notify channel list. -pub fn notify_channels() -> Vec { - let ch = |name, listeners, active| NotifyChannel { - name, - listeners, - active, - }; - vec![ - ch("user_events", "12", true), - ch("deploy_hooks", "3", false), - ch("cache_invalidate", "5", false), - ch("alerts", "8", false), - ch("jobs_done", "14", false), - ch("presence_room_1", "22", false), - ] -} - -/// The pub/sub message tail for the active channel. -pub fn notify_messages() -> Vec { - vec![ - NotifyMessage { - time: "04:23:18.041", - source: "api-server-2", - payload: doc(vec![("event", Str("login")), ("user", Str("u_44182"))]), - }, - NotifyMessage { - time: "04:23:17.812", - source: "webhook-relay", - payload: doc(vec![ - ("event", Str("signup")), - ("user", Str("u_99001")), - ("plan", Str("pro")), - ]), - }, - NotifyMessage { - time: "04:23:17.501", - source: "api-server-1", - payload: doc(vec![ - ("event", Str("profile_update")), - ("user", Str("u_77103")), - ]), - }, - NotifyMessage { - time: "04:23:16.998", - source: "analytics", - payload: doc(vec![ - ("event", Str("page_view")), - ("user", Str("u_44182")), - ("path", Str("/pricing")), - ]), - }, - NotifyMessage { - time: "04:23:16.422", - source: "api-server-2", - payload: doc(vec![("event", Str("logout")), ("user", Str("u_31001"))]), - }, - ] -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn payload_serializes_with_fields_in_declared_order() { - // The orders INSERT row: keys must come out in insertion order, not - // the alphabetical/HashMap order a `Value::Object` would impose. - let json = sonic_rs::to_string(&cdc_events()[3].payload).unwrap(); - assert_eq!( - json, - r#"{"id":442004,"user_id":"u_77103","total":89.4,"currency":"USD"}"# - ); - } - - #[test] - fn nested_document_serializes() { - let json = sonic_rs::to_string(&cdc_events()[0].payload).unwrap(); - assert_eq!( - json, - r#"{"_id":"evt_01HMNJ…","type":"page_view","user_id":"u_44182","props":{"path":"/dashboard"}}"# - ); - } -} diff --git a/nodedb-studio/src/data/mock/cdc.rs b/nodedb-studio/src/data/mock/cdc.rs new file mode 100644 index 0000000..de3e5c9 --- /dev/null +++ b/nodedb-studio/src/data/mock/cdc.rs @@ -0,0 +1,149 @@ +use super::docs::{FieldValue, MockDoc, doc}; +use FieldValue::{Float, Int, Nested, Str}; + +/// The change operation in a CDC event. +#[derive(Clone, Copy)] +pub enum ChangeOp { + Insert, + Update, + Delete, +} + +/// One row in the Streams · CDC live tail. +pub struct ChangeEvent { + pub time: &'static str, + pub op: ChangeOp, + pub collection: &'static str, + pub payload: MockDoc, + /// Optional annotation appended after the document (e.g. `⤳ +1 field`). + pub note: Option<&'static str>, +} + +/// The CDC change feed, newest first. +pub fn cdc_events() -> Vec { + vec![ + ChangeEvent { + time: "04:23:18.041", + op: ChangeOp::Insert, + collection: "events", + payload: doc(vec![ + ("_id", Str("evt_01HMNJ…")), + ("type", Str("page_view")), + ("user_id", Str("u_44182")), + ("props", Nested(doc(vec![("path", Str("/dashboard"))]))), + ]), + note: None, + }, + ChangeEvent { + time: "04:23:18.039", + op: ChangeOp::Update, + collection: "sessions", + payload: doc(vec![ + ("_id", Str("s_88209")), + ("last_seen", Str("2026-06-13T04:23:18Z")), + ]), + note: Some("⤳ +1 field"), + }, + ChangeEvent { + time: "04:23:18.037", + op: ChangeOp::Insert, + collection: "events", + payload: doc(vec![ + ("_id", Str("evt_01HMNJ…")), + ("type", Str("click")), + ("user_id", Str("u_77103")), + ("props", Nested(doc(vec![("el", Str("#cta-buy"))]))), + ]), + note: None, + }, + ChangeEvent { + time: "04:23:18.035", + op: ChangeOp::Insert, + collection: "orders", + payload: doc(vec![ + ("id", Int(442004)), + ("user_id", Str("u_77103")), + ("total", Float(89.40)), + ("currency", Str("USD")), + ]), + note: None, + }, + ChangeEvent { + time: "04:23:18.033", + op: ChangeOp::Delete, + collection: "sessions_cache", + payload: doc(vec![("key", Str("session:u_91002"))]), + note: None, + }, + ChangeEvent { + time: "04:23:18.031", + op: ChangeOp::Insert, + collection: "events", + payload: doc(vec![ + ("_id", Str("evt_01HMNJ…")), + ("type", Str("scroll")), + ("user_id", Str("u_12998")), + ]), + note: None, + }, + ChangeEvent { + time: "04:23:18.028", + op: ChangeOp::Update, + collection: "users", + payload: doc(vec![ + ("_id", Str("u_44182")), + ("last_login", Str("2026-06-13T04:23:18Z")), + ]), + note: None, + }, + ChangeEvent { + time: "04:23:18.025", + op: ChangeOp::Insert, + collection: "events", + payload: doc(vec![ + ("_id", Str("evt_01HMNJ…")), + ("type", Str("page_view")), + ("user_id", Str("u_31001")), + ("props", Nested(doc(vec![("path", Str("/pricing"))]))), + ]), + note: None, + }, + ChangeEvent { + time: "04:23:18.022", + op: ChangeOp::Insert, + collection: "events", + payload: doc(vec![ + ("_id", Str("evt_01HMNJ…")), + ("type", Str("form_submit")), + ("user_id", Str("u_44182")), + ("props", Nested(doc(vec![("form", Str("feedback"))]))), + ]), + note: None, + }, + ] +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn payload_serializes_with_fields_in_declared_order() { + // The orders INSERT row: keys must come out in insertion order, not + // the alphabetical/HashMap order a `Value::Object` would impose. + let json = sonic_rs::to_string(&cdc_events()[3].payload).unwrap(); + assert_eq!( + json, + r#"{"id":442004,"user_id":"u_77103","total":89.4,"currency":"USD"}"# + ); + } + + #[test] + fn nested_document_serializes() { + let json = sonic_rs::to_string(&cdc_events()[0].payload).unwrap(); + assert_eq!( + json, + r#"{"_id":"evt_01HMNJ…","type":"page_view","user_id":"u_44182","props":{"path":"/dashboard"}}"# + ); + } +} diff --git a/nodedb-studio/src/data/mock/connections.rs b/nodedb-studio/src/data/mock/connections.rs new file mode 100644 index 0000000..9311e34 --- /dev/null +++ b/nodedb-studio/src/data/mock/connections.rs @@ -0,0 +1,224 @@ +use crate::models::collection::{Collection, StorageMode}; +use crate::models::notification::{Notification, NotificationTarget, Severity}; +use crate::state::connection::{Capabilities, Capability}; +use crate::state::connections_registry::{ConnStatus, ConnectionProfile, SavedConnection}; + +/// The four saved connections shown on the Connection Manager. Three are +/// reachable; `test-nodedb` is offline (no profile), matching the mockup. +pub fn connections() -> Vec { + vec![ + SavedConnection { + name: "local-nodedb-dev".into(), + meta: "nodedb · localhost:2480".into(), + sub: "nodedb · 8.4ms · 3 dbs".into(), + status: ConnStatus::Online, + db_count: Some(3), + ping: Some("8.4ms".into()), + server: "dev".into(), + profile: Some(ConnectionProfile { + user: "root".into(), + role: "admin".into(), + capabilities: Capabilities { + graph: true, + vector: true, + streams: true, + timeseries: true, + spatial: true, + fts: true, + sync: false, + cluster: false, + readonly: false, + }, + databases: vec![ + "analytics".into(), + "events_log".into(), + "social_graph".into(), + ], + default_database: "analytics".into(), + }), + }, + SavedConnection { + name: "staging-cluster".into(), + meta: "nodedb · 3-node cluster · vpn".into(), + sub: "nodedb · 22ms · 12 dbs".into(), + status: ConnStatus::Online, + db_count: Some(12), + ping: Some("22ms".into()), + server: "dev".into(), + profile: Some(ConnectionProfile { + user: "hatta_admin".into(), + role: "admin".into(), + capabilities: Capabilities { + graph: true, + vector: true, + streams: true, + timeseries: true, + spatial: true, + fts: true, + sync: true, + cluster: true, + readonly: false, + }, + databases: vec![ + "analytics".into(), + "events_log".into(), + "social_graph".into(), + "iot_telemetry".into(), + "docs_corpus".into(), + "cache_layer".into(), + "staging_a".into(), + "staging_b".into(), + "audit_trail".into(), + "temp_workspace".into(), + "sandbox".into(), + "archive_2025".into(), + ], + default_database: "analytics".into(), + }), + }, + SavedConnection { + name: "prod-replica-eu".into(), + meta: "nodedb · eu-fra-1 · tls".into(), + sub: "nodedb · 98ms · 28 dbs".into(), + status: ConnStatus::ReadOnly, + db_count: Some(28), + ping: Some("98ms".into()), + server: "dev".into(), + profile: Some(ConnectionProfile { + user: "hatta_ro".into(), + role: "analyst (read-only)".into(), + capabilities: Capabilities { + graph: true, + vector: true, + streams: true, + timeseries: true, + spatial: true, + fts: true, + sync: true, + cluster: true, + readonly: true, + }, + databases: vec![ + "analytics_prod".into(), + "events_prod".into(), + "social_prod".into(), + "orders_prod".into(), + "iot_prod".into(), + "docs_prod".into(), + "audit_prod".into(), + "cache_prod".into(), + ], + default_database: "analytics_prod".into(), + }), + }, + SavedConnection { + name: "test-nodedb".into(), + meta: "nodedb · localhost:2480".into(), + sub: String::new(), + status: ConnStatus::Offline, + db_count: None, + ping: None, + server: "dev".into(), + profile: None, + }, + ] +} + +/// Explorer collections, in sidebar display order (grouped by storage mode). +/// One NodeDB instance exposes all eight modes; these are not separate engines. +pub fn explorer_collections() -> Vec { + let c = |name: &str, mode, count: &str| Collection { + name: name.to_string(), + mode, + count: count.to_string(), + }; + vec![ + c("users", StorageMode::Document, "12,481"), + c("events", StorageMode::Document, "2.4M"), + c("sessions", StorageMode::Document, "88,209"), + c("orders", StorageMode::Strict, "442,003"), + c("invoices", StorageMode::Strict, "95,818"), + c("doc_embeddings", StorageMode::Vector, "1.1M"), + c("product_embeds", StorageMode::Vector, "88,400"), + c("social_graph", StorageMode::Graph, "3.2M"), + c("metrics", StorageMode::Timeseries, "48M"), + c("sensor_temps", StorageMode::Timeseries, "5.1M"), + c("sessions_cache", StorageMode::Kv, "18,200"), + c("feature_flags", StorageMode::Kv, "42"), + c("store_locations", StorageMode::Spatial, "2,108"), + c("articles_idx", StorageMode::Fts, "241,005"), + ] +} + +/// The notification feed. Capability gating is applied at render time against +/// the active connection (see `state::notifications`). +pub fn notifications() -> Vec { + vec![ + Notification { + id: "sync-conflict-1".into(), + severity: Severity::Warn, + group: "Sync conflicts".into(), + required_cap: Some(Capability::Sync), + title: "users / u_44182 / email".into(), + desc: "Two writes within 14ms · LWW pending".into(), + when: "2m ago".into(), + target: NotificationTarget::Sync, + unread: true, + }, + Notification { + id: "sync-conflict-2".into(), + severity: Severity::Warn, + group: "Sync conflicts".into(), + required_cap: Some(Capability::Sync), + title: "users / u_77103 / preferences".into(), + desc: "Concurrent edit on edge-sg-1 · merge needed".into(), + when: "4m ago".into(), + target: NotificationTarget::Sync, + unread: true, + }, + Notification { + id: "cron-fail".into(), + severity: Severity::Err, + group: "Scheduled jobs".into(), + required_cap: None, + title: "vector_reindex failed".into(), + desc: "Out of memory at step 4 · last 3 runs failed".into(), + when: "3d ago".into(), + target: NotificationTarget::StreamsCron, + unread: true, + }, + Notification { + id: "stream-stalled".into(), + severity: Severity::Warn, + group: "Streams".into(), + required_cap: Some(Capability::Streams), + title: "payment_failed consumer stalled".into(), + desc: "lag 4.2s · group=fraud-svc · 1 of 2 stalled".into(), + when: "8m ago".into(), + target: NotificationTarget::StreamsTopics, + unread: true, + }, + Notification { + id: "lag-resolved".into(), + severity: Severity::Info, + group: "Cluster".into(), + required_cap: Some(Capability::Cluster), + title: "Replication lag normalized".into(), + desc: "node-5 caught up · was 1,309 behind".into(), + when: "12m ago".into(), + target: NotificationTarget::Admin, + unread: false, + }, + Notification { + id: "query-done".into(), + severity: Severity::Info, + group: "Queries".into(), + required_cap: None, + title: "Long-running query finished".into(), + desc: "\"orders_by_region_2024.sql\" · 18 rows in 2m 41s".into(), + when: "22m ago".into(), + target: NotificationTarget::Query, + unread: false, + }, + ] +} diff --git a/nodedb-studio/src/data/mock/docs.rs b/nodedb-studio/src/data/mock/docs.rs new file mode 100644 index 0000000..fbaa900 --- /dev/null +++ b/nodedb-studio/src/data/mock/docs.rs @@ -0,0 +1,47 @@ +use serde::ser::{Serialize, SerializeMap, Serializer}; + +// ── Streams payloads ───────────────────────────────────────────────────────── +// +// The CDC and LISTEN/NOTIFY tails show database records. Those records are +// modelled here as native typed documents (`MockDoc`), mirroring how the real +// `ConnectionService` will hand back `nodedb_types::Value` documents from the +// client. The viewers serialize them to JSON via `sonic_rs` purely for display +// — no JSON strings are carried around as data. `MockDoc` preserves field order +// (unlike `Value::Object`'s `HashMap`), so the rendered JSON is deterministic. + +/// A scalar (or nested-document) field value inside a [`MockDoc`]. +pub enum FieldValue { + Str(&'static str), + Int(i64), + Float(f64), + Nested(MockDoc), +} + +/// An ordered document: `(key, value)` pairs in display order. Stands in for a +/// `nodedb_types::Value::Object` returned by the client. +pub struct MockDoc(pub Vec<(&'static str, FieldValue)>); + +impl Serialize for MockDoc { + fn serialize(&self, serializer: S) -> Result { + let mut map = serializer.serialize_map(Some(self.0.len()))?; + for (key, value) in &self.0 { + map.serialize_entry(key, value)?; + } + map.end() + } +} + +impl Serialize for FieldValue { + fn serialize(&self, serializer: S) -> Result { + match self { + FieldValue::Str(s) => serializer.serialize_str(s), + FieldValue::Int(n) => serializer.serialize_i64(*n), + FieldValue::Float(f) => serializer.serialize_f64(*f), + FieldValue::Nested(doc) => doc.serialize(serializer), + } + } +} + +pub fn doc(fields: Vec<(&'static str, FieldValue)>) -> MockDoc { + MockDoc(fields) +} diff --git a/nodedb-studio/src/data/mock/mod.rs b/nodedb-studio/src/data/mock/mod.rs new file mode 100644 index 0000000..f3e8ed8 --- /dev/null +++ b/nodedb-studio/src/data/mock/mod.rs @@ -0,0 +1,16 @@ +//! Mock fixtures, split per domain. Module root re-exports the same public +//! surface the single `mock.rs` used to expose, so callers are unchanged. +//! +//! Naming note: the mockup's legacy "arcadedb" labels, "local-arcade-dev" +//! name, "arcade-5" node, and per-version server tags are deliberately NOT +//! reproduced. NodeDB version numbers are undecided (CLAUDE.md §2), so the +//! server stat is a neutral "dev" placeholder rather than an invented version. + +mod cdc; +mod connections; +mod docs; +mod notify; + +pub use cdc::{ChangeOp, cdc_events}; +pub use connections::{connections, explorer_collections, notifications}; +pub use notify::{notify_channels, notify_messages}; diff --git a/nodedb-studio/src/data/mock/notify.rs b/nodedb-studio/src/data/mock/notify.rs new file mode 100644 index 0000000..8dab253 --- /dev/null +++ b/nodedb-studio/src/data/mock/notify.rs @@ -0,0 +1,75 @@ +use super::docs::{FieldValue, MockDoc, doc}; +use FieldValue::Str; + +/// A LISTEN/NOTIFY channel in the sidebar. +pub struct NotifyChannel { + pub name: &'static str, + pub listeners: &'static str, + pub active: bool, +} + +/// One row in the LISTEN/NOTIFY live tail. +pub struct NotifyMessage { + pub time: &'static str, + pub source: &'static str, + pub payload: MockDoc, +} + +/// The notify channel list. +pub fn notify_channels() -> Vec { + let ch = |name, listeners, active| NotifyChannel { + name, + listeners, + active, + }; + vec![ + ch("user_events", "12", true), + ch("deploy_hooks", "3", false), + ch("cache_invalidate", "5", false), + ch("alerts", "8", false), + ch("jobs_done", "14", false), + ch("presence_room_1", "22", false), + ] +} + +/// The pub/sub message tail for the active channel. +pub fn notify_messages() -> Vec { + vec![ + NotifyMessage { + time: "04:23:18.041", + source: "api-server-2", + payload: doc(vec![("event", Str("login")), ("user", Str("u_44182"))]), + }, + NotifyMessage { + time: "04:23:17.812", + source: "webhook-relay", + payload: doc(vec![ + ("event", Str("signup")), + ("user", Str("u_99001")), + ("plan", Str("pro")), + ]), + }, + NotifyMessage { + time: "04:23:17.501", + source: "api-server-1", + payload: doc(vec![ + ("event", Str("profile_update")), + ("user", Str("u_77103")), + ]), + }, + NotifyMessage { + time: "04:23:16.998", + source: "analytics", + payload: doc(vec![ + ("event", Str("page_view")), + ("user", Str("u_44182")), + ("path", Str("/pricing")), + ]), + }, + NotifyMessage { + time: "04:23:16.422", + source: "api-server-2", + payload: doc(vec![("event", Str("logout")), ("user", Str("u_31001"))]), + }, + ] +} diff --git a/nodedb-studio/src/models/cdc.rs b/nodedb-studio/src/models/cdc.rs new file mode 100644 index 0000000..2cec9a5 --- /dev/null +++ b/nodedb-studio/src/models/cdc.rs @@ -0,0 +1,67 @@ +//! Typed model for a Streams · CDC live-tail row. Replaces the anonymous tuple +//! the view used to carry, so the row has a stable `id` for keyed lists. + +/// The change operation in a CDC event. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub enum CdcOp { + Insert, + Update, + Delete, +} + +impl CdcOp { + /// Uppercase label shown in the op column. + pub fn label(self) -> &'static str { + match self { + CdcOp::Insert => "INSERT", + CdcOp::Update => "UPDATE", + CdcOp::Delete => "DELETE", + } + } + + /// CSS modifier class for the op pill. + pub fn css(self) -> &'static str { + match self { + CdcOp::Insert => "ins", + CdcOp::Update => "upd", + CdcOp::Delete => "del", + } + } +} + +/// One row in the Streams · CDC live tail. `payload_json` is already serialized +/// for display (the seam serializes the native document at its boundary). +#[derive(Clone, PartialEq, Debug)] +pub struct CdcRow { + pub id: String, + pub time: String, + pub op: CdcOp, + pub collection: String, + pub payload_json: String, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn op_label_and_css_are_stable() { + assert_eq!(CdcOp::Insert.label(), "INSERT"); + assert_eq!(CdcOp::Insert.css(), "ins"); + assert_eq!(CdcOp::Update.label(), "UPDATE"); + assert_eq!(CdcOp::Delete.css(), "del"); + } + + #[test] + fn row_carries_stable_id() { + let row = CdcRow { + id: "cdc-0".to_string(), + time: "04:23:18.041".to_string(), + op: CdcOp::Insert, + collection: "events".to_string(), + payload_json: "{}".to_string(), + }; + assert_eq!(row.id, "cdc-0"); + assert_eq!(row.op.label(), "INSERT"); + } +} diff --git a/nodedb-studio/src/models/mod.rs b/nodedb-studio/src/models/mod.rs index 4a31759..3ace1d9 100644 --- a/nodedb-studio/src/models/mod.rs +++ b/nodedb-studio/src/models/mod.rs @@ -1,5 +1,6 @@ //! Typed domain models shared across views. These describe *data* (collections, //! databases, notifications); live UI state lives in `crate::state`. +pub mod cdc; pub mod collection; pub mod notification; diff --git a/nodedb-studio/src/services/async_state.rs b/nodedb-studio/src/services/async_state.rs index 273e6e2..28b150c 100644 --- a/nodedb-studio/src/services/async_state.rs +++ b/nodedb-studio/src/services/async_state.rs @@ -22,6 +22,11 @@ impl IsEmpty for Vec { /// states. Wired views call `from_value` directly (cloning the resource value /// out of its guard — `StudioError` is `Clone`) and then drive `AsyncView` via /// the accessors below, so no view re-implements the match arms inline. +/// +/// `Clone` allows this to be a Dioxus prop. `PartialEq` is hand-written so we +/// do not depend on `NodeDbError: PartialEq` — errors compare equal by their +/// `Display` output, which is cheap and sufficient for prop diffing. +#[derive(Clone)] pub enum AsyncState { Loading, Empty, @@ -29,6 +34,19 @@ pub enum AsyncState { Error(StudioError), } +impl PartialEq for AsyncState { + fn eq(&self, other: &Self) -> bool { + match (self, other) { + (AsyncState::Loading, AsyncState::Loading) => true, + (AsyncState::Empty, AsyncState::Empty) => true, + (AsyncState::Loaded(a), AsyncState::Loaded(b)) => a == b, + // Compare errors by Display — avoids requiring NodeDbError: PartialEq. + (AsyncState::Error(a), AsyncState::Error(b)) => a.to_string() == b.to_string(), + _ => false, + } + } +} + impl AsyncState { /// Pure mapping from a `use_resource` read (`Option>`): /// None -> Loading (first run not finished) diff --git a/nodedb-studio/src/services/backend.rs b/nodedb-studio/src/services/backend.rs new file mode 100644 index 0000000..0095a14 --- /dev/null +++ b/nodedb-studio/src/services/backend.rs @@ -0,0 +1,13 @@ +//! The single front door. `Backend` composes every per-topic data trait plus the +//! connection/session/notification trait, so the app provides ONE `Rc` +//! and a screen reads the one method it needs without knowing which trait owns it. +//! +//! Adding a new domain trait later means extending this bound and implementing the +//! trait on the mock + stub — additive, never a reshape of existing methods. + +use crate::services::connection_service::ConnectionService; +use crate::services::streams_data::StreamsData; + +pub trait Backend: ConnectionService + StreamsData {} + +impl Backend for T {} diff --git a/nodedb-studio/src/services/connection_service.rs b/nodedb-studio/src/services/connection_service.rs index 9736b5c..f1fab73 100644 --- a/nodedb-studio/src/services/connection_service.rs +++ b/nodedb-studio/src/services/connection_service.rs @@ -6,11 +6,16 @@ //! wraps it here — and async (`use_resource`) is introduced at this boundary, //! not sprinkled through the views. +use std::cell::Cell; +use std::rc::Rc; + use async_trait::async_trait; use crate::data::mock; +use crate::models::cdc::CdcRow; use crate::models::notification::Notification; use crate::services::error::StudioError; +use crate::services::streams_data::{StreamsData, cdc_rows_from_mock}; use crate::state::connection::ActiveConnection; use crate::state::connections_registry::SavedConnection; @@ -30,12 +35,65 @@ pub trait ConnectionService { /// Open a session by saved-connection name. `StudioError::NotConnected` if /// the name is unknown or the connection is offline. async fn connect(&self, name: &str) -> Result; + + /// Mark every notification read. A seam write: the real client persists this + /// server-side; the mock persists it in-process so the unread badge does not + /// revert on reload (POP-03). + async fn mark_all_read(&self) -> Result<(), StudioError>; +} + +/// Drives which result the mock returns, so every screen's four async states are +/// reachable in demos and tests. +// Variants are public API for demos and tests; not all are used in the app binary. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum MockBehavior { + #[default] + Ready, + #[allow(dead_code)] + Empty, + #[allow(dead_code)] + Erroring, } /// Hardcoded implementation used by the skeleton. Data is identical to before; /// the methods are merely `async` now (and resolve instantly). -#[derive(Debug, Clone, Copy, Default)] -pub struct MockConnectionService; +/// +/// `all_read` uses `Rc>` for interior mutability so mark_all_read +/// persists across subsequent `notifications()` calls without blocking the +/// single-threaded Dioxus runtime. `Copy` is intentionally dropped — `Rc` is +/// not `Copy`. +#[derive(Debug, Clone, Default)] +pub struct MockConnectionService { + behavior: MockBehavior, + all_read: Rc>, +} + +// Constructors are public API for demos and tests; not all are used in the app binary. +impl MockConnectionService { + /// Rich fixtures (the default). + pub fn ready() -> Self { + Self { + behavior: MockBehavior::Ready, + all_read: Rc::default(), + } + } + /// Every read returns an empty collection. + #[allow(dead_code)] + pub fn empty() -> Self { + Self { + behavior: MockBehavior::Empty, + all_read: Rc::default(), + } + } + /// Every read fails with a retriable server error. + #[allow(dead_code)] + pub fn erroring() -> Self { + Self { + behavior: MockBehavior::Erroring, + all_read: Rc::default(), + } + } +} #[async_trait(?Send)] impl ConnectionService for MockConnectionService { @@ -44,7 +102,18 @@ impl ConnectionService for MockConnectionService { } async fn notifications(&self) -> Result, StudioError> { - Ok(mock::notifications()) + let mut notifs = mock::notifications(); + if self.all_read.get() { + for n in &mut notifs { + n.unread = false; + } + } + Ok(notifs) + } + + async fn mark_all_read(&self) -> Result<(), StudioError> { + self.all_read.set(true); + Ok(()) } async fn connect(&self, name: &str) -> Result { @@ -56,13 +125,41 @@ impl ConnectionService for MockConnectionService { } } +#[async_trait(?Send)] +impl StreamsData for MockConnectionService { + async fn cdc_feed(&self) -> Result, StudioError> { + match self.behavior { + MockBehavior::Ready => Ok(cdc_rows_from_mock()), + MockBehavior::Empty => Ok(Vec::new()), + MockBehavior::Erroring => Err(StudioError::from( + nodedb_client::NodeDbError::node_unreachable("mock"), + )), + } + } +} + #[cfg(test)] mod tests { use super::*; + use crate::services::async_state::AsyncState; + use crate::services::streams_data::StreamsData; + + #[tokio::test] + async fn mark_all_read_persists_across_reads() { + let svc = MockConnectionService::ready(); + let before = svc.notifications().await.expect("mock infallible"); + assert!(before.iter().any(|n| n.unread), "fixture has unread items"); + svc.mark_all_read().await.expect("mock write infallible"); + let after = svc.notifications().await.expect("mock infallible"); + assert!( + after.iter().all(|n| !n.unread), + "all read after mark_all_read" + ); + } #[tokio::test] async fn mock_notifications_returns_data() { - let svc = MockConnectionService; + let svc = MockConnectionService::ready(); let notifs = svc .notifications() .await @@ -72,7 +169,7 @@ mod tests { #[tokio::test] async fn mock_list_connections_returns_data() { - let svc = MockConnectionService; + let svc = MockConnectionService::ready(); let conns = svc .list_connections() .await @@ -82,7 +179,7 @@ mod tests { #[tokio::test] async fn mock_connect_known_name_returns_session() { - let svc = MockConnectionService; + let svc = MockConnectionService::ready(); // `staging-cluster` is a connectable Online mock connection (data/mock.rs). let session = svc.connect("staging-cluster").await; assert!(session.is_ok()); @@ -90,10 +187,49 @@ mod tests { #[tokio::test] async fn mock_connect_unknown_name_is_not_connected() { - let svc = MockConnectionService; + let svc = MockConnectionService::ready(); assert!(matches!( svc.connect("does-not-exist").await, Err(StudioError::NotConnected) )); } + + #[tokio::test] + async fn cdc_feed_ready_is_loaded() { + let svc = MockConnectionService::ready(); + let state = AsyncState::from_value(Some(svc.cdc_feed().await)); + assert!(matches!(state, AsyncState::Loaded(_))); + } + + #[tokio::test] + async fn cdc_feed_empty_is_empty() { + let svc = MockConnectionService::empty(); + let state = AsyncState::from_value(Some(svc.cdc_feed().await)); + assert!(state.is_empty()); + } + + #[tokio::test] + async fn cdc_feed_erroring_is_retriable_error() { + let svc = MockConnectionService::erroring(); + let state = AsyncState::from_value(Some(svc.cdc_feed().await)); + assert!(state.error_message().is_some()); + assert!(state.is_retriable()); + } + + // Verifies the two-step transition pattern: set Loading, await the read, + // set Loaded from the result. This test has no Dioxus signals/guards (unit-level), + // so it only demonstrates the transition logic; the actual no-guard-across-await + // discipline is exercised by the real streaming view (later task) and enforced + // by the AGENTS.md convention. + #[tokio::test] + async fn delayed_read_transitions_loading_to_loaded_without_guard() { + let svc = MockConnectionService::ready(); + let mut latest: AsyncState> = AsyncState::Loading; + assert!(latest.is_loading()); + // simulate the scheduler interleaving other work before the read resolves + tokio::task::yield_now().await; + let result = svc.cdc_feed().await; + latest = AsyncState::from_value(Some(result)); + assert!(matches!(latest, AsyncState::Loaded(_))); + } } diff --git a/nodedb-studio/src/services/mod.rs b/nodedb-studio/src/services/mod.rs index 4f4863c..3a007ca 100644 --- a/nodedb-studio/src/services/mod.rs +++ b/nodedb-studio/src/services/mod.rs @@ -2,6 +2,8 @@ //! a NodeDB-client-backed impl plugs in here later. pub mod async_state; +pub mod backend; pub mod connection_service; pub mod error; pub mod nodedb_service; +pub mod streams_data; diff --git a/nodedb-studio/src/services/nodedb_service.rs b/nodedb-studio/src/services/nodedb_service.rs index 3e5aace..abfff74 100644 --- a/nodedb-studio/src/services/nodedb_service.rs +++ b/nodedb-studio/src/services/nodedb_service.rs @@ -10,9 +10,11 @@ use async_trait::async_trait; +use crate::models::cdc::CdcRow; use crate::models::notification::Notification; use crate::services::connection_service::ConnectionService; use crate::services::error::StudioError; +use crate::services::streams_data::StreamsData; use crate::state::connection::ActiveConnection; use crate::state::connections_registry::SavedConnection; @@ -37,6 +39,17 @@ impl ConnectionService for NodeDbConnectionService { async fn connect(&self, _name: &str) -> Result { Err(StudioError::NotConnected) } + + async fn mark_all_read(&self) -> Result<(), StudioError> { + Err(StudioError::NotConnected) + } +} + +#[async_trait(?Send)] +impl StreamsData for NodeDbConnectionService { + async fn cdc_feed(&self) -> Result, StudioError> { + Err(StudioError::NotConnected) + } } #[cfg(test)] @@ -60,6 +73,10 @@ mod tests { svc.connect("anything").await, Err(StudioError::NotConnected) )); + assert!(matches!( + svc.mark_all_read().await, + Err(StudioError::NotConnected) + )); } #[test] @@ -68,4 +85,10 @@ mod tests { // exactly as `app.rs` provides it via context. let _s: Rc = Rc::new(NodeDbConnectionService); } + + #[test] + fn stub_is_object_safe_behind_backend() { + use crate::services::backend::Backend; + let _b: Rc = Rc::new(NodeDbConnectionService); + } } diff --git a/nodedb-studio/src/services/streams_data.rs b/nodedb-studio/src/services/streams_data.rs new file mode 100644 index 0000000..49692a1 --- /dev/null +++ b/nodedb-studio/src/services/streams_data.rs @@ -0,0 +1,59 @@ +//! Streams-tier reads at the backend seam. One method per Streams screen's data. +//! Today: CDC. Notify/MV/Topics/Cron methods are added when those screens are built. + +use async_trait::async_trait; + +use crate::models::cdc::{CdcOp, CdcRow}; +use crate::services::error::StudioError; + +#[async_trait(?Send)] +pub trait StreamsData { + /// The CDC change feed, newest first. + async fn cdc_feed(&self) -> Result, StudioError>; +} + +/// Build display rows from the static mock change feed. Lives here (not in the +/// view) so the seam — not the screen — owns the native-doc → display mapping. +pub(crate) fn cdc_rows_from_mock() -> Vec { + use crate::data::mock; + mock::cdc_events() + .into_iter() + .enumerate() + .map(|(i, ev)| { + let mut payload_json = sonic_rs::to_string(&ev.payload).unwrap_or_default(); + if let Some(note) = ev.note { + payload_json.push(' '); + payload_json.push_str(note); + } + let op = match ev.op { + mock::ChangeOp::Insert => CdcOp::Insert, + mock::ChangeOp::Update => CdcOp::Update, + mock::ChangeOp::Delete => CdcOp::Delete, + }; + CdcRow { + id: format!("cdc-{i}"), + time: ev.time.to_string(), + op, + collection: ev.collection.to_string(), + payload_json, + } + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn mock_rows_have_unique_stable_ids() { + let rows = cdc_rows_from_mock(); + assert!(!rows.is_empty()); + let mut ids: Vec<&str> = rows.iter().map(|r| r.id.as_str()).collect(); + let count = ids.len(); + ids.sort_unstable(); + ids.dedup(); + assert_eq!(ids.len(), count, "ids must be unique"); + assert_eq!(rows[0].id, "cdc-0"); + } +} diff --git a/nodedb-studio/src/views/connection_manager.rs b/nodedb-studio/src/views/connection_manager.rs index 9a1596c..12dd77a 100644 --- a/nodedb-studio/src/views/connection_manager.rs +++ b/nodedb-studio/src/views/connection_manager.rs @@ -5,7 +5,7 @@ use std::rc::Rc; use dioxus::prelude::*; -use crate::services::connection_service::ConnectionService; +use crate::services::backend::Backend; use crate::state::connection::ActiveConnection; use crate::state::connections_registry::{ConnStatus, SavedConnection}; use crate::state::ui::ModalKind; @@ -15,7 +15,7 @@ pub fn ConnectionManager() -> Element { let registry = use_context::>>(); let mut active = use_context::>>(); let mut modal = use_context::>>(); - let service = use_context::>(); + let service = use_context::>(); rsx! { div { class: "conn-manager", diff --git a/nodedb-studio/src/views/streams/cdc.rs b/nodedb-studio/src/views/streams/cdc.rs index 3634b17..87e49e5 100644 --- a/nodedb-studio/src/views/streams/cdc.rs +++ b/nodedb-studio/src/views/streams/cdc.rs @@ -1,57 +1,242 @@ -//! Streams · CDC: a live tail of change events. Payloads are native documents -//! (see `data::mock`); each is serialized to JSON here, at the view, for -//! display — the same seam the real Explorer will use on `Value`s from the -//! client. +//! Streams · CDC: a live tail of change events, read through the seam. Split into +//! a fetch wrapper (`StreamsCdc`, owns the async read) and a pure presentational +//! component (`CdcTail`, takes `AsyncState` as input) so the four states are +//! render-testable without a runtime. + +use std::rc::Rc; +use std::time::Duration; use dioxus::prelude::*; -use crate::data::mock; +use crate::components::async_view::AsyncView; +use crate::components::live_tail::LiveTail; +use crate::models::cdc::{CdcOp, CdcRow}; +use crate::services::async_state::AsyncState; +use crate::services::backend::Backend; + +/// Build a synthetic CDC row for the simulated stream. Pure (no clock) so it is +/// unit-testable; the streaming task supplies an incrementing `seq`. +pub fn synthetic_row(seq: u64) -> CdcRow { + let op = match seq % 3 { + 0 => CdcOp::Insert, + 1 => CdcOp::Update, + _ => CdcOp::Delete, + }; + CdcRow { + id: format!("cdc-live-{seq}"), + time: format!("04:23:{:02}.{:03}", seq % 60, (seq * 37) % 1000), + op, + collection: "events".to_string(), + payload_json: format!("{{\"seq\":{seq}}}"), + } +} #[component] pub fn StreamsCdc() -> Element { - // (time, op-label, op-css, collection, payload-json) - let rows: Vec<(&str, &str, &str, &str, String)> = mock::cdc_events() - .into_iter() - .map(|ev| { - let mut payload = sonic_rs::to_string(&ev.payload).unwrap_or_default(); - if let Some(note) = ev.note { - payload.push(' '); - payload.push_str(note); + let backend = use_context::>(); + let mut feed = use_resource(move || { + let backend = backend.clone(); + async move { backend.cdc_feed().await } + }); + + let mut paused = use_signal(|| false); + let mut live_rows = use_signal(Vec::::new); + + use_future(move || async move { + let mut seq: u64 = 0; + loop { + tokio::time::sleep(Duration::from_millis(1200)).await; + if *paused.peek() { + continue; } - (ev.time, ev.op.label(), ev.op.css(), ev.collection, payload) - }) - .collect(); + let row = synthetic_row(seq); // built before touching the signal + seq += 1; + let mut rows = live_rows.write(); + rows.insert(0, row); + rows.truncate(200); // bounded buffer + // guard `rows` drops here, before the next .await + } + }); + + // Map the resource read into the canonical four-state value (clone out of the + // guard — never hold it across an await; there is none here). + let base = AsyncState::from_value(feed.read().clone()); + let state = base.project(|fetched: &Vec| { + let mut merged = live_rows.read().clone(); // clone out of the guard immediately + merged.extend(fetched.iter().cloned()); + merged + }); + + rsx! { + CdcTail { + state, + on_retry: move |_| feed.restart(), + paused: *paused.read(), + on_toggle_pause: move |_| { + let now = *paused.peek(); + paused.set(!now); + }, + } + } +} + +#[derive(Props, Clone, PartialEq)] +pub struct CdcTailProps { + pub state: AsyncState>, + #[props(default)] + pub on_retry: EventHandler<()>, + /// Whether the live stream is currently paused. Drives the button label. + #[props(default)] + pub paused: bool, + /// Called when the user clicks the Pause/Resume button. + #[props(default)] + pub on_toggle_pause: EventHandler<()>, +} + +#[component] +pub fn CdcTail(props: CdcTailProps) -> Element { + let state = &props.state; + let rows = state.loaded().cloned().unwrap_or_default(); + let pause_label = if props.paused { "Resume" } else { "Pause" }; rsx! { - div { class: "live-tail", - div { class: "tail-toolbar", - strong { style: "font-size:13px;", "events_cdc" } - span { class: "pill ok", span { class: "dot" } "live · 2,103 ev/s" } - div { style: "margin-left:auto; display:flex; gap:6px; align-items:center;", - input { - placeholder: "filter: type=signup", - style: "padding:4px 8px; background: var(--bg-primary); border: 0.5px solid var(--border-mid); border-radius: 4px; font-family: var(--font-mono); font-size: 11px;", + AsyncView { + loading: state.is_loading(), + empty: state.is_empty(), + error: state.error_message(), + retriable: state.is_retriable(), + on_retry: move |_| props.on_retry.call(()), + empty_message: "No change events.".to_string(), + } + if state.loaded().is_some() { + LiveTail { + toolbar: rsx! { + strong { style: "font-size:13px;", "events_cdc" } + span { class: "pill ok", span { class: "dot" } "live · 2,103 ev/s" } + div { style: "margin-left:auto; display:flex; gap:6px; align-items:center;", + input { + placeholder: "filter: type=signup", + style: "padding:4px 8px; background: var(--bg-primary); border: 0.5px solid var(--border-mid); border-radius: 4px; font-family: var(--font-mono); font-size: 11px;", + } + button { + class: "btn small", + onclick: move |_| props.on_toggle_pause.call(()), + "{pause_label}" + } + button { class: "btn small", "⇣ Export" } } - button { class: "btn small", "Pause" } - button { class: "btn small", "⇣ Export" } - } - } - div { class: "tail-body", - for r in rows { - div { class: "tail-row", - span { class: "time", "{r.0}" } - span { class: "op {r.2}", "{r.1}" } - span { class: "coll", "{r.3}" } - span { class: "payload", "{r.4}" } + }, + body: rsx! { + for row in rows.iter() { + div { key: "{row.id}", class: "tail-row", + span { class: "time", "{row.time}" } + span { class: "op {row.op.css()}", "{row.op.label()}" } + span { class: "coll", "{row.collection}" } + span { class: "payload", "{row.payload_json}" } + } } - } + }, + footer: rsx! { + span { class: "tail-pulse" } + span { "following tail" } + span { "buffer: {rows.len()} / 5,000" } + span { "lag from leader: 12 ms" } + span { style: "margin-left:auto;", "columns: time, op, collection, payload" } + }, + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::models::cdc::CdcOp; + use crate::services::error::StudioError; + + fn render(app: fn() -> Element) -> String { + let mut dom = VirtualDom::new(app); + dom.rebuild_in_place(); + dioxus_ssr::render(&dom) + } + + fn sample_rows() -> Vec { + vec![CdcRow { + id: "cdc-0".to_string(), + time: "04:23:18.041".to_string(), + op: CdcOp::Insert, + collection: "events".to_string(), + payload_json: "{\"_id\":\"evt\"}".to_string(), + }] + } + + fn app_loading() -> Element { + rsx! { CdcTail { state: AsyncState::Loading } } + } + + fn app_empty() -> Element { + rsx! { + CdcTail { + state: AsyncState::from_value(Some(Ok(Vec::::new()))), } - div { class: "tail-footer", - span { class: "tail-pulse" } - span { "following tail" } - span { "buffer: 9 / 5,000" } - span { "lag from leader: 12 ms" } - span { style: "margin-left:auto;", "columns: time, op, collection, payload" } + } + } + + fn app_error() -> Element { + rsx! { CdcTail { state: AsyncState::Error(StudioError::NotConnected) } } + } + + fn app_ready() -> Element { + rsx! { + CdcTail { + state: AsyncState::from_value(Some(Ok(sample_rows()))), } } } + + #[test] + fn loading_state_renders_spinner() { + let html = render(app_loading); + assert!(html.contains("async-loading")); + } + + #[test] + fn empty_state_renders_message() { + let html = render(app_empty); + assert!(html.contains("async-empty")); + assert!(html.contains("No change events")); + } + + #[test] + fn error_state_renders_error() { + let html = render(app_error); + assert!(html.contains("async-error")); + } + + #[test] + fn ready_state_renders_keyed_rows() { + let html = render(app_ready); + assert!(html.contains("live-tail")); + assert!(html.contains("INSERT")); + assert!(html.contains("events")); + assert!(html.contains("buffer: 1 / 5,000")); + } + + #[test] + fn synthetic_row_is_stable_and_keyed_by_seq() { + let a = synthetic_row(7); + assert_eq!(a.id, "cdc-live-7"); + assert!(!a.time.is_empty()); + // distinct seq -> distinct id (keeps list keys stable & unique) + assert_ne!(synthetic_row(7).id, synthetic_row(8).id); + } + + #[tokio::test] + async fn stream_step_appends_without_guard_panic() { + let mut rows: Vec = Vec::new(); + // mimic one loop iteration: await first, then mutate — never a guard across await + tokio::time::sleep(Duration::from_millis(1)).await; + rows.insert(0, synthetic_row(0)); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].id, "cdc-live-0"); + } }