From 9dcb0d6dc8fe2b4e3613e4eeea197e8d6afc79d7 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 11 Mar 2026 22:37:33 +0000 Subject: [PATCH] Refactor Query contract: uniform ergonomics for SQLite + PostgreSQL MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 26 ++++--- examples/todo/app.rs | 82 +++++++------------- examples/todo/queries.rs | 52 +++++++------ src/context.rs | 27 +++++++ src/lib.rs | 9 +++ src/pg_query.rs | 133 ++++++++++++++++++++++++++++++++ src/query.rs | 69 +++++++++++++++++ src/store.rs | 17 ++++ 8 files changed, 331 insertions(+), 84 deletions(-) create mode 100644 src/pg_query.rs diff --git a/examples/pg_replication/main.rs b/examples/pg_replication/main.rs index 88716a3..884835b 100644 --- a/examples/pg_replication/main.rs +++ b/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 { - // Subscribe to PG-sourced invalidation events for the messages table. - // The fanout task will re-render this connection whenever the WAL emitter - // fires a `CacheKey::Channel { key: "pg:public.messages" }` event. - cx.subscribe(pg_table_key("public", "messages")); - - let messages: Vec = sqlx::query_as("SELECT id, body FROM messages ORDER BY id DESC") - .fetch_all(&rs.pg) - .await?; + // Fetch messages and auto-subscribe to the WAL invalidation key. + let messages = cx.run_pg(&rs.pg, ListMessages).await?; let new_msg = cx.signal("newMsg", livevue_rs::global()); let _conn_id_signal = cx.signal("connId", livevue_rs::global()); diff --git a/examples/todo/app.rs b/examples/todo/app.rs index 76246fe..7d08439 100644 --- a/examples/todo/app.rs +++ b/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> + 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, Json(body): Json, ) -> 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( - &state, - body, - async move { - if !title.is_empty() { - crate::queries::create_todo(&db, &title).await?; - } - Ok(()) - }, - CacheKey::Table { table: "todos" }, - ) - .await + if title.is_empty() { + return StatusCode::ACCEPTED; + } + + if let Err(e) = state.store.exec(crate::queries::CreateTodo { title }).await { + eprintln!("action_add_todo error: {e}"); + return StatusCode::INTERNAL_SERVER_ERROR; + } + + StatusCode::ACCEPTED } async fn action_toggle_todo( @@ -206,16 +186,15 @@ async fn action_toggle_todo( AxumQuery(params): AxumQuery, Json(body): Json, ) -> impl IntoResponse { - let db = state.store.db.clone(); - let id = params.id; + let conn_id = extract_conn_and_tree(&SignalStore::from_json(body.clone())).0; + ingest_signals(&state, conn_id, body); - action_then_publish( - &state, - body, - async move { crate::queries::toggle_todo(&db, id).await }, - CacheKey::Table { table: "todos" }, - ) - .await + if let Err(e) = state.store.exec(crate::queries::ToggleTodo { id: params.id }).await { + eprintln!("action_toggle_todo error: {e}"); + return StatusCode::INTERNAL_SERVER_ERROR; + } + + StatusCode::ACCEPTED } async fn action_delete_todo( @@ -223,14 +202,13 @@ async fn action_delete_todo( AxumQuery(params): AxumQuery, Json(body): Json, ) -> impl IntoResponse { - let db = state.store.db.clone(); - let id = params.id; + let conn_id = extract_conn_and_tree(&SignalStore::from_json(body.clone())).0; + ingest_signals(&state, conn_id, body); - action_then_publish( - &state, - body, - async move { crate::queries::delete_todo(&db, id).await }, - CacheKey::Table { table: "todos" }, - ) - .await + if let Err(e) = state.store.exec(crate::queries::DeleteTodo { id: params.id }).await { + eprintln!("action_delete_todo error: {e}"); + return StatusCode::INTERNAL_SERVER_ERROR; + } + + StatusCode::ACCEPTED } \ No newline at end of file diff --git a/examples/todo/queries.rs b/examples/todo/queries.rs index f692032..aceecb9 100644 --- a/examples/todo/queries.rs +++ b/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 -// produce subscription keys. They're called directly by action handlers, -// which then trigger a full rerender. +// Each mutation struct implements Query with publish_key() set to the cache +// key it invalidates. Use Store::exec() in action handlers — it executes the +// mutation and publishes the key automatically. -/// Insert a new todo. Returns the new row's id. -pub async fn create_todo(db: &sqlx::SqlitePool, title: &str) -> anyhow::Result { - let result = sqlx::query("INSERT INTO todos (title, completed) VALUES (?, false)") - .bind(title) - .execute(db) - .await?; - Ok(result.last_insert_rowid()) +/// Insert a new todo. +pub struct CreateTodo { + pub title: String, } +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<()> { - sqlx::query("UPDATE todos SET completed = NOT completed WHERE id = ?") - .bind(id) - .execute(db) - .await?; - Ok(()) +pub struct ToggleTodo { + pub id: i64, } +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<()> { - sqlx::query("DELETE FROM todos WHERE id = ?") - .bind(id) - .execute(db) - .await?; - Ok(()) +pub struct DeleteTodo { + pub id: i64, } + +livevue_rs::mutation_exec!(DeleteTodo { + sql: "DELETE FROM todos WHERE id = ?", + params: [|s: &DeleteTodo| s.id], + publish: |_| CacheKey::Table { table: "todos" }, +}); diff --git a/src/context.rs b/src/context.rs index 00cc17e..0a261c9 100644 --- a/src/context.rs +++ b/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( + &mut self, + pool: &sqlx::PgPool, + query: Q, + ) -> anyhow::Result { + let key = query.cache_key(); + if key != CacheKey::None { + self.subscription_keys.push(key); + } + query.execute(pool).await + } + // ----------------------------------------------------------------------- // Manual subscriptions // ----------------------------------------------------------------------- diff --git a/src/lib.rs b/src/lib.rs index 7a6d34e..ea7c829 100644 --- a/src/lib.rs +++ b/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}; diff --git a/src/pg_query.rs b/src/pg_query.rs new file mode 100644 index 0000000..a75cd9d --- /dev/null +++ b/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> + 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> + 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> + 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) + }) + } + } + }; +} diff --git a/src/query.rs b/src/query.rs index e650d0a..003a600 100644 --- a/src/query.rs +++ b/src/query.rs @@ -52,6 +52,11 @@ pub enum CacheKey { /// The trait uses `Pin>` 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> + 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 { + 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> + 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 // --------------------------------------------------------------------------- diff --git a/src/store.rs b/src/store.rs index ed3c304..96df5c4 100644 --- a/src/store.rs +++ b/src/store.rs @@ -137,4 +137,21 @@ impl Store { pub fn subscribe(&self) -> broadcast::Receiver { 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(&self, query: Q) -> anyhow::Result { + let result = query.execute(&self.db).await?; + if let Some(key) = query.publish_key() { + self.publish(key); + } + Ok(result) + } } \ No newline at end of file