bodged refactor

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()
     }
-}
+}

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)
     }
-}
+}

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};

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 {
             }
         }
     };
-}
+}

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
+}

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()
+    }
+}