Refactor Query contract: uniform ergonomics for SQLite + PostgreSQL

- Add `publish_key()` default method to `Query` trait (returns `None`)
- Add `mutation_exec!` macro for SQLite INSERT/UPDATE/DELETE ops; overrides
  `publish_key()` so the mutation carries its own invalidation key
- Add `Store::exec()` which runs a mutation and auto-publishes its key —
  action handlers no longer need a separate `store.publish()` call
- Add `PgQuery` trait + `pg_query_one!` / `pg_query_all!` macros
  (feature-gated on `pg_replication`) mirroring the SQLite Query API
- Add `cx.run_pg(pool, query)` to `RenderContext` which executes a PgQuery
  and auto-subscribes its cache key — no more manual `cx.subscribe()` calls
- Update todo example: replace plain mutation fns with `mutation_exec!` structs
  and `store.exec()` calls; remove `action_then_publish` helper
- Update pg_replication example: add `ListMessages` via `pg_query_all!` and
  replace manual subscribe + raw query with `cx.run_pg()`

https://claude.ai/code/session_01HXe8rAZPweU9j2piBo9MdG

examples/pg_replication/main.rs

@@ -46,12 +46,26 @@ use livevue_rs::{
     spawn_fanout, RenderContext, Store,
 };
 
+// ---------------------------------------------------------------------------
+// Queries
+// ---------------------------------------------------------------------------
+
+/// Fetch all messages, newest first.
+pub struct ListMessages;
+
+livevue_rs::pg_query_all!(ListMessages {
+    sql: "SELECT id, body FROM messages ORDER BY id DESC",
+    params: [],
+    output: Message,
+    cache: |_| pg_table_key("public", "messages"),
+});
+
 // ---------------------------------------------------------------------------
 // Domain types
 // ---------------------------------------------------------------------------
 
 #[derive(Debug, Clone, FromRow)]
