From b10326b2427cf3a34bd87c5dac2a2de14fb64a23 Mon Sep 17 00:00:00 2001
From: Mark Kalsbeek <Mark@Kalsbeek.dev>
Date: Sun, 8 Mar 2026 20:10:15 +0100
Subject: [PATCH] bodged refactor

---
 Cargo.lock                 | 136 +++++++++++++++++
 Cargo.toml                 |   6 +
 examples/todo/about.rs     |  23 +--
 examples/todo/app.rs       | 210 +++++++--------------------
 examples/todo/main.rs      |  70 +++++++--
 examples/todo/todo_list.rs |  62 ++------
 src/connection.rs          |  67 ++++++---
 src/context.rs             |  86 +++++++----
 src/lib.rs                 |   4 +-
 src/query.rs               |  47 +++---
 src/server.rs              | 289 +++++++++++++++++++++++++++++++++----
 src/store.rs               | 140 ++++++++++++++++++
 12 files changed, 801 insertions(+), 339 deletions(-)
 create mode 100644 src/store.rs

diff --git a/Cargo.lock b/Cargo.lock
index 6a0e83e..cfaa446 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -8,6 +8,15 @@ version = "0.2.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
 
+[[package]]
+name = "android_system_properties"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "anyhow"
 version = "1.0.102"
@@ -177,6 +186,19 @@ version = "1.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801"
 
+[[package]]
+name = "chrono"
+version = "0.4.44"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0"
+dependencies = [
+ "iana-time-zone",
+ "js-sys",
+ "num-traits",
+ "wasm-bindgen",
+ "windows-link",
+]
+
 [[package]]
 name = "concurrent-queue"
 version = "2.5.0"
@@ -192,6 +214,12 @@ version = "0.9.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
 
+[[package]]
+name = "core-foundation-sys"
+version = "0.8.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
+
 [[package]]
 name = "cpufeatures"
 version = "0.2.17"
@@ -374,6 +402,21 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "futures"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d"
+dependencies = [
+ "futures-channel",
+ "futures-core",
+ "futures-executor",
+ "futures-io",
+ "futures-sink",
+ "futures-task",
+ "futures-util",
+]
+
 [[package]]
 name = "futures-channel"
 version = "0.3.32"
@@ -418,6 +461,17 @@ version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718"
 
+[[package]]
+name = "futures-macro"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "futures-sink"
 version = "0.3.32"
@@ -436,8 +490,10 @@ version = "0.3.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6"
 dependencies = [
+ "futures-channel",
  "futures-core",
  "futures-io",
+ "futures-macro",
  "futures-sink",
  "futures-task",
  "memchr",
@@ -631,6 +687,30 @@ dependencies = [
  "tower-service",
 ]
 
+[[package]]
+name = "iana-time-zone"
+version = "0.1.65"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470"
+dependencies = [
+ "android_system_properties",
+ "core-foundation-sys",
+ "iana-time-zone-haiku",
+ "js-sys",
+ "log",
+ "wasm-bindgen",
+ "windows-core",
+]
+
+[[package]]
+name = "iana-time-zone-haiku"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f"
+dependencies = [
+ "cc",
+]
+
 [[package]]
 name = "icu_collections"
 version = "2.1.1"
@@ -829,13 +909,16 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "axum",
+ "chrono",
  "dashmap",
+ "futures",
  "maud",
  "serde",
  "serde_json",
  "sqlx",
  "tokio",
  "tokio-stream",
+ "tracing",
  "uuid",
 ]
 
@@ -1966,12 +2049,65 @@ dependencies = [
  "wasite",
 ]
 
+[[package]]
+name = "windows-core"
+version = "0.62.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb"
+dependencies = [
+ "windows-implement",
+ "windows-interface",
+ "windows-link",
+ "windows-result",
+ "windows-strings",
+]
+
+[[package]]
+name = "windows-implement"
+version = "0.60.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "windows-interface"
+version = "0.59.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "windows-link"
 version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
 
+[[package]]
+name = "windows-result"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5"
+dependencies = [
+ "windows-link",
+]
+
+[[package]]
+name = "windows-strings"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091"
+dependencies = [
+ "windows-link",
+]
+
 [[package]]
 name = "windows-sys"
 version = "0.48.0"
diff --git a/Cargo.toml b/Cargo.toml
index 242bf96..b511823 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -16,3 +16,9 @@ uuid = { version = "1", features = ["v4", "serde"] }
 dashmap = "6"
 anyhow = "1"
 tokio-stream = "0.1"
+tracing = "0.1.44"
+futures = "0.3.32"
+
+
+[dev-dependencies]
+chrono = { version = "0.4", features = ["clock"] }
\ No newline at end of file
diff --git a/examples/todo/about.rs b/examples/todo/about.rs
index ff996f3..fffe1a7 100644
--- a/examples/todo/about.rs
+++ b/examples/todo/about.rs
@@ -1,14 +1,6 @@
 use livevue_rs::RenderContext;
 use maud::{html, Markup};
 
-/// Second page component: a simple "About" page.
-///
-/// This exists specifically to exercise:
-/// - The `tree` signal page routing (resolve_root match arm)
-/// - SPA-like navigation via tree signal mutation
-/// - Round-trip: TodoList → About → TodoList without full page reload
-///
-/// In a real app this might show user info, settings, etc.
 pub async fn about(_cx: &mut RenderContext) -> anyhow::Result<Markup> {
     Ok(html! {
         div.about-page {
@@ -16,7 +8,7 @@ pub async fn about(_cx: &mut RenderContext) -> anyhow::Result<Markup> {
 
             nav {
                 button.nav-link
-                    data-on:click="$tree = {page: 'TodoList'}; @post('/render')" {
+                    data-on:click="$tree = {page: 'TodoList'}" {
                     "Todos"
                 }
                 " | "
@@ -34,14 +26,15 @@ pub async fn about(_cx: &mut RenderContext) -> anyhow::Result<Markup> {
                     strong { "full rerender is cheap" }
                     ", so skip diffing, skip partial updates, skip client-side state management."
                 }
-                h2 { "v0 Status" }
+                h2 { "Architecture" }
                 ul {
-                    li { "Single process, single node" }
-                    li { "SQLite for storage" }
-                    li { "Subscription keys collected (Phase 2 seam)" }
-                    li { "No cache, no auth, no WAL consumer" }
+                    li { "Actions post signals → 202, no response body" }
+                    li { "Rerenders come down the SSE pipe from the fanout task" }
+                    li { "KV store writes auto-publish broadcast invalidation events" }
+                    li { "SubscriptionRegistry maps CacheKeys → connections" }
+                    li { "Clock ticks via " code { "store.kv.set(\"clock\", ...)" } " every second" }
                 }
             }
         }
     })
-}
+}
\ No newline at end of file
diff --git a/examples/todo/app.rs b/examples/todo/app.rs
index c0afc26..76246fe 100644
--- a/examples/todo/app.rs
+++ b/examples/todo/app.rs
@@ -1,6 +1,6 @@
 use livevue_rs::{
     AppState, CacheKey, ComponentTree, RenderContext, SharedState, SignalStore,
-    format_patch_elements, sse_handler,
+    ingest_signals, sse_handler,
 };
 
 use axum::extract::{Query as AxumQuery, State};
@@ -19,7 +19,6 @@ pub fn router(state: SharedState) -> Router {
         .route("/", get(initial_page_load))
         .route("/about", get(initial_about_load))
         .route("/sse", get(sse_handler))
-        .route("/render", post(render_handler))
         .route("/action/add_todo", post(action_add_todo))
         .route("/action/toggle_todo", post(action_toggle_todo))
         .route("/action/delete_todo", post(action_delete_todo))
@@ -30,14 +29,7 @@ pub fn router(state: SharedState) -> Router {
 // Document shell
 // ---------------------------------------------------------------------------
 
-/// Full HTML document for initial page load.
-///
-/// DataStar v1 contract:
-/// - `data-signals` — initialize signals (JS object literal syntax)
-/// - `data-on-signal-patch__debounce.300ms` — fires when any signal is patched,
-///   debounced 300ms, posts all non-_ signals as JSON body to /render
-/// - `data-init` — fires when element enters DOM, opens long-lived SSE pipe
-fn document_shell(conn_id: Uuid, initial_signals: &str, inner_html: &str) -> String {
+fn document_shell(conn_id: Uuid) -> String {
     format!(
         r#"<!DOCTYPE html>
 <html lang="en">
@@ -45,47 +37,42 @@ fn document_shell(conn_id: Uuid, initial_signals: &str, inner_html: &str) -> Str
     <meta charset="utf-8">
     <meta name="viewport" content="width=device-width, initial-scale=1">
     <title>LiveVue.rs — Todo</title>
-    <script type="module" src="https://cdn.jsdelivr.net/gh/starfederation/datastar@v1.0.0-RC.8/bundles/datastar.js"></script>
+    <script type="module" src="https://cdn.jsdelivr.net/gh/starfederation/datastar@1.0.0-RC.8/bundles/datastar.js"></script>
     <style>
         {css}
     </style>
 </head>
-<body>
-    {app}
+<body 
+    data-init="@get('/sse?conn={conn}')">
+    <div id="app">
+    <div/>
 </body>
 </html>"#,
         css = INLINE_CSS,
-        app = app_wrapper(conn_id, initial_signals, inner_html),
+        conn = conn_id,
     )
 }
 
-/// The `<div id="app">` wrapper with DataStar bindings.
-///
-/// Extracted separately so `/render` can return just this element as
-/// `text/html` — DataStar morphs it into the existing DOM by id.
-fn app_wrapper(conn_id: Uuid, signals: &str, inner_html: &str) -> String {
+/// Used by the fanout task to wrap rendered inner HTML for morphing.
+pub fn app_wrapper(conn_id: Uuid, signals: &str, inner_html: &str) -> String {
     format!(
         r#"<div id="app"
-         data-signals='{signals}'
-         data-on-signal-patch="@post('/render')" 
-         data-on-init="@get('/sse?conn={conn}')">
-        {inner}
+            data-signals='{signals}'>
+            {inner}
         <pre data-json-signals></pre>
-    </div>"#, 
-        // Could __debounce.300ms on the signal patch
+        </div>"#,
         signals = signals,
-        conn = conn_id,
         inner = inner_html,
     )
 }
 
-/// Minimal CSS for the todo app.
 const INLINE_CSS: &str = r#"
     body { font-family: system-ui, sans-serif; max-width: 600px; margin: 2rem auto; padding: 0 1rem; background: #fafafa; color: #333; }
     h1 { color: #2563eb; }
     nav { margin-bottom: 1.5rem; }
     .nav-link { background: none; border: none; color: #2563eb; cursor: pointer; text-decoration: underline; font-size: inherit; padding: 0; }
     .nav-current { font-weight: bold; }
+    .clock { font-variant-numeric: tabular-nums; color: #555; font-size: 0.95rem; margin-bottom: 1rem; }
     .add-form { display: flex; gap: 0.5rem; margin-bottom: 1rem; }
     .add-form input { flex: 1; padding: 0.5rem; border: 1px solid #ccc; border-radius: 4px; }
     .add-form button { padding: 0.5rem 1rem; background: #2563eb; color: white; border: none; border-radius: 4px; cursor: pointer; }
@@ -101,42 +88,18 @@ const INLINE_CSS: &str = r#"
 "#;
 
 // ---------------------------------------------------------------------------
-// Page resolution
+// Page resolution + render pipeline
 // ---------------------------------------------------------------------------
 
-async fn resolve_and_render(cx: &mut RenderContext) -> anyhow::Result<String> {
+pub async fn resolve_and_render(cx: &mut RenderContext) -> anyhow::Result<String> {
     let markup = match cx.tree.page.as_str() {
         "TodoList" => crate::todo_list::todo_list(cx).await?,
         "About" => crate::about::about(cx).await?,
-        other => maud::html! {
-            div {
-                h1 { "404 — Unknown page" }
-                p { "No component registered for: " (other) }
-                button.nav-link
-                    data-on-click="$tree = {page: 'TodoList'}; @post('/render')" {
-                    "Go to Todos"
-                }
-            }
-        },
+        other  => crate::todo_list::todo_list(cx).await?
     };
     Ok(markup.into_string())
 }
 
-/// Shared render pipeline. Returns `(rendered_inner_html, subscription_keys, signals_json)`.
-async fn render_page(
-    db: &sqlx::SqlitePool,
-    conn_id: Uuid,
-    signals: SignalStore,
-    tree: ComponentTree,
-) -> anyhow::Result<(String, Vec<CacheKey>, String)> {
-    let signals_json = signals.to_json_string();
-    let mut cx = RenderContext::new(conn_id, signals, tree, db.clone());
-    let html = resolve_and_render(&mut cx).await?;
-    let keys = cx.take_subscription_keys();
-    Ok((html, keys, signals_json))
-}
-
-/// Extract conn_id and tree from a signal store.
 fn extract_conn_and_tree(signals: &SignalStore) -> (Uuid, ComponentTree) {
     let conn_id = signals
         .get("connId")
@@ -153,133 +116,65 @@ fn extract_conn_and_tree(signals: &SignalStore) -> (Uuid, ComponentTree) {
 }
 
 // ---------------------------------------------------------------------------
-// Handlers
+// Initial page load handlers
 // ---------------------------------------------------------------------------
 
-/// `GET /` — Initial page load for the todo list.
 async fn initial_page_load(State(state): State<SharedState>) -> impl IntoResponse {
     let conn_id = Uuid::new_v4();
 
-    let tree = ComponentTree {
-        page: "TodoList".into(),
-        ..Default::default()
-    };
+    // Seed the initial signal state so the SSE handler can trigger the first
+    // render as soon as the pipe opens, without a client round-trip.
+    let tree = ComponentTree { page: "TodoList".into(), ..Default::default() };
     let signals = SignalStore::from_pairs(vec![
         ("connId".into(), serde_json::json!(conn_id.to_string())),
         ("tree".into(), serde_json::to_value(&tree).unwrap()),
         ("newTodo".into(), serde_json::json!("")),
     ]);
+    state.connections.update_signals(&conn_id, signals.to_json_string());
 
-    let initial_signals_attr = signals.to_data_signals_attr();
-
-    match render_page(&state.db, conn_id, signals, tree).await {
-        Ok((html, _keys, _signals_json)) => {
-            Html(document_shell(conn_id, &initial_signals_attr, &html))
-        }
-        Err(e) => Html(document_shell(
-            conn_id,
-            &initial_signals_attr,
-            &format!("<p>Render error: {}</p>", e),
-        )),
-    }
+    Html(document_shell(conn_id))
 }
 
-/// `GET /about` — Initial page load for the about page.
 async fn initial_about_load(State(state): State<SharedState>) -> impl IntoResponse {
     let conn_id = Uuid::new_v4();
-
-    let tree = ComponentTree {
-        page: "About".into(),
-        ..Default::default()
-    };
+    let tree = ComponentTree { page: "About".into(), ..Default::default() };
     let signals = SignalStore::from_pairs(vec![
         ("connId".into(), serde_json::json!(conn_id.to_string())),
         ("tree".into(), serde_json::to_value(&tree).unwrap()),
     ]);
-
-    let initial_signals_attr = signals.to_data_signals_attr();
-
-    match render_page(&state.db, conn_id, signals, tree).await {
-        Ok((html, _keys, _signals_json)) => {
-            Html(document_shell(conn_id, &initial_signals_attr, &html))
-        }
-        Err(e) => Html(document_shell(
-            conn_id,
-            &initial_signals_attr,
-            &format!("<p>Render error: {}</p>", e),
-        )),
-    }
+    state.connections.update_signals(&conn_id, signals.to_json_string());
+    Html(document_shell(conn_id))
 }
 
-/// `POST /render` — Signal-triggered rerender.
-///
-/// DataStar sends all non-`_` signals as JSON body on every signal patch.
-/// Returns `text/html` with the full `<div id="app">` wrapper — DataStar
-/// morphs it into the existing DOM by matching the element id.
-async fn render_handler(
-    State(state): State<SharedState>,
-    Json(body): Json<serde_json::Value>,
-) -> impl IntoResponse {
-    let signals = SignalStore::from_json(body);
-    let (conn_id, tree) = extract_conn_and_tree(&signals);
+// ---------------------------------------------------------------------------
+// Action handlers
+// ---------------------------------------------------------------------------
 
-    match render_page(&state.db, conn_id, signals.clone(), tree).await {
-        Ok((html, keys, signals_json)) => {
-            state.connections.update_signals(&conn_id, signals_json.clone());
-            state.subscriptions.update(conn_id, keys);
-            Html(app_wrapper(conn_id, &signals_json, &html))
-        }
-        Err(e) => Html(app_wrapper(
-            conn_id,
-            &signals.to_json_string(),
-            &format!("<p>Render error: {}</p>", e),
-        )),
-    }
-}
-
-/// Query params for action endpoints that need an entity id.
 #[derive(serde::Deserialize)]
 struct ActionIdParams {
     id: i64,
 }
 
-/// Helper: run a mutation, push the rerendered page down the SSE pipe, return 200.
-///
-/// Actions do not return a body — the result comes down the SSE pipe as a
-/// `datastar-patch-elements` event.
-async fn action_then_push(
+/// Ingest signals, run a mutation, then publish an invalidation event so the
+/// fanout task handles the rerender. Returns 202 Accepted.
+async fn action_then_publish(
     state: &AppState,
-    signals: SignalStore,
+    body: serde_json::Value,
     mutation: impl std::future::Future<Output = anyhow::Result<()>> + Send,
+    invalidation_key: CacheKey,
 ) -> StatusCode {
-    let (conn_id, tree) = extract_conn_and_tree(&signals);
+    let signals = ingest_signals(state, extract_conn_and_tree(&SignalStore::from_json(body.clone())).0, body);
+    let _ = signals; // signals are stored; mutation drives the rerender
 
     if let Err(e) = mutation.await {
-        eprintln!("Action error for {}: {}", conn_id, e);
+        eprintln!("Action mutation error: {}", e);
         return StatusCode::INTERNAL_SERVER_ERROR;
     }
 
-    match render_page(&state.db, conn_id, signals, tree).await {
-        Ok((html, keys, signals_json)) => {
-            state.connections.update_signals(&conn_id, signals_json.clone());
-            state.subscriptions.update(conn_id, keys);
-
-            if let Some(sender) = state.connections.get_sender(&conn_id) {
-                let event = format_patch_elements(&app_wrapper(conn_id, &signals_json, &html));
-                // Non-blocking: if the pipe is closed the cleanup task handles it
-                let _ = sender.try_send(event);
-            }
-
-            StatusCode::OK
-        }
-        Err(e) => {
-            eprintln!("Render error for {}: {}", conn_id, e);
-            StatusCode::INTERNAL_SERVER_ERROR
-        }
-    }
+    state.store.publish(invalidation_key);
+    StatusCode::ACCEPTED
 }
 
-/// `POST /action/add_todo` — Insert a new todo, push rerender down SSE pipe.
 async fn action_add_todo(
     State(state): State<SharedState>,
     Json(body): Json<serde_json::Value>,
@@ -290,55 +185,52 @@ async fn action_add_todo(
         .unwrap_or("")
         .trim()
         .to_string();
+    let db = state.store.db.clone();
 
-    let signals = SignalStore::from_json(body);
-    let db = state.db.clone();
-
-    action_then_push(
+    action_then_publish(
         &state,
-        signals,
+        body,
         async move {
             if !title.is_empty() {
                 crate::queries::create_todo(&db, &title).await?;
             }
             Ok(())
         },
+        CacheKey::Table { table: "todos" },
     )
     .await
 }
 
-/// `POST /action/toggle_todo?id=<i64>` — Toggle completion, push rerender.
 async fn action_toggle_todo(
     State(state): State<SharedState>,
     AxumQuery(params): AxumQuery<ActionIdParams>,
     Json(body): Json<serde_json::Value>,
 ) -> impl IntoResponse {
-    let signals = SignalStore::from_json(body);
-    let db = state.db.clone();
+    let db = state.store.db.clone();
     let id = params.id;
 
-    action_then_push(
+    action_then_publish(
         &state,
-        signals,
+        body,
         async move { crate::queries::toggle_todo(&db, id).await },
+        CacheKey::Table { table: "todos" },
     )
     .await
 }
 
-/// `POST /action/delete_todo?id=<i64>` — Delete a todo, push rerender.
 async fn action_delete_todo(
     State(state): State<SharedState>,
     AxumQuery(params): AxumQuery<ActionIdParams>,
     Json(body): Json<serde_json::Value>,
 ) -> impl IntoResponse {
-    let signals = SignalStore::from_json(body);
-    let db = state.db.clone();
+    let db = state.store.db.clone();
     let id = params.id;
 
-    action_then_push(
+    action_then_publish(
         &state,
-        signals,
+        body,
         async move { crate::queries::delete_todo(&db, id).await },
+        CacheKey::Table { table: "todos" },
     )
     .await
 }
\ No newline at end of file
diff --git a/examples/todo/main.rs b/examples/todo/main.rs
index d758d8c..83063c6 100644
--- a/examples/todo/main.rs
+++ b/examples/todo/main.rs
@@ -3,8 +3,10 @@ mod queries;
 mod todo_list;
 mod about;
 
-use livevue_rs::{AppState, ConnectionManager, SubscriptionRegistry};
+use livevue_rs::{AppState, CacheKey, RenderContext, Store, spawn_fanout};
+use crate::app::{app_wrapper, resolve_and_render};
 use std::sync::Arc;
+use std::time::Duration;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -12,10 +14,8 @@ async fn main() -> anyhow::Result<()> {
     // Database setup
     // -----------------------------------------------------------------------
 
-    // SQLite with create-if-missing. In-memory `:memory:` also works for dev.
     let db = sqlx::SqlitePool::connect("sqlite:todo.db?mode=rwc").await?;
 
-    // Run migrations inline (no migration files needed for v0).
     sqlx::query(
         "CREATE TABLE IF NOT EXISTS todos (
             id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -26,7 +26,6 @@ async fn main() -> anyhow::Result<()> {
     .execute(&db)
     .await?;
 
-    // Seed a few todos if the table is empty (nice for first run)
     let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM todos")
         .fetch_one(&db)
         .await?;
@@ -39,14 +38,65 @@ async fn main() -> anyhow::Result<()> {
         }
     }
 
+    // -----------------------------------------------------------------------
+    // Store
+    // -----------------------------------------------------------------------
+
+    let store = Store::new(db);
+
+    // -----------------------------------------------------------------------
+    // Render function
+    // -----------------------------------------------------------------------
+
+    // The render_fn is the app's half of the framework contract. It receives
+    // a RenderContext (with signals, tree, store access) and returns the
+    // rendered inner HTML plus the accumulated subscription keys.
+    //
+    // The fanout task calls this for every server-initiated rerender.
+    let render_fn: livevue_rs::RenderFn = Arc::new(|mut cx: RenderContext| {
+        Box::pin(async move {
+            println!("Render Called");
+            let signals_json = cx.signals.to_json_string();
+            let conn_id = cx.connection_id;
+            let html = resolve_and_render(&mut cx).await?;
+            let keys = cx.take_subscription_keys();
+            // Wrap in app_wrapper so DataStar morphs the full #app element.
+            // Signals are re-emitted from the stored json so the client stays
+            // in sync even on server-initiated rerenders.
+            let wrapped = app_wrapper(conn_id, &signals_json, &html);
+            Ok((wrapped, keys))
+        })
+    });
+
     // -----------------------------------------------------------------------
     // Application state
     // -----------------------------------------------------------------------
 
-    let state = Arc::new(AppState {
-        db,
-        connections: ConnectionManager::new(),
-        subscriptions: SubscriptionRegistry::new(),
+    let state = AppState::new(store.clone(), render_fn);
+
+    // -----------------------------------------------------------------------
+    // Fanout task
+    // -----------------------------------------------------------------------
+
+    spawn_fanout(state.clone());
+
+    // -----------------------------------------------------------------------
+    // Clock ticker
+    //
+    // Publishes CacheKey::Channel { key: "clock" } every second. Any
+    // connection whose last render called cx.kv_run("clock") will be
+    // scheduled for a rerender by the fanout task.
+    // -----------------------------------------------------------------------
+
+    let clock_store = store.clone();
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(Duration::from_secs(1));
+        loop {
+            interval.tick().await;
+            let now = chrono::Utc::now().format("%H:%M:%S UTC").to_string();
+            // set() writes the value and publishes the invalidation event.
+            clock_store.kv.set("clock", serde_json::json!(now));
+        }
     });
 
     // -----------------------------------------------------------------------
@@ -54,11 +104,9 @@ async fn main() -> anyhow::Result<()> {
     // -----------------------------------------------------------------------
 
     let app = app::router(state);
-
     let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await?;
     println!("Todo example listening on http://localhost:3000");
-
     axum::serve(listener, app).await?;
 
     Ok(())
-}
+}
\ No newline at end of file
diff --git a/examples/todo/todo_list.rs b/examples/todo/todo_list.rs
index 13fee11..c975b15 100644
--- a/examples/todo/todo_list.rs
+++ b/examples/todo/todo_list.rs
@@ -2,24 +2,17 @@ use crate::queries::{ListTodos, Todo};
 use livevue_rs::{RenderContext, global, mangle};
 use maud::{html, Markup};
 
-/// Root page component: renders the full todo list with add form.
-///
-/// Reads signals:
-/// - `newTodo` (global, server-relevant): the text input value
-///
-/// Exercises:
-/// - `cx.run()` for data fetching
-/// - `cx.signal()` with `global()` scoping + `Signal::val` in expressions
-/// - `cx.signal()` with `mangle().client_only()` scoping (in todo_item)
-/// - DataStar `data-bind`, `data-on:click`, `data-attr` attributes
-/// - `@post()` actions for mutations
 pub async fn todo_list(cx: &mut RenderContext) -> anyhow::Result<Markup> {
-    // Fetch all todos
+    // Fetch all todos (registers CacheKey::Table { table: "todos" })
     let todos = cx.run(ListTodos).await?;
 
-    // Global signal for the "new todo" input. This call does two things:
-    // 1. Records "newTodo" as a touched signal (for demangling tracking)
-    // 2. Returns a Signal struct whose .val is "$newTodo" for expressions
+    // Read the clock from KV (registers CacheKey::Channel { key: "clock" })
+    let clock = cx
+        .kv_run("clock")
+        .await
+        .and_then(|v| v.as_str().map(|s| s.to_string()))
+        .unwrap_or_else(|| "—".to_string());
+
     let new_todo = cx.signal("newTodo", global());
 
     let todo_count = todos.len();
@@ -29,21 +22,17 @@ pub async fn todo_list(cx: &mut RenderContext) -> anyhow::Result<Markup> {
         div.todo-app {
             h1 { "LiveVue.rs — Todo List" }
 
-            // Navigation link to About page (exercises tree signal + page routing)
             nav {
                 span.nav-current { "Todos" }
                 " | "
                 button.nav-link
-                    data-on:click="$tree = {page: 'About'};" {
+                    data-on:click="$tree = {page: 'About'}" {
                     "About"
                 }
             }
 
-            // Add todo form.
-            // data-bind="newTodo" binds the input to the $newTodo signal.
-            // Value syntax is used instead of data-bind-newTodo because HTML
-            // normalises attribute names to lowercase, which would corrupt the
-            // camelCase signal name (newTodo → newtodo).
+            p.clock { "🕐 " (clock) }
+
             div.add-form {
                 input
                     type="text"
@@ -57,12 +46,10 @@ pub async fn todo_list(cx: &mut RenderContext) -> anyhow::Result<Markup> {
                 }
             }
 
-            // Stats
             p.stats {
                 (format!("{} items, {} completed", todo_count, done_count))
             }
 
-            // Todo items
             ul.todo-list {
                 @for todo in &todos {
                     (todo_item(cx, todo))
@@ -76,21 +63,9 @@ pub async fn todo_list(cx: &mut RenderContext) -> anyhow::Result<Markup> {
     })
 }
 
-/// Renders a single todo item.
-///
-/// This is a static child component — called directly from `todo_list`,
-/// not resolved through the component registry.
-///
-/// Demonstrates per-instance, client-only signal via `mangle().client_only()`.
 fn todo_item(cx: &mut RenderContext, todo: &Todo) -> Markup {
-    // Per-instance, client-only signal for edit mode.
-    // Not used in v0 rendering, but demonstrates the mangled signal API
-    // and ensures it's tracked as touched. Produces: "_editing__<id>"
     let _editing = cx.signal("editing", mangle(todo.id).client_only());
 
-    // Build class string server-side. Since we do a full rerender on every
-    // state change, there's no need for data-class here — the server knows
-    // whether the todo is completed.
     let title_class = if todo.completed {
         "todo-title completed"
     } else {
@@ -99,25 +74,16 @@ fn todo_item(cx: &mut RenderContext, todo: &Todo) -> Markup {
 
     html! {
         li.todo-item id=(format!("todo-{}", todo.id)) {
-            // Checkbox to toggle completion.
-            // `checked[expr]` is maud's syntax for conditional boolean attributes.
             input
                 type="checkbox"
                 checked[todo.completed]
                 data-on:click=(format!("@post('/action/toggle_todo?id={}')", todo.id))
                 ;
-
-            // Title with strikethrough when completed (class set server-side)
-            span class=(title_class) {
-                (todo.title)
-            }
-
-            // Delete button
+            span class=(title_class) { (todo.title) }
             button.delete-btn
-                data-on:click=(format!("@post('/action/delete_todo?id={}')", todo.id))
-                {
+                data-on:click=(format!("@post('/action/delete_todo?id={}')", todo.id)) {
                 "×"
             }
         }
     }
-}
+}
\ No newline at end of file
diff --git a/src/connection.rs b/src/connection.rs
index 23f739e..5c48f88 100644
--- a/src/connection.rs
+++ b/src/connection.rs
@@ -1,4 +1,5 @@
 use crate::query::CacheKey;
+use axum::response::sse::Event;
 use dashmap::DashMap;
 use std::collections::HashSet;
 use tokio::sync::mpsc;
@@ -10,13 +11,17 @@ use uuid::Uuid;
 
 /// Per-connection state stored in the manager.
 pub struct ConnectionState {
-    /// Sender half of the SSE channel. Messages written here get forwarded
-    /// to the client's long-lived SSE pipe (`GET /sse`).
-    pub sse_sender: mpsc::Sender<String>,
+    /// Sender half of the per-connection SSE channel.
+    ///
+    /// The fanout task sends fully-constructed axum `Event`s here; the SSE
+    /// handler streams them directly to the client without re-wrapping.
+    pub sse_sender: mpsc::Sender<Event>,
 
     /// The full signal JSON from the most recent render for this connection.
-    /// Phase 2 uses this to re-enter the render pipeline on data invalidation
-    /// without a client request.
+    ///
+    /// Stored so the fanout task can reconstruct a `SignalStore` and build a
+    /// `RenderContext` for server-initiated rerenders, without a client
+    /// request carrying the signals.
     pub last_signals_json: Option<String>,
 }
 
@@ -28,9 +33,9 @@ pub struct ConnectionState {
 ///
 /// Accessed from:
 /// - `GET /sse` handler (insert)
-/// - `POST /render` handler (read sender, update signals)
+/// - `POST /action` handler (update signals)
 /// - SSE drop cleanup (remove)
-/// - Phase 2: invalidation handler (read sender + signals)
+/// - Fanout task (read sender + signals for rerender)
 pub struct ConnectionManager {
     connections: DashMap<Uuid, ConnectionState>,
 }
@@ -43,7 +48,7 @@ impl ConnectionManager {
     }
 
     /// Register a new connection with its SSE sender.
-    pub fn insert(&self, id: Uuid, sender: mpsc::Sender<String>) {
+    pub fn insert(&self, id: Uuid, sender: mpsc::Sender<Event>) {
         self.connections.insert(
             id,
             ConnectionState {
@@ -53,12 +58,24 @@ impl ConnectionManager {
         );
     }
 
-    /// Get the SSE sender for a connection (for sending server-initiated events).
-    pub fn get_sender(&self, id: &Uuid) -> Option<mpsc::Sender<String>> {
+    /// Get the SSE sender for a connection.
+    pub fn get_sender(&self, id: &Uuid) -> Option<mpsc::Sender<Event>> {
         self.connections.get(id).map(|c| c.sse_sender.clone())
     }
 
-    /// Update the stored signals for a connection after a render.
+    /// Get the last signals JSON for a connection.
+    ///
+    /// Called by the fanout task to reconstruct signal state for a
+    /// server-initiated rerender.
+    pub fn get_signals(&self, id: &Uuid) -> Option<String> {
+        self.connections
+            .get(id)
+            .and_then(|c| c.last_signals_json.clone())
+    }
+
+    /// Update the stored signals for a connection.
+    ///
+    /// Called by `POST /action` after ingesting the client's signal blob.
     pub fn update_signals(&self, id: &Uuid, signals_json: String) {
         if let Some(mut conn) = self.connections.get_mut(id) {
             conn.last_signals_json = Some(signals_json);
@@ -83,16 +100,27 @@ impl ConnectionManager {
 /// Bidirectional map: CacheKey ↔ ConnectionId.
 ///
 /// After each render, the connection's key associations are replaced with
-/// the new set from `cx.subscription_keys`. On connection drop, all entries
-/// for that connection are removed.
+/// the new set from `cx.take_subscription_keys()`. On connection drop, all
+/// entries for that connection are removed.
 ///
-/// Phase 2: on data invalidation event, look up `key_to_conns` to find
-/// which connections need rerender.
+/// This is the hot path of the reactive pipeline: the fanout task calls
+/// `get_connections_for_key` on every broadcast invalidation event to find
+/// which connections need rerendering.
 pub struct SubscriptionRegistry {
+    // DashMap is Arc-backed internally, so clone is cheap.
     key_to_conns: DashMap<CacheKey, HashSet<Uuid>>,
     conn_to_keys: DashMap<Uuid, HashSet<CacheKey>>,
 }
 
+impl Clone for SubscriptionRegistry {
+    fn clone(&self) -> Self {
+        Self {
+            key_to_conns: self.key_to_conns.clone(),
+            conn_to_keys: self.conn_to_keys.clone(),
+        }
+    }
+}
+
 impl SubscriptionRegistry {
     pub fn new() -> Self {
         Self {
@@ -103,7 +131,7 @@ impl SubscriptionRegistry {
 
     /// Replace a connection's subscription keys after a render.
     ///
-    /// Removes old associations and inserts new ones.
+    /// Removes old associations and inserts new ones atomically per-key.
     pub fn update(&self, conn_id: Uuid, new_keys: Vec<CacheKey>) {
         // Remove old associations
         if let Some((_, old_keys)) = self.conn_to_keys.remove(&conn_id) {
@@ -136,12 +164,13 @@ impl SubscriptionRegistry {
         }
     }
 
-    /// Phase 2: look up which connections are subscribed to a given key.
-    #[allow(dead_code)]
+    /// Look up which connections are subscribed to a given key.
+    ///
+    /// Called by the fanout task on every invalidation event.
     pub fn get_connections_for_key(&self, key: &CacheKey) -> HashSet<Uuid> {
         self.key_to_conns
             .get(key)
             .map(|c| c.clone())
             .unwrap_or_default()
     }
-}
+}
\ No newline at end of file
diff --git a/src/context.rs b/src/context.rs
index 3c73976..c9c220d 100644
--- a/src/context.rs
+++ b/src/context.rs
@@ -1,7 +1,9 @@
 use crate::signal::{Signal, SignalOpts, SignalStore};
 use crate::query::{CacheKey, Query};
+use crate::store::Store;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use std::sync::Arc;
 use uuid::Uuid;
 
 // ---------------------------------------------------------------------------
@@ -12,9 +14,8 @@ use uuid::Uuid;
 /// signal.
 ///
 /// Static children (always rendered by their parent) do NOT appear here —
-/// only dynamic/conditional slots. In v0 we primarily use `page` to resolve
-/// the root component. The `children` map is the seam for nested dynamic
-/// components in Phase 2.
+/// only dynamic/conditional slots. `page` resolves the root component;
+/// `children` is the seam for nested dynamic components.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ComponentTree {
     /// The root page component name, e.g. `"TodoList"`.
@@ -47,26 +48,25 @@ impl Default for ComponentTree {
 /// The component's window into the framework.
 ///
 /// Passed as `&mut cx` to every component function. Provides access to
-/// signals, database queries, and subscription key collection.
+/// signals, DB queries, KV reads, and subscription key collection.
 ///
-/// Phase 2 adds: cache handle, auth/session info.
+/// After rendering, call `take_subscription_keys()` to flush subscriptions
+/// to the `SubscriptionRegistry` for this connection.
 pub struct RenderContext {
     /// This connection's UUID.
     pub connection_id: Uuid,
 
-    /// Signal values populated from the incoming POST body (or initial
-    /// defaults on first page load).
+    /// Signal values populated from `last_signals_json` (or initial defaults
+    /// on first render).
     pub signals: SignalStore,
 
     /// Parsed component tree from the `tree` signal.
     pub tree: ComponentTree,
 
-    /// Database pool. Cloneable (it's an Arc internally).
-    pub db: sqlx::SqlitePool,
+    /// Access to DB, KV store, and the broadcast event channel.
+    pub store: Arc<Store>,
 
-    /// Subscription keys accumulated during this render. After render
-    /// completes, these are flushed to the SubscriptionRegistry for this
-    /// connection.
+    /// Subscription keys accumulated during this render.
     pub(crate) subscription_keys: Vec<CacheKey>,
 
     /// Signal names touched during this render (for demangling tracking).
@@ -78,13 +78,13 @@ impl RenderContext {
         connection_id: Uuid,
         signals: SignalStore,
         tree: ComponentTree,
-        db: sqlx::SqlitePool,
+        store: Arc<Store>,
     ) -> Self {
         Self {
             connection_id,
             signals,
             tree,
-            db,
+            store,
             subscription_keys: Vec::new(),
             touched_signals: Vec::new(),
         }
@@ -109,10 +109,10 @@ impl RenderContext {
     }
 
     // -----------------------------------------------------------------------
-    // Queries
+    // DB Queries
     // -----------------------------------------------------------------------
 
-    /// Execute a single query, collecting its subscription key.
+    /// Execute a single DB query, collecting its subscription key.
     ///
     /// ```ignore
     /// let todos = cx.run(ListTodos).await?;
@@ -122,14 +122,10 @@ impl RenderContext {
         if key != CacheKey::None {
             self.subscription_keys.push(key);
         }
-        query.execute(&self.db).await
+        query.execute(&self.store.db).await
     }
 
-    /// Execute two queries concurrently, collecting both subscription keys.
-    ///
-    /// This avoids the `&mut self` borrow conflict by collecting keys
-    /// synchronously first, then running the async executions against a
-    /// cloned pool handle (pool clones are cheap — they share the inner Arc).
+    /// Execute two DB queries concurrently, collecting both subscription keys.
     ///
     /// ```ignore
     /// let (board, columns) = cx.run_all(
@@ -142,7 +138,6 @@ impl RenderContext {
         a: A,
         b: B,
     ) -> anyhow::Result<(A::Output, B::Output)> {
-        // Collect subscription keys synchronously (no borrow conflict)
         let key_a = a.cache_key();
         let key_b = b.cache_key();
         if key_a != CacheKey::None {
@@ -151,14 +146,12 @@ impl RenderContext {
         if key_b != CacheKey::None {
             self.subscription_keys.push(key_b);
         }
-
-        // Execute concurrently against a cloned pool handle
-        let db = self.db.clone();
+        let db = self.store.db.clone();
         let (ra, rb) = tokio::try_join!(a.execute(&db), b.execute(&db))?;
         Ok((ra, rb))
     }
 
-    /// Three-query variant of `run_all`.
+    /// Three-query concurrent variant.
     pub async fn run_all3<A: Query, B: Query, C: Query>(
         &mut self,
         a: A,
@@ -177,18 +170,47 @@ impl RenderContext {
         if key_c != CacheKey::None {
             self.subscription_keys.push(key_c);
         }
-
-        let db = self.db.clone();
+        let db = self.store.db.clone();
         let (ra, rb, rc) = tokio::try_join!(a.execute(&db), b.execute(&db), c.execute(&db))?;
         Ok((ra, rb, rc))
     }
 
+    // -----------------------------------------------------------------------
+    // KV / Channel queries
+    // -----------------------------------------------------------------------
+
+    /// Read a value from the KV store and register a subscription to its key.
+    ///
+    /// When `store.kv.set(key, ...)` is called (or `store.publish(...)` for
+    /// that key), all connections that called `kv_run` with that key during
+    /// their last render will be scheduled for rerender.
+    ///
+    /// Returns `None` if the key has not been set yet.
+    ///
+    /// ```ignore
+    /// let tick: Option<serde_json::Value> = cx.kv_run("clock").await;
+    /// ```
+    pub async fn kv_run(&mut self, key: impl Into<String>) -> Option<serde_json::Value> {
+        let key = key.into();
+        self.subscription_keys
+            .push(CacheKey::Channel { key: key.clone() });
+        self.store.kv.get(&key)
+    }
+
     // -----------------------------------------------------------------------
     // Post-render access
     // -----------------------------------------------------------------------
 
-    /// Take the subscription keys collected during this render.
-    /// Called by the handler after rendering to flush them to the registry.
+    /// Take the subscription keys accumulated during this render.
+    ///
+    /// Call this at the top of the component call stack before returning from
+    /// `render_fn`. The returned `Vec<CacheKey>` is the second element of the
+    /// `(String, Vec<CacheKey>)` tuple — the framework uses it to update the
+    /// `SubscriptionRegistry` for this connection after each successful render.
+    ///
+    /// Keys accumulate naturally as the call stack unwinds: each `cx.run()`
+    /// and `cx.kv_run()` call pushes its key, so the top-level collect gets
+    /// the full set from the entire render.
     pub fn take_subscription_keys(&mut self) -> Vec<CacheKey> {
         std::mem::take(&mut self.subscription_keys)
     }
@@ -197,4 +219,4 @@ impl RenderContext {
     pub fn take_touched_signals(&mut self) -> Vec<String> {
         std::mem::take(&mut self.touched_signals)
     }
-}
+}
\ No newline at end of file
diff --git a/src/lib.rs b/src/lib.rs
index 0fcf1bf..a1bb017 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,5 +1,6 @@
 pub mod signal;
 pub mod query;
+pub mod store;
 pub mod context;
 pub mod connection;
 pub mod sse;
@@ -7,7 +8,8 @@ pub mod server;
 
 pub use signal::{Signal, SignalStore, SignalOpts, mangle, client_only, global, data_signals};
 pub use query::{CacheKey, Query};
+pub use store::{Store, KVStore};
 pub use context::{RenderContext, ComponentTree, ComponentNode};
 pub use connection::{ConnectionManager, ConnectionState, SubscriptionRegistry};
 pub use sse::{format_patch_elements, format_patch_signals};
-pub use server::{AppState, SharedState, sse_handler};
+pub use server::{AppState, SharedState, RenderFn, sse_handler, spawn_fanout, ingest_signals, ActionParams, SseParams};
\ No newline at end of file
diff --git a/src/query.rs b/src/query.rs
index 2060552..e650d0a 100644
--- a/src/query.rs
+++ b/src/query.rs
@@ -5,37 +5,50 @@ use std::pin::Pin;
 // CacheKey
 // ---------------------------------------------------------------------------
 
-/// Identifies what data a query depends on, for subscription tracking.
+/// Identifies what data a query depends on, for subscription tracking and
+/// reactive invalidation.
 ///
-/// In v0 these are collected but not acted on. Phase 2 uses them to map
-/// data changes → affected connections → rerender.
-///
-/// Phase 2 adds: `Derived { key: String, value: i64 }` for composite keys
-/// like `"posts:user_id:42"`.
+/// When something changes (a DB mutation, a KV write, an external event),
+/// the responsible code publishes the relevant `CacheKey` to the broadcast
+/// channel in `Store`. The fanout task maps that key to subscribed connections
+/// and triggers rerenders.
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub enum CacheKey {
     /// Query has no cacheable identity (rare — side-effect-only reads).
     None,
+
     /// Single row by primary key: `("todos", 42)`.
     PrimaryKey { table: &'static str, id: i64 },
-    /// future, something like this
-    //ColumnValue { table: &'static str, column: &'static str, val: Box<dyn Any> },
+
     /// Entire table: `("todos")`. Used for list queries.
     Table { table: &'static str },
+
+    /// A named channel key, used by KVStore entries and pure pubsub events.
+    ///
+    /// KV reads register a subscription under this key. KV writes (and any
+    /// external publisher — timers, webhooks, etc.) publish this key to the
+    /// broadcast channel, triggering rerenders for all subscribed connections.
+    ///
+    /// ```ignore
+    /// // Component reads:
+    /// let val = cx.kv_run("clock").await?;
+    ///
+    /// // Timer publishes:
+    /// store.events.send(CacheKey::Channel { key: "clock".into() }).ok();
+    /// ```
+    Channel { key: String },
 }
 
 // ---------------------------------------------------------------------------
 // Query trait
 // ---------------------------------------------------------------------------
 
-/// The central data-access abstraction.
+/// The central data-access abstraction for DB-backed queries.
 ///
 /// Every read goes through `cx.run(SomeQuery { ... })`, which:
 /// 1. Collects the query's `cache_key` for subscription tracking.
 /// 2. Executes the query against the database.
 ///
-/// Phase 2 inserts a cache check between steps 1 and 2.
-///
 /// The trait uses `Pin<Box<dyn Future>>` instead of `async fn` to guarantee
 /// `Send` bounds needed for `tokio::try_join!` in `cx.run_all()`. The
 /// `query_one!` / `query_all!` macros hide this behind a clean declaration.
@@ -56,9 +69,6 @@ pub trait Query: Send + Sync {
 
 /// Generate a `Query` impl that fetches a single row.
 ///
-/// Both `params` and `cache` take closures receiving `&StructName`, so `self`
-/// is always available without hygiene issues.
-///
 /// ```ignore
 /// pub struct GetTodo { pub id: i64 }
 ///
@@ -69,8 +79,6 @@ pub trait Query: Send + Sync {
 ///     cache: |s: &GetTodo| CacheKey::PrimaryKey { table: "todos", id: s.id },
 /// });
 /// ```
-///
-/// The `output` type must implement `sqlx::FromRow<'_, SqliteRow>`.
 #[macro_export]
 macro_rules! query_one {
     ($name:ident {
@@ -111,9 +119,6 @@ macro_rules! query_one {
 
 /// Generate a `Query` impl that fetches multiple rows.
 ///
-/// Both `params` and `cache` take closures receiving `&StructName`, so the
-/// style is consistent with `query_one!` even when `s` goes unused.
-///
 /// ```ignore
 /// pub struct ListTodos;
 ///
@@ -124,8 +129,6 @@ macro_rules! query_one {
 ///     cache: |_s: &ListTodos| CacheKey::Table { table: "todos" },
 /// });
 /// ```
-///
-/// `Output` is `Vec<output_type>`.
 #[macro_export]
 macro_rules! query_all {
     ($name:ident {
@@ -158,4 +161,4 @@ macro_rules! query_all {
             }
         }
     };
-}
+}
\ No newline at end of file
diff --git a/src/server.rs b/src/server.rs
index 0e14bae..b6bddfd 100644
--- a/src/server.rs
+++ b/src/server.rs
@@ -1,7 +1,12 @@
 use crate::connection::{ConnectionManager, SubscriptionRegistry};
+use crate::context::{ComponentTree, RenderContext};
+use crate::signal::SignalStore;
+use crate::store::Store;
+use crate::query::CacheKey;
 
 use axum::extract::{Query as AxumQuery, State};
 use axum::response::sse::{Event, KeepAlive, Sse};
+use futures::future::BoxFuture;
 use std::convert::Infallible;
 use std::sync::Arc;
 use tokio::sync::mpsc;
@@ -9,21 +14,167 @@ use tokio_stream::wrappers::ReceiverStream;
 use uuid::Uuid;
 
 // ---------------------------------------------------------------------------
-// Shared application state
+// RenderFn
 // ---------------------------------------------------------------------------
 
-/// All shared state, wrapped in Arc for axum's State extractor.
+/// Type alias for the app-registered render callback.
 ///
-/// Applications can use this directly or define their own state type.
-/// The framework's `sse_handler` is tied to this type.
+/// The app registers this at startup. The fanout task calls it to produce
+/// an SSE `datastar-patch-elements` payload for a given connection's state.
+///
+/// The function receives a fully constructed `RenderContext` and must return
+/// the rendered HTML string plus the full set of `CacheKey`s accumulated
+/// during the render (via `cx.take_subscription_keys()`). Keys naturally
+/// bubble up through the component call stack — the top-level render function
+/// just collects them all before returning.
+///
+/// The framework handles SSE framing and updates the `SubscriptionRegistry`
+/// from the returned keys after each successful render.
+///
+/// ```ignore
+/// let render_fn: RenderFn = Arc::new(|mut cx| Box::pin(async move {
+///     let html = render_page(&mut cx).await?;
+///     let keys = cx.take_subscription_keys();
+///     Ok((html, keys))
+/// }));
+/// ```
+pub type RenderFn = Arc<dyn Fn(RenderContext) -> BoxFuture<'static, anyhow::Result<(String, Vec<CacheKey>)>> + Send + Sync>;
+
+// ---------------------------------------------------------------------------
+// AppState
+// ---------------------------------------------------------------------------
+
+/// All shared framework + app state.
+///
+/// Construct with `AppState::new(store, render_fn)` and wrap in `Arc` to
+/// produce a `SharedState` for axum's `State` extractor.
 pub struct AppState {
-    pub db: sqlx::SqlitePool,
+    pub store: Arc<Store>,
     pub connections: ConnectionManager,
     pub subscriptions: SubscriptionRegistry,
+    pub render_fn: RenderFn,
 }
 
 pub type SharedState = Arc<AppState>;
 
+impl AppState {
+    pub fn new(store: Arc<Store>, render_fn: RenderFn) -> SharedState {
+        Arc::new(Self {
+            store,
+            connections: ConnectionManager::new(),
+            subscriptions: SubscriptionRegistry::new(),
+            render_fn,
+        })
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Fanout task
+// ---------------------------------------------------------------------------
+
+/// Spawn the global invalidation fanout task.
+///
+/// Must be called once at startup, after building `AppState`. The task owns
+/// a `broadcast::Receiver<CacheKey>` and runs for the lifetime of the server.
+///
+/// On each invalidation event:
+/// 1. Look up subscribed connections via `SubscriptionRegistry`.
+/// 2. For each connection, fetch `last_signals_json` + SSE sender.
+/// 3. Reconstruct `SignalStore` + `ComponentTree`, build `RenderContext`.
+/// 4. Call `render_fn` to produce HTML.
+/// 5. Frame as SSE and send down the connection's SSE channel.
+/// 6. Update subscription keys and `last_signals_json` from the new render.
+///
+/// ```ignore
+/// let state = AppState::new(store.clone(), render_fn);
+/// spawn_fanout(state.clone());
+/// ```
+pub fn spawn_fanout(state: SharedState) {
+    let mut rx = state.store.subscribe();
+
+    tokio::spawn(async move {
+        loop {
+            let key = match rx.recv().await {
+                Ok(k) => k,
+                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
+                    tracing::warn!("fanout lagged, missed {} invalidation events", n);
+                    continue;
+                }
+                Err(tokio::sync::broadcast::error::RecvError::Closed) => {
+                    tracing::info!("fanout channel closed, shutting down");
+                    break;
+                }
+            };
+
+            let conn_ids = state.subscriptions.get_connections_for_key(&key);
+
+            for conn_id in conn_ids {
+                let Some(signals_json) = state.connections.get_signals(&conn_id) else {
+                    continue;
+                };
+                let Some(sse_tx) = state.connections.get_sender(&conn_id) else {
+                    continue;
+                };
+
+                let signals = match serde_json::from_str::<serde_json::Value>(&signals_json) {
+                    Ok(v) => SignalStore::from_json(v),
+                    Err(e) => {
+                        tracing::error!("fanout: failed to parse signals for {}: {}", conn_id, e);
+                        continue;
+                    }
+                };
+
+                // Parse the component tree from the `tree` signal.
+                let tree = signals
+                    .get("tree")
+                    .and_then(|v| serde_json::from_value(v.clone()).ok())
+                    .unwrap_or_default();
+
+                let mut cx = RenderContext::new(
+                    conn_id,
+                    signals,
+                    tree,
+                    state.store.clone(),
+                );
+
+                let render_fn = state.render_fn.clone();
+                let state_ref = state.clone();
+
+                tokio::spawn(async move {
+                    match render_fn(cx).await {
+                        Ok((html, keys)) => {
+                            state_ref.subscriptions.update(conn_id, keys);
+
+                            // Build a single datastar-patch-elements SSE event.
+                            // DataStar expects the event name set once, then one
+                            // `data: elements <line>` entry per line of HTML.
+                            // axum's Event::data() sets a single data field, so
+                            // we pre-format the multi-line data payload manually
+                            // and send it as one data value.
+                            let data_payload: String = html
+                                .lines()
+                                .map(|line| format!("elements {}", line))
+                                .collect::<Vec<_>>()
+                                .join("\ndata: ");
+
+                            let event = Event::default()
+                                .event("datastar-patch-elements")
+                                .data(data_payload);
+
+                            if sse_tx.send(event).await.is_err() {
+                                tracing::debug!("fanout: SSE channel closed for {}", conn_id);
+                            }
+                        }
+                        Err(e) => {
+                            tracing::error!("fanout: render error for {}: {}", conn_id, e);
+                        }
+                    }
+                });
+            }
+        }
+    });
+}
+
 // ---------------------------------------------------------------------------
 // SSE handler
 // ---------------------------------------------------------------------------
@@ -36,50 +187,124 @@ pub struct SseParams {
 
 /// `GET /sse?conn=<uuid>` — Long-lived SSE pipe for server-initiated push.
 ///
-/// In v0, nothing actively pushes events through this pipe (no invalidation
-/// events yet). It exists to:
-/// 1. Register the connection in ConnectionManager.
-/// 2. Keep the SSE connection alive (DataStar expects it).
-/// 3. Provide the Phase 2 seam for server-initiated rerenders.
-///
-/// On drop, the connection is cleaned up from both the ConnectionManager
-/// and the SubscriptionRegistry.
+/// The conn_id is generated by `GET /` and pre-seeded with initial signals
+/// in `ConnectionManager` before this handler is called. On connect, the
+/// handler immediately triggers a render and pushes the result down the pipe,
+/// so the client sees content without any further round-trip.
 pub async fn sse_handler(
     State(state): State<SharedState>,
     AxumQuery(params): AxumQuery<SseParams>,
 ) -> Sse<impl tokio_stream::Stream<Item = Result<Event, Infallible>>> {
     let conn_id = params.conn;
-
-    // Create a channel for sending events to this connection
-    let (tx, rx) = mpsc::channel::<String>(32);
-
-    // Clone sender for the cleanup watcher. Sender::closed() resolves when
-    // the receiver is dropped (i.e. client disconnects and the SSE stream
-    // drops). This is the proper cleanup mechanism — no arbitrary sleep.
+    let (tx, rx) = mpsc::channel::<Event>(32);
     let tx_watch = tx.clone();
 
-    // Register the connection
-    state.connections.insert(conn_id, tx);
+    state.connections.insert(conn_id, tx.clone());
 
-    // Spawn cleanup task: waits for the receiver to drop (client disconnect),
-    // then removes the connection from all registries.
+    // Cleanup on disconnect
     let state_cleanup = state.clone();
     tokio::spawn(async move {
         tx_watch.closed().await;
         state_cleanup.connections.remove(&conn_id);
         state_cleanup.subscriptions.remove_connection(&conn_id);
+        tracing::debug!("cleaned up connection {}", conn_id);
     });
+    println!("SSE Handler called");
+
+    // Trigger first render immediately using the pre-seeded signals.
+    let signals_json = state.connections.get_signals(&conn_id)
+        .unwrap_or_else(|| "{}".to_string());
+
+    let render_fn = state.render_fn.clone();
+    let store = state.store.clone();
+    let subscriptions = state.subscriptions.clone();
+    let tx_init = tx.clone();
+
+    tokio::spawn(async move {
+        let signals = match serde_json::from_str::<serde_json::Value>(&signals_json) {
+            Ok(v) => SignalStore::from_json(v),
+            Err(e) => {
+                tracing::error!("sse_handler: failed to parse seeded signals: {}", e);
+                return;
+            }
+        };
+        let tree = signals
+            .get("tree")
+            .and_then(|v| serde_json::from_value(v.clone()).ok())
+            .unwrap_or_default();
+
+        let cx = RenderContext::new(conn_id, signals, tree, store);
+        match render_fn(cx).await {
+            Ok((html, keys)) => {
+                subscriptions.update(conn_id, keys);
+                let data_payload = html
+                    .lines()
+                    .map(|line| format!("elements {}", line))
+                    .collect::<Vec<_>>()
+                    .join("\ndata: ");
+                let event = Event::default()
+                    .event("datastar-patch-elements")
+                    .data(data_payload);
+                tx_init.send(event).await.ok();
+            }
+            Err(e) => tracing::error!("sse_handler: initial render failed: {}", e),
+        }
+    });
+    
 
-    // Convert the receiver into an SSE stream.
-    // DESIGN NOTE: For v0, the SSE pipe is mostly idle — POST /render
-    // returns SSE directly in its response, not through this pipe. This
-    // channel exists for Phase 2 server-initiated pushes (data invalidation).
-    // Messages sent via the channel should be plain content strings (not
-    // pre-formatted SSE); axum's Sse wrapper handles the SSE framing.
     let stream = ReceiverStream::new(rx);
-    let event_stream = tokio_stream::StreamExt::map(stream, |msg| {
-        Ok(Event::default().data(msg))
+    let event_stream = tokio_stream::StreamExt::map(stream, |event| {
+        Ok::<Event, Infallible>(event)
     });
 
     Sse::new(event_stream).keep_alive(KeepAlive::default())
 }
+
+// ---------------------------------------------------------------------------
+// Action handler helper
+// ---------------------------------------------------------------------------
+
+/// Query params for the action endpoint.
+#[derive(serde::Deserialize)]
+pub struct ActionParams {
+    pub conn: Uuid,
+}
+
+/// Ingest a signal update from the client and store it for future rerenders.
+///
+/// This is the framework's half of the action pipeline. The app's action
+/// handler should call this to persist the updated signal state, then perform
+/// its own side effects (DB writes, KV writes, event publishes) which will
+/// drive rerenders through the fanout task.
+///
+/// Returns the parsed `SignalStore` so the app handler can read signal values.
+///
+/// ```ignore
+/// async fn delete_todo(
+///     State(state): State<SharedState>,
+///     AxumQuery(params): AxumQuery<ActionParams>,
+///     Json(body): Json<serde_json::Value>,
+/// ) -> impl IntoResponse {
+///     let signals = ingest_signals(&state, params.conn, body);
+///     let todo_id: i64 = signals.get("todoId")
+///         .and_then(|v| v.as_i64())
+///         .unwrap_or_default();
+///     sqlx::query!("DELETE FROM todos WHERE id = ?", todo_id)
+///         .execute(&state.store.db)
+///         .await
+///         .ok();
+///     state.store.publish(CacheKey::Table { table: "todos" });
+///     StatusCode::ACCEPTED
+/// }
+/// ```
+pub fn ingest_signals(
+    state: &AppState,
+    conn_id: Uuid,
+    body: serde_json::Value,
+) -> SignalStore {
+    let store = SignalStore::from_json(body);
+    state
+        .connections
+        .update_signals(&conn_id, store.to_json_string());
+    store
+}
\ No newline at end of file
diff --git a/src/store.rs b/src/store.rs
new file mode 100644
index 0000000..ed3c304
--- /dev/null
+++ b/src/store.rs
@@ -0,0 +1,140 @@
+use crate::query::CacheKey;
+use dashmap::DashMap;
+use std::sync::Arc;
+use tokio::sync::broadcast;
+
+// ---------------------------------------------------------------------------
+// KVStore
+// ---------------------------------------------------------------------------
+
+/// A reactive key-value store backed by a `DashMap`.
+///
+/// Reads are plain lookups. Writes automatically publish a `CacheKey::Channel`
+/// invalidation event to the broadcast channel, which the fanout task uses to
+/// trigger rerenders for all connections subscribed to that key.
+///
+/// The KV store is the source of truth for non-relational state: counters,
+/// flags, computed values, timer ticks, etc.
+///
+/// ```ignore
+/// // In a component:
+/// let tick = cx.kv_run("clock").await?;  // registers subscription
+///
+/// // In an action handler or timer task:
+/// store.kv.set("clock", serde_json::json!(timestamp));
+/// // ↑ automatically publishes CacheKey::Channel { key: "clock" }
+/// ```
+pub struct KVStore {
+    store: DashMap<String, serde_json::Value>,
+    events: broadcast::Sender<CacheKey>,
+}
+
+impl KVStore {
+    pub fn new(events: broadcast::Sender<CacheKey>) -> Self {
+        Self {
+            store: DashMap::new(),
+            events,
+        }
+    }
+
+    /// Read a value by key. Does not register a subscription — use
+    /// `cx.kv_run(key)` in components to also track the subscription.
+    pub fn get(&self, key: &str) -> Option<serde_json::Value> {
+        self.store.get(key).map(|v| v.clone())
+    }
+
+    /// Write a value and publish an invalidation event.
+    ///
+    /// All connections subscribed to `CacheKey::Channel { key }` will be
+    /// scheduled for rerender by the fanout task.
+    pub fn set(&self, key: impl Into<String>, value: serde_json::Value) {
+        let key = key.into();
+        self.store.insert(key.clone(), value);
+        // Ignore send errors — no receivers just means no connections are
+        // currently subscribed, which is fine.
+        self.events
+            .send(CacheKey::Channel { key })
+            .ok();
+    }
+
+    /// Delete a key and publish an invalidation event.
+    pub fn delete(&self, key: &str) {
+        self.store.remove(key);
+        self.events
+            .send(CacheKey::Channel { key: key.to_string() })
+            .ok();
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Store
+// ---------------------------------------------------------------------------
+
+/// The cluster of data resources available to both the framework and the app.
+///
+/// `Arc<Store>` is embedded in `AppState` and passed to action handlers.
+/// The app uses it to read/write data; writes to `kv` or direct publishes
+/// to `events` drive the reactive rerender pipeline.
+///
+/// ```ignore
+/// // Register at startup:
+/// let store = Store::new(db_pool);
+///
+/// // In an action handler:
+/// store.kv.set("counter", json!(42));
+///
+/// // Publishing a pure event (no stored value, e.g. for PubSub-style triggers):
+/// store.publish(CacheKey::Channel { key: "refresh".into() });
+/// ```
+pub struct Store {
+    /// SQLite connection pool. Cloneable (Arc internally).
+    pub db: sqlx::SqlitePool,
+
+    /// Reactive key-value store. Writes auto-publish invalidation events.
+    pub kv: Arc<KVStore>,
+
+    /// Broadcast sender for invalidation events. Clone this to publish from
+    /// anywhere (action handlers, timer tasks, webhooks).
+    ///
+    /// Use `store.publish(key)` as a convenience wrapper.
+    pub events: broadcast::Sender<CacheKey>,
+}
+
+impl Store {
+    /// Construct with the given DB pool. Creates the broadcast channel and
+    /// KV store internally.
+    ///
+    /// The broadcast channel capacity (256) is the number of unread events
+    /// that can be buffered before slow receivers start lagging. The fanout
+    /// task should be fast enough that this is never hit in practice.
+    pub fn new(db: sqlx::SqlitePool) -> Arc<Self> {
+        let (events_tx, _) = broadcast::channel(256);
+        let kv = Arc::new(KVStore::new(events_tx.clone()));
+        Arc::new(Self {
+            db,
+            kv,
+            events: events_tx,
+        })
+    }
+
+    /// Publish a raw invalidation event without writing to KV.
+    ///
+    /// Use this for pure pubsub triggers (timers, external webhooks) where
+    /// there is no stored value to update — only a rerender signal to send.
+    ///
+    /// ```ignore
+    /// // Every second, trigger rerenders for connections subscribed to "clock":
+    /// store.publish(CacheKey::Channel { key: "clock".into() });
+    /// ```
+    pub fn publish(&self, key: CacheKey) {
+        self.events.send(key).ok();
+    }
+
+    /// Subscribe to the invalidation broadcast channel.
+    ///
+    /// Called once at startup by the fanout task. Each call returns an
+    /// independent receiver that sees all published events from this point on.
+    pub fn subscribe(&self) -> broadcast::Receiver<CacheKey> {
+        self.events.subscribe()
+    }
+}
\ No newline at end of file