-struct Message {
+pub struct Message {
     id: i64,
     body: String,
 }
@@ -101,14 +115,8 @@ struct RenderState {
 }
 
 async fn render_page(cx: &mut RenderContext, rs: &RenderState) -> anyhow::Result<Markup> {
-    // Subscribe to PG-sourced invalidation events for the messages table.
+    // Fetch messages and auto-subscribe to the WAL invalidation key.
-    // The fanout task will re-render this connection whenever the WAL emitter
+    let messages = cx.run_pg(&rs.pg, ListMessages).await?;
-    // fires a `CacheKey::Channel { key: "pg:public.messages" }` event.
-    cx.subscribe(pg_table_key("public", "messages"));
-
-    let messages: Vec<Message> = sqlx::query_as("SELECT id, body FROM messages ORDER BY id DESC")
-        .fetch_all(&rs.pg)
-        .await?;
 
     let new_msg = cx.signal("newMsg", livevue_rs::global());
     let _conn_id_signal = cx.signal("connId", livevue_rs::global());

examples/todo/app.rs

@@ -1,5 +1,5 @@
 use livevue_rs::{
-    AppState, CacheKey, ComponentTree, RenderContext, SharedState, SignalStore,
+    ComponentTree, RenderContext, SharedState, SignalStore,
     ingest_signals, sse_handler,
 };
 
@@ -155,50 +155,30 @@ struct ActionIdParams {
     id: i64,
 }
 
-/// 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,
-    body: serde_json::Value,
-    mutation: impl std::future::Future<Output = anyhow::Result<()>> + Send,
-    invalidation_key: CacheKey,
-) -> StatusCode {
-    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 mutation error: {}", e);
-        return StatusCode::INTERNAL_SERVER_ERROR;
-    }
-
-    state.store.publish(invalidation_key);
-    StatusCode::ACCEPTED
-}
-
 async fn action_add_todo(
     State(state): State<SharedState>,
     Json(body): Json<serde_json::Value>,
 ) -> impl IntoResponse {
+    let conn_id = extract_conn_and_tree(&SignalStore::from_json(body.clone())).0;
+    ingest_signals(&state, conn_id, body.clone());
+
     let title = body
         .get("newTodo")
         .and_then(|v| v.as_str())
         .unwrap_or("")
         .trim()
         .to_string();
-    let db = state.store.db.clone();
 
-    action_then_publish(
+    if title.is_empty() {
-        &state,
+        return StatusCode::ACCEPTED;
-        body,
+    }
-        async move {
+
-            if !title.is_empty() {
+    if let Err(e) = state.store.exec(crate::queries::CreateTodo { title }).await {
-                crate::queries::create_todo(&db, &title).await?;
+        eprintln!("action_add_todo error: {e}");
-            }
+        return StatusCode::INTERNAL_SERVER_ERROR;
-            Ok(())
+    }
-        },
+
-        CacheKey::Table { table: "todos" },
+    StatusCode::ACCEPTED
-    )
-    .await
 }
 
 async fn action_toggle_todo(
@@ -206,16 +186,15 @@ async fn action_toggle_todo(
     AxumQuery(params): AxumQuery<ActionIdParams>,
     Json(body): Json<serde_json::Value>,
 ) -> impl IntoResponse {
-    let db = state.store.db.clone();
+    let conn_id = extract_conn_and_tree(&SignalStore::from_json(body.clone())).0;
-    let id = params.id;
+    ingest_signals(&state, conn_id, body);
 
-    action_then_publish(
+    if let Err(e) = state.store.exec(crate::queries::ToggleTodo { id: params.id }).await {
-        &state,
+        eprintln!("action_toggle_todo error: {e}");
-        body,
+        return StatusCode::INTERNAL_SERVER_ERROR;
-        async move { crate::queries::toggle_todo(&db, id).await },
+    }
-        CacheKey::Table { table: "todos" },
+
-    )
+    StatusCode::ACCEPTED
-    .await
 }
 
 async fn action_delete_todo(
@@ -223,14 +202,13 @@ async fn action_delete_todo(
     AxumQuery(params): AxumQuery<ActionIdParams>,
     Json(body): Json<serde_json::Value>,
 ) -> impl IntoResponse {
-    let db = state.store.db.clone();
+    let conn_id = extract_conn_and_tree(&SignalStore::from_json(body.clone())).0;
-    let id = params.id;
+    ingest_signals(&state, conn_id, body);
 
-    action_then_publish(
+    if let Err(e) = state.store.exec(crate::queries::DeleteTodo { id: params.id }).await {
-        &state,
+        eprintln!("action_delete_todo error: {e}");
-        body,
+        return StatusCode::INTERNAL_SERVER_ERROR;
-        async move { crate::queries::delete_todo(&db, id).await },
+    }
-        CacheKey::Table { table: "todos" },
+
-    )
+    StatusCode::ACCEPTED
-    .await
 }

examples/todo/queries.rs

@@ -40,36 +40,42 @@ livevue_rs::query_one!(GetTodo {
 });
 
 // ---------------------------------------------------------------------------
-// Write operations (plain sqlx, not through Query trait)
+// Write operations (mutation structs via mutation_exec!)
 // ---------------------------------------------------------------------------
 //
-// Mutations don't go through cx.run() because they're not reads and don't
+// Each mutation struct implements Query with publish_key() set to the cache
-// produce subscription keys. They're called directly by action handlers,
+// key it invalidates. Use Store::exec() in action handlers — it executes the
-// which then trigger a full rerender.
+// mutation and publishes the key automatically.
 
-/// Insert a new todo. Returns the new row's id.
+/// Insert a new todo.
-pub async fn create_todo(db: &sqlx::SqlitePool, title: &str) -> anyhow::Result<i64> {
+pub struct CreateTodo {
-    let result = sqlx::query("INSERT INTO todos (title, completed) VALUES (?, false)")
+    pub title: String,
-        .bind(title)
-        .execute(db)
-        .await?;
-    Ok(result.last_insert_rowid())
 }
 
+livevue_rs::mutation_exec!(CreateTodo {
+    sql: "INSERT INTO todos (title, completed) VALUES (?, false)",
+    params: [|s: &CreateTodo| s.title.clone()],
+    publish: |_| CacheKey::Table { table: "todos" },
+});
+
 /// Toggle a todo's completed status.
-pub async fn toggle_todo(db: &sqlx::SqlitePool, id: i64) -> anyhow::Result<()> {
+pub struct ToggleTodo {
-    sqlx::query("UPDATE todos SET completed = NOT completed WHERE id = ?")
+    pub id: i64,
-        .bind(id)
-        .execute(db)
-        .await?;
-    Ok(())
 }
 
+livevue_rs::mutation_exec!(ToggleTodo {
+    sql: "UPDATE todos SET completed = NOT completed WHERE id = ?",
+    params: [|s: &ToggleTodo| s.id],
+    publish: |_| CacheKey::Table { table: "todos" },
+});
+
 /// Delete a todo by id.
-pub async fn delete_todo(db: &sqlx::SqlitePool, id: i64) -> anyhow::Result<()> {
+pub struct DeleteTodo {
-    sqlx::query("DELETE FROM todos WHERE id = ?")
+    pub id: i64,
-        .bind(id)
-        .execute(db)
-        .await?;
-    Ok(())
 }
+
+livevue_rs::mutation_exec!(DeleteTodo {
+    sql: "DELETE FROM todos WHERE id = ?",
+    params: [|s: &DeleteTodo| s.id],
+    publish: |_| CacheKey::Table { table: "todos" },
+});

src/context.rs

@@ -197,6 +197,33 @@ impl RenderContext {
         self.store.kv.get(&key)
     }
 
+    // -----------------------------------------------------------------------
+    // PostgreSQL queries
+    // -----------------------------------------------------------------------
+
+    /// Execute a PostgreSQL query, automatically registering its subscription key.
+    ///
+    /// This is the pg equivalent of `cx.run()`. The query's `cache_key()` is
+    /// pushed into the subscription set so the connection rerenders whenever
+    /// the WAL listener publishes that key.
+    ///
+    /// ```ignore
+    /// let messages = cx.run_pg(&rs.pg, ListMessages).await?;
+    /// // No manual cx.subscribe() needed.
+    /// ```
+    #[cfg(feature = "pg_replication")]
+    pub async fn run_pg<Q: crate::pg_query::PgQuery>(
+        &mut self,
+        pool: &sqlx::PgPool,
+        query: Q,
+    ) -> anyhow::Result<Q::Output> {
+        let key = query.cache_key();
+        if key != CacheKey::None {
+            self.subscription_keys.push(key);
+        }
+        query.execute(pool).await
+    }
+
     // -----------------------------------------------------------------------
     // Manual subscriptions
     // -----------------------------------------------------------------------

src/lib.rs

@@ -15,9 +15,18 @@ pub mod server;
 #[cfg(feature = "pg_replication")]
 pub mod pg_replication;
 
+/// PostgreSQL query trait and macros (`pg_query_one!`, `pg_query_all!`).
+///
+/// Use `cx.run_pg(pool, query)` in render functions to execute a `PgQuery`
+/// and auto-subscribe to its cache key, exactly like `cx.run()` for SQLite.
+#[cfg(feature = "pg_replication")]
+pub mod pg_query;
+
 pub use signal::{Signal, SignalStore, SignalOpts, mangle, client_only, global, data_signals};
 pub use query::{CacheKey, Query};
 pub use store::{Store, KVStore};
+#[cfg(feature = "pg_replication")]
+pub use pg_query::PgQuery;
 pub use context::{RenderContext, ComponentTree, ComponentNode};
 pub use connection::{ConnectionManager, ConnectionState, SubscriptionRegistry};
 pub use sse::{format_patch_elements, format_patch_signals};

src/pg_query.rs

@@ -0,0 +1,133 @@
+use std::future::Future;
+use std::pin::Pin;
+
+use crate::query::CacheKey;
+
+// ---------------------------------------------------------------------------
+// PgQuery trait
+// ---------------------------------------------------------------------------
+
+/// The data-access abstraction for PostgreSQL-backed queries.
+///
+/// Mirrors the SQLite [`Query`](crate::Query) trait but binds to
+/// `sqlx::PgPool`. Use `cx.run_pg(pool, query)` in render functions — it
+/// calls `cache_key()` to register the subscription automatically, exactly
+/// like `cx.run()` does for SQLite queries.
+///
+/// Pg mutations do **not** implement this trait and do **not** call
+/// `publish_key()` — the WAL replication listener fires invalidation events
+/// automatically for any write from any source.
+///
+/// The `pg_query_one!` / `pg_query_all!` macros generate impls of this trait.
+pub trait PgQuery: Send + Sync {
+    type Output: Send;
+
+    fn cache_key(&self) -> CacheKey;
+
+    fn execute<'a>(
+        &'a self,
+        db: &'a sqlx::PgPool,
+    ) -> Pin<Box<dyn Future<Output = anyhow::Result<Self::Output>> + Send + 'a>>;
+}
+
+// ---------------------------------------------------------------------------
+// pg_query_one! — fetch exactly one row from PostgreSQL
+// ---------------------------------------------------------------------------
+
+/// Generate a `PgQuery` impl that fetches a single row from PostgreSQL.
+///
+/// ```ignore
+/// pub struct GetMessage { pub id: i64 }
+///
+/// pg_query_one!(GetMessage {
+///     sql: "SELECT id, body FROM messages WHERE id = $1",
+///     params: [|s: &GetMessage| s.id],
+///     output: Message,
+///     cache: |s: &GetMessage| pg_table_key("public", "messages"),
+/// });
+/// ```
+#[cfg(feature = "pg_replication")]
+#[macro_export]
+macro_rules! pg_query_one {
+    ($name:ident {
+        sql: $sql:expr,
+        params: [$($param:expr),* $(,)?],
+        output: $output:ty,
+        cache: $cache:expr $(,)?
+    }) => {
+        impl $crate::PgQuery for $name {
+            type Output = $output;
+
+            fn cache_key(&self) -> $crate::CacheKey {
+                ($cache)(self)
+            }
+
+            fn execute<'a>(
+                &'a self,
+                db: &'a sqlx::PgPool,
+            ) -> ::std::pin::Pin<
+                Box<dyn ::std::future::Future<Output = anyhow::Result<Self::Output>> + Send + 'a>,
+            > {
+                let zelf = self;
+                Box::pin(async move {
+                    let row = sqlx::query_as::<_, $output>($sql)
+                        $(.bind(($param)(zelf)))*
+                        .fetch_one(db)
+                        .await?;
+                    Ok(row)
+                })
+            }
+        }
+    };
+}
+
+// ---------------------------------------------------------------------------
+// pg_query_all! — fetch zero or more rows from PostgreSQL
+// ---------------------------------------------------------------------------
+
+/// Generate a `PgQuery` impl that fetches multiple rows from PostgreSQL.
+///
+/// ```ignore
+/// pub struct ListMessages;
+///
+/// pg_query_all!(ListMessages {
+///     sql: "SELECT id, body FROM messages ORDER BY id DESC",
+///     params: [],
+///     output: Message,
+///     cache: |_| pg_table_key("public", "messages"),
+/// });
+/// ```
+#[cfg(feature = "pg_replication")]
+#[macro_export]
+macro_rules! pg_query_all {
+    ($name:ident {
+        sql: $sql:expr,
+        params: [$($param:expr),* $(,)?],
+        output: $output:ty,
+        cache: $cache:expr $(,)?
+    }) => {
+        impl $crate::PgQuery for $name {
+            type Output = Vec<$output>;
+
+            fn cache_key(&self) -> $crate::CacheKey {
+                ($cache)(self)
+            }
+
+            fn execute<'a>(
+                &'a self,
+                db: &'a sqlx::PgPool,
+            ) -> ::std::pin::Pin<
+                Box<dyn ::std::future::Future<Output = anyhow::Result<Self::Output>> + Send + 'a>,
+            > {
+                let zelf = self;
+                Box::pin(async move {
+                    let rows = sqlx::query_as::<_, $output>($sql)
+                        $(.bind(($param)(zelf)))*
+                        .fetch_all(db)
+                        .await?;
+                    Ok(rows)
+                })
+            }
+        }
+    };
+}

src/query.rs

@@ -52,6 +52,11 @@ pub enum CacheKey {
 /// 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.
+///
+/// Mutation structs (generated by `mutation_exec!`) also implement this trait.
+/// They override `publish_key()` to return the `CacheKey` that should be
+/// invalidated after the write. `Store::exec()` calls `publish_key()` and
+/// fires the broadcast automatically — no manual `store.publish()` needed.
 pub trait Query: Send + Sync {
     type Output: Send;
 
@@ -61,6 +66,14 @@ pub trait Query: Send + Sync {
         &'a self,
         db: &'a sqlx::SqlitePool,
     ) -> Pin<Box<dyn Future<Output = anyhow::Result<Self::Output>> + Send + 'a>>;
+
+    /// The cache key to publish after a successful mutation.
+    ///
+    /// Read queries return `None` (the default). Mutation structs generated
+    /// by `mutation_exec!` override this to return the key they invalidate.
+    fn publish_key(&self) -> Option<CacheKey> {
+        None
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -113,6 +126,62 @@ macro_rules! query_one {
     };
 }
 
+// ---------------------------------------------------------------------------
+// mutation_exec! — INSERT / UPDATE / DELETE (no RETURNING)
+// ---------------------------------------------------------------------------
+
+/// Generate a `Query` impl for write operations (INSERT / UPDATE / DELETE).
+///
+/// The struct returns `()` from `execute` and overrides `publish_key()` with
+/// the provided closure. Use `Store::exec()` to run it — the key is published
+/// automatically after a successful write.
+///
+/// ```ignore
+/// pub struct CreateTodo { pub title: String }
+///
+/// mutation_exec!(CreateTodo {
+///     sql: "INSERT INTO todos (title, completed) VALUES (?, false)",
+///     params: [|s: &CreateTodo| s.title.as_str()],
+///     publish: |_| CacheKey::Table { table: "todos" },
+/// });
+/// ```
+#[macro_export]
+macro_rules! mutation_exec {
+    ($name:ident {
+        sql: $sql:expr,
+        params: [$($param:expr),* $(,)?],
+        publish: $publish:expr $(,)?
+    }) => {
+        impl $crate::Query for $name {
+            type Output = ();
+
+            fn cache_key(&self) -> $crate::CacheKey {
+                $crate::CacheKey::None
+            }
+
+            fn publish_key(&self) -> Option<$crate::CacheKey> {
+                Some(($publish)(self))
+            }
+
+            fn execute<'a>(
+                &'a self,
+                db: &'a sqlx::SqlitePool,
+            ) -> ::std::pin::Pin<
+                Box<dyn ::std::future::Future<Output = anyhow::Result<Self::Output>> + Send + 'a>,
+            > {
+                let zelf = self;
+                Box::pin(async move {
+                    sqlx::query($sql)
+                        $(.bind(($param)(zelf)))*
+                        .execute(db)
+                        .await?;
+                    Ok(())
+                })
+            }
+        }
+    };
+}
+
 // ---------------------------------------------------------------------------
 // query_all! — fetch zero or more rows
 // ---------------------------------------------------------------------------

src/store.rs

@@ -137,4 +137,21 @@ impl Store {
     pub fn subscribe(&self) -> broadcast::Receiver<CacheKey> {
         self.events.subscribe()
     }
+
+    /// Execute a mutation query and auto-publish its invalidation key.
+    ///
+    /// Use this in action handlers instead of calling the plain async mutation
+    /// function and then `store.publish(key)` separately. The mutation's
+    /// `publish_key()` drives the fanout automatically.
+    ///
+    /// ```ignore
+    /// state.store.exec(CreateTodo { title }).await?;
+    /// ```
+    pub async fn exec<Q: crate::query::Query>(&self, query: Q) -> anyhow::Result<Q::Output> {
+        let result = query.execute(&self.db).await?;
+        if let Some(key) = query.publish_key() {
+            self.publish(key);
+        }
+        Ok(result)
+    }
 }