diff --git a/docker-compose.yml b/examples/pg_replication/docker-compose.yml similarity index 85% rename from docker-compose.yml rename to examples/pg_replication/docker-compose.yml index 2efddd9..6c980c2 100644 --- a/docker-compose.yml +++ b/examples/pg_replication/docker-compose.yml @@ -11,7 +11,7 @@ services: - "5432:5432" volumes: - postgres_data:/var/lib/postgresql/data - # Enable logical replication — required for the pg_replication example. + - ./pg_hba.conf:/etc/postgresql/pg_hba.conf command: - postgres - -c @@ -20,11 +20,13 @@ services: - max_replication_slots=10 - -c - max_wal_senders=10 + - -c + - hba_file=/etc/postgresql/pg_hba.conf healthcheck: test: ["CMD-SHELL", "pg_isready -U postgres -d livevue"] interval: 5s timeout: 5s retries: 5 - volumes: postgres_data: + diff --git a/examples/pg_replication/main.rs b/examples/pg_replication/main.rs index 2f06ce2..88716a3 100644 --- a/examples/pg_replication/main.rs +++ b/examples/pg_replication/main.rs @@ -37,13 +37,12 @@ use axum::{ Router, }; use maud::{html, Markup, DOCTYPE}; -use serde::Deserialize; use sqlx::FromRow; use uuid::Uuid; use livevue_rs::{ pg_replication::{pg_table_key, PgReplicationConfig, PgReplicationListener}, - server::{ingest_signals, ActionParams, AppState, SharedState}, + server::{ingest_signals, patch_signal, AppState, SharedState}, spawn_fanout, RenderContext, Store, }; @@ -112,6 +111,7 @@ async fn render_page(cx: &mut RenderContext, rs: &RenderState) -> anyhow::Result .await?; let new_msg = cx.signal("newMsg", livevue_rs::global()); + let _conn_id_signal = cx.signal("connId", livevue_rs::global()); Ok(html! { div #app { @@ -127,11 +127,10 @@ async fn render_page(cx: &mut RenderContext, rs: &RenderState) -> anyhow::Result type="text" placeholder="Type a message…" data-bind=(new_msg.name) - data-on:keydown="{if (event.key === 'Enter') $post('/action/post_message')}" + data-on:keydown="{if (event.key === 'Enter') @post('/action/post_message')}" ; button - data-on:click="$post('/action/post_message')" - data-attr=(format!("{{disabled: {}.trim() === ''}}", new_msg.val)) + data-on:click="@post('/action/post_message')" { "Send" } } @@ -179,16 +178,14 @@ fn page_shell(conn_id: Uuid, signals_json: &str, inner: &Markup) -> String { // DataStar SDK (loaded from CDN for the example) script type="module" - src="https://cdn.jsdelivr.net/gh/starfederation/datastar@v1.0.0-beta.11/bundles/datastar.js" + src="https://cdn.jsdelivr.net/gh/starfederation/datastar@1.0.0-RC.8/bundles/datastar.js" {} } - body { - div - data-signals=(signals_json) - data-on:load=(format!("@get('/sse?conn={conn_id}')")) - { - (inner) - } + body + data-signals=(signals_json) + data-init=(format!("@get('/sse?conn={conn_id}')")) + { + (inner) } } } @@ -202,7 +199,7 @@ fn page_shell(conn_id: Uuid, signals_json: &str, inner: &Markup) -> String { async fn index_handler(State(state): State<(SharedState, RenderState)>) -> impl IntoResponse { let (app_state, rs) = &state; let conn_id = Uuid::new_v4(); - let signals_json = r#"{"newMsg":""}"#; + let signals_json = &format!(r#"{{"newMsg":"","connId":"{}"}}"#, conn_id); app_state.connections.insert(conn_id, { // We need a sender for the connection, but the SSE handler will replace // it. Use a dummy channel that immediately drops. @@ -214,7 +211,7 @@ async fn index_handler(State(state): State<(SharedState, RenderState)>) -> impl .update_signals(&conn_id, signals_json.to_string()); let mut cx = - RenderContext::new(conn_id, livevue_rs::SignalStore::from_json(serde_json::json!({"newMsg": ""})), Default::default(), app_state.store.clone()); + RenderContext::new(conn_id, livevue_rs::SignalStore::from_json(serde_json::json!({"newMsg": "", "connId": conn_id.to_string()})), Default::default(), app_state.store.clone()); let inner = render_page(&mut cx, rs).await.unwrap_or_else(|e| { html! { p { "Render error: " (e) } } @@ -232,18 +229,17 @@ async fn sse_adapter( livevue_rs::sse_handler(State(state.0), query).await } -#[derive(Deserialize)] -struct PostMessageBody { - #[serde(rename = "newMsg")] - new_msg: Option, -} - async fn post_message_handler( State(state): State<(SharedState, RenderState)>, - axum::extract::Query(params): axum::extract::Query, Json(body): Json, ) -> impl IntoResponse { - let signals = ingest_signals(&state.0, params.conn, body); + let conn_id = body + .get("connId") + .and_then(|v| v.as_str()) + .and_then(|s| s.parse().ok()) + .unwrap_or_else(Uuid::new_v4); + + let signals = ingest_signals(&state.0, conn_id, body); let msg: String = signals .get("newMsg") .and_then(|v| v.as_str().map(|s| s.trim().to_string())) @@ -262,6 +258,9 @@ async fn post_message_handler( return StatusCode::INTERNAL_SERVER_ERROR; } + // Update the stored signals to reset the input field + patch_signal(&state.0, &conn_id, "newMsg", serde_json::json!("")); + // Note: we do NOT call store.publish() here — the PgReplicationListener // will detect the INSERT through the WAL stream and fire the invalidation // automatically. This demonstrates the key value of the module: mutations @@ -384,4 +383,4 @@ async fn main() -> anyhow::Result<()> { axum::serve(listener_tcp, app).await?; Ok(()) -} +} \ No newline at end of file diff --git a/examples/pg_replication/pg_hba.conf b/examples/pg_replication/pg_hba.conf new file mode 100644 index 0000000..f54e7e9 --- /dev/null +++ b/examples/pg_replication/pg_hba.conf @@ -0,0 +1,4 @@ +# docker/pg_hba.conf +local all all trust +host all all 0.0.0.0/0 trust +host replication all 0.0.0.0/0 trust diff --git a/maybe_sql_integration/README.md b/maybe_sql_integration/README.md deleted file mode 100644 index 679d946..0000000 --- a/maybe_sql_integration/README.md +++ /dev/null @@ -1,358 +0,0 @@ -# PostgreSQL Logical Replication Cache Invalidation System - -A high-performance cache invalidation system that streams PostgreSQL logical replication messages, buffers changes per transaction, and emits **transaction-atomic** cache invalidation events via Tokio broadcast channels. - -## Key Advantage: Transaction Atomicity - -Changes are buffered until commit, so all changes in a transaction are delivered to subscribers in a single `CacheInvalidationEvent`. This guarantees: - -- **No partial state**: If a transaction changes 5 rows across 3 tables, subscribers see all 5 changes together -- **Correct semantics**: Subscribers can invalidate cache atomically and notify clients in a single batch -- **Simpler logic**: No need to track partial transactions or worry about out-of-order delivery -- **Crash recovery**: Can checkpoint at commit boundary with full event - -## Architecture - -### Three Core Components - -#### 1. **pg_wal_parser.rs** - Logical Replication Message Parser -Low-level binary protocol parser implementing the pgoutput format spec. - -**Handles:** -- All PostgreSQL 14+ logical replication message types (Begin, Relation, Insert, Update, Delete, Commit, Truncate, Type, Message) -- TupleData parsing with column values (Null, UnchangedToasted, Text, Binary) -- Streaming protocol with relation OID caching -- Multi-version protocol support via graceful degradation - -**Key Types:** -- `LogicalMessage` - enum of all message types with their data -- `MessageParser` - stateful parser that maintains relation metadata -- `ColumnInfo`, `RelationInfo` - schema metadata from Relation messages - -#### 2. **cache_invalidation.rs** - Transaction-Aware Event Emitter -Transforms low-level replication messages into transaction-atomic high-level events. - -**Handles:** -- Per-transaction event buffering (keyed by XID) -- Tracks current transaction via `Begin`/`Commit` boundary -- Accumulates Insert/Update/Delete/Truncate changes -- Emits single atomic event on Commit with all changes -- Row data extraction with column names and values -- Primary key identification (via REPLICA IDENTITY flags) -- Tokio broadcast channel emission - -**Key Types:** -- `CacheInvalidationEvent` - contains XID, LSNs, timestamp, and **all changes in transaction** -- `Change` - a single table operation (Insert/Update/Delete/Truncate) with row data -- `CacheValue` - serializable column values -- `RowData` - structured row with column map and optional keys -- `TransactionBuffer` - internal state for buffering per XID - -**Entry Point:** -```rust -let mut emitter = CacheInvalidationEmitter::new(tx); -emitter.process_raw(&raw_wal_bytes).await?; -``` - -#### 3. **example.rs** - pgwire-replication Integration -Shows how to wire everything together with the pgwire-replication crate. - -**Handles:** -- Connecting to PostgreSQL logical replication -- Processing XLogData events from replication stream -- Graceful error handling and LSN checkpointing -- Example subscriber pattern for transaction-atomic cache invalidation -- Multiple concurrent subscribers - -## Data Flow - -``` -PostgreSQL WAL (multi-statement transaction) - ↓ -pgwire-replication (ReplicationClient) - ↓ -XLogData { bytes, wal_end } - ↓ -MessageParser::parse_message() [binary protocol parsing] - ↓ -LogicalMessage enum - ↓ -CacheInvalidationEmitter::process_message() [buffering per XID] - ↓ -Begin: Create buffer for XID -Insert/Update/Delete: Append to buffer for XID -Commit: Emit CacheInvalidationEvent with all buffered changes - ↓ -CacheInvalidationEvent { xid, commit_lsn, changes: [Change, Change, ...] } - ↓ -broadcast channel → multiple subscribers receive transaction atomically -``` - -## Example Transaction Flow - -PostgreSQL executes: -```sql -BEGIN; - INSERT INTO users (id, name) VALUES (1, 'alice'); - INSERT INTO posts (id, user_id, title) VALUES (100, 1, 'Hello'); - UPDATE posts SET views = 1 WHERE id = 100; -COMMIT; -``` - -Replication protocol sends: -``` -Begin XID=5000 LSN=0/123456 - Relation OID=16384 name="users" - Relation OID=16385 name="posts" - Insert rel=16384 [1, 'alice'] - Insert rel=16385 [100, 1, 'Hello'] - Update rel=16385 key=[100] new=[100, 1, 'Hello', 1] -Commit XID=5000 LSN=0/789ABC -``` - -Emitter produces (single event): -```rust -CacheInvalidationEvent { - xid: 5000, - commit_lsn: 0x789ABC, - end_lsn: 0x789ABC, - timestamp: 1234567890, - changes: [ - Change { - table: "users", - schema: "public", - operation: Insert, - row_data: Some({id: 1, name: 'alice'}) - }, - Change { - table: "posts", - schema: "public", - operation: Insert, - row_data: Some({id: 100, user_id: 1, title: 'Hello'}) - }, - Change { - table: "posts", - schema: "public", - operation: Update, - row_data: Some({ - columns: {id: 100, user_id: 1, title: 'Hello', views: 1}, - key_values: Some({id: 100}) - }) - } - ] -} -``` - -Subscribers receive this single event and can: -- Invalidate cache for users:1, posts:100 atomically -- Notify WebSocket clients in one batch -- Checkpoint the LSN -- Update metrics - -## Protocol Details - -### Message Format (from PostgreSQL docs) - -Each logical replication message starts with a type byte: - -| Type | Meaning | When | -|------|---------|------| -| B | Begin | Transaction start (1 per txn) | -| R | Relation | Schema definition (1+ per txn) | -| I | Insert | New row written to table | -| U | Update | Row modified (includes old/new) | -| D | Delete | Row removed (includes key/old) | -| C | Commit | Transaction committed (1 per txn) | -| T | Truncate | Table truncated | -| Y | Type | Custom data type definition | -| M | Message | Custom pg_logical_emit_message() | - -### Transaction Boundaries - -- **Begin** message: `XID` field identifies the transaction -- **Changes** (Insert/Update/Delete/Truncate): Include same `XID` -- **Commit** message: Completes the transaction (no XID in Commit itself) - -The emitter tracks `current_xid` to match Commit to its Begin, ensuring all changes are buffered and emitted together. - -## Setup & Usage - -### PostgreSQL Setup (one-time) - -```sql --- Enable logical replication -ALTER SYSTEM SET wal_level = logical; -SELECT pg_reload_conf(); - --- Create replication slot -SELECT * FROM pg_create_logical_replication_slot('my_slot', 'pgoutput'); - --- Create publication -CREATE PUBLICATION my_pub FOR TABLE users, posts, comments; -``` - -### Rust Code Integration - -```rust -use tokio::sync::broadcast; -use cache_invalidation::CacheInvalidationEmitter; - -#[tokio::main] -async fn main() -> Result<(), Box> { - let (tx, mut rx) = broadcast::channel(10000); - - // Spawn WAL listener (uses pgwire-replication) - tokio::spawn(async move { - let mut client = pgwire_replication::ReplicationClient::connect(config).await?; - let mut emitter = CacheInvalidationEmitter::new(tx); - - while let Some(event) = client.recv().await? { - match event { - pgwire_replication::ReplicationEvent::XLogData { data, wal_end, .. } => { - // Process all messages in data; buffered per transaction - emitter.process_raw(&data).await?; - client.update_applied_lsn(wal_end); - } - _ => {} - } - } - Ok::<(), Box>(()) - }); - - // Subscribe to cache invalidation events - loop { - match rx.recv().await { - Ok(event) => { - println!("Transaction XID={} committed with {} changes", - event.xid, event.changes.len()); - - // Invalidate cache for all changes atomically - for change in &event.changes { - match &change.operation { - Operation::Insert => { - // cache.delete_by_table(change.table) - } - Operation::Update => { - if let Some(row) = &change.row_data { - if let Some(keys) = &row.key_values { - // cache.delete(change.table, keys) - } - } - } - Operation::Delete => { - if let Some(row) = &change.row_data { - if let Some(keys) = &row.key_values { - // cache.delete(change.table, keys) - } - } - } - Operation::Truncate => { - // cache.clear_table(change.table) - } - } - } - - // Notify WebSocket clients once, with all changes - // websocket_broadcast(event) - } - Err(broadcast::error::RecvError::Lagged(_)) => { - eprintln!("Subscriber lagged, reloading cache from DB"); - } - Err(broadcast::error::RecvError::Closed) => break, - } - } - - Ok(()) -} -``` - -## Performance Characteristics - -- **Zero-copy parsing**: Uses `bytes::BytesMut` with in-place operations -- **Async-first**: Compatible with Tokio task scheduling -- **Streaming**: Processes messages as they arrive, no buffering of raw data -- **Transaction buffering**: Per-XID HashMap (typically 1-10 concurrent txns) -- **Memory**: Relation metadata cached by OID + per-txn change buffer -- **LSN tracking**: Explicit control for checkpointing and crash recovery - -## Typical Latencies - -- PostgreSQL writes change to WAL: < 1ms -- WAL transmitted to client: network latency (localhost ~0.1ms) -- Message parsing: ~10-100µs per message -- Buffering: ~1µs per change -- Commit event emission: ~10-100µs -- **End-to-end (localhost)**: ~1-5ms from commit to subscriber notification - -## Transaction Buffering Internals - -```rust -// On Begin: Create buffer -txn_buffers.insert(xid, TransactionBuffer { xid, changes: Vec::new() }); -current_xid = Some(xid); - -// On Insert/Update/Delete/Truncate: Append to buffer -if let Some(buf) = txn_buffers.get_mut(&xid) { - buf.changes.push(change); -} - -// On Commit: Emit and clean up -if let Some(xid) = current_xid.take() { - emit_transaction(xid, commit_lsn, end_lsn, timestamp)?; - txn_buffers.remove(&xid); -} -``` - -## Testing - -Each module has tests. Run with: - -```bash -cargo test -``` - -## Caveats & Notes - -### What This Doesn't Do -- Custom output plugins (uses standard pgoutput only) -- Two-phase commits (v3+) would need extended buffering -- Concurrent transaction correlation (assumes sequential Begins/Commits) - -### REPLICA IDENTITY Configuration - -For UPDATE/DELETE to include key information: -```sql -ALTER TABLE users REPLICA IDENTITY USING INDEX users_pkey; --- or for full old row: -ALTER TABLE users REPLICA IDENTITY FULL; -``` - -Default is PRIMARY KEY, which is usually what you want. - -### Slot Management - -Replication slots persist until explicitly dropped. If the stream falls behind, the WAL won't be deleted and disk will fill up: - -```sql --- Monitor slot lag -SELECT slot_name, active, restart_lsn, confirmed_flush_lsn -FROM pg_replication_slots; - --- Drop a slot when done -SELECT pg_drop_replication_slot('my_slot'); -``` - -## Future Enhancements - -1. **Metrics**: Instrument parsing and emit latency histograms -2. **Backpressure**: Honor broadcast channel capacity -3. **Checkpointing**: Persist LSN to disk/Redis for recovery -4. **Filtering**: Table/operation-level subscription filtering -5. **Parallel streams**: Support multiple concurrent WAL consumers - -## References - -- [PostgreSQL Logical Replication Protocol](https://www.postgresql.org/docs/current/protocol-logicalrep-message-formats.html) -- [pgoutput Output Plugin](https://www.postgresql.org/docs/current/sql-createpublication.html) -- [pgwire-replication Crate](https://crates.io/crates/pgwire-replication) -- [Tokio Broadcast Channel](https://tokio.rs/tokio/tutorial/select#broadcast) diff --git a/maybe_sql_integration/cache_invalidation.rs b/maybe_sql_integration/cache_invalidation.rs deleted file mode 100644 index 2a68da3..0000000 --- a/maybe_sql_integration/cache_invalidation.rs +++ /dev/null @@ -1,366 +0,0 @@ -use crate::pg_wal_parser::{ColumnValue, LogicalMessage, MessageParser, RelationInfo, TupleData}; -use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use tokio::sync::broadcast; - -/// High-level cache invalidation event emitted at transaction commit -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct CacheInvalidationEvent { - /// Transaction ID from PostgreSQL - pub xid: u32, - /// Commit LSN/watermark for checkpoint - pub commit_lsn: u64, - /// End LSN of the transaction - pub end_lsn: u64, - /// Timestamp of commit (microseconds since PG epoch) - pub timestamp: i64, - /// All operations in this transaction, emitted atomically - pub changes: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct Change { - /// Table name that changed - pub table: String, - /// Schema/namespace - pub schema: String, - /// Operation type - pub operation: Operation, - /// Row data if available - pub row_data: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub enum Operation { - Insert, - Update, - Delete, - Truncate, -} - -/// Simplified row representation for broadcasting -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct RowData { - /// Mapping of column name -> value - pub columns: HashMap, - /// Primary key values if available - pub key_values: Option>, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub enum CacheValue { - Null, - String(String), - Binary(Vec), - UnchangedToasted, -} - -/// Transaction buffer for a single transaction -struct TransactionBuffer { - xid: u32, - changes: Vec, -} - -pub struct CacheInvalidationEmitter { - parser: MessageParser, - tx: broadcast::Sender, - /// Buffers for active transactions, keyed by XID - txn_buffers: HashMap, - /// Track the current transaction being processed - current_xid: Option, -} - -impl CacheInvalidationEmitter { - pub fn new(tx: broadcast::Sender) -> Self { - Self { - parser: MessageParser::new(), - tx, - txn_buffers: HashMap::new(), - current_xid: None, - } - } - - /// Process a single logical replication message - /// Events are buffered until Commit, then emitted atomically - pub async fn process_message(&mut self, msg: LogicalMessage) -> Result<(), String> { - match msg { - LogicalMessage::Begin { xid, .. } => { - // Start buffering for this transaction - self.current_xid = Some(xid); - self.txn_buffers.insert(xid, TransactionBuffer { - xid, - changes: Vec::new(), - }); - Ok(()) - } - LogicalMessage::Relation(_) => { - // Relation updates internal state via parser - Ok(()) - } - LogicalMessage::Insert { - xid, - rel_oid, - new_tuple, - } => { - let change = self.build_insert_change(rel_oid, new_tuple)?; - self.buffer_change(xid, change); - Ok(()) - } - LogicalMessage::Update { - xid, - rel_oid, - key_tuple, - old_tuple: _, - new_tuple, - } => { - let change = self.build_update_change(rel_oid, new_tuple, key_tuple)?; - self.buffer_change(xid, change); - Ok(()) - } - LogicalMessage::Delete { - xid, - rel_oid, - key_tuple, - old_tuple, - } => { - let change = self.build_delete_change(rel_oid, key_tuple.or(old_tuple))?; - self.buffer_change(xid, change); - Ok(()) - } - LogicalMessage::Commit { - commit_lsn, - end_lsn, - timestamp, - .. - } => { - // Emit the buffered transaction atomically - if let Some(xid) = self.current_xid.take() { - self.emit_transaction(xid, commit_lsn, end_lsn, timestamp)?; - } - Ok(()) - } - LogicalMessage::Truncate { - xid, - rel_oids, - .. - } => { - for oid in rel_oids { - let change = self.build_truncate_change(oid)?; - self.buffer_change(xid, change); - } - Ok(()) - } - LogicalMessage::Type { .. } => { - // Type information, no action needed - Ok(()) - } - LogicalMessage::Message { .. } => { - // Custom messages, can be ignored for cache invalidation - Ok(()) - } - } - } - - fn buffer_change(&mut self, xid: u32, change: Change) { - if let Some(buf) = self.txn_buffers.get_mut(&xid) { - buf.changes.push(change); - } - } - - fn build_insert_change(&self, rel_oid: u32, new_tuple: TupleData) -> Result { - let relation = self - .parser - .get_relation(rel_oid) - .ok_or_else(|| format!("Relation OID {} not found", rel_oid))?; - - let row_data = self.build_row_data(relation, &new_tuple)?; - - Ok(Change { - table: relation.name.clone(), - schema: relation.namespace.clone(), - operation: Operation::Insert, - row_data: Some(row_data), - }) - } - - fn build_update_change( - &self, - rel_oid: u32, - new_tuple: TupleData, - key_tuple: Option, - ) -> Result { - let relation = self - .parser - .get_relation(rel_oid) - .ok_or_else(|| format!("Relation OID {} not found", rel_oid))?; - - let mut row_data = self.build_row_data(relation, &new_tuple)?; - - // If we have a key tuple, extract key values - if let Some(key) = key_tuple { - let key_values = self.extract_key_values(relation, &key)?; - row_data.key_values = Some(key_values); - } - - Ok(Change { - table: relation.name.clone(), - schema: relation.namespace.clone(), - operation: Operation::Update, - row_data: Some(row_data), - }) - } - - fn build_delete_change( - &self, - rel_oid: u32, - key_or_old: Option, - ) -> Result { - let relation = self - .parser - .get_relation(rel_oid) - .ok_or_else(|| format!("Relation OID {} not found", rel_oid))?; - - let mut row_data = None; - if let Some(tuple) = key_or_old { - let mut data = self.build_row_data(relation, &tuple)?; - let key_values = self.extract_key_values(relation, &tuple)?; - data.key_values = Some(key_values); - row_data = Some(data); - } - - Ok(Change { - table: relation.name.clone(), - schema: relation.namespace.clone(), - operation: Operation::Delete, - row_data, - }) - } - - fn build_truncate_change(&self, rel_oid: u32) -> Result { - let relation = self - .parser - .get_relation(rel_oid) - .ok_or_else(|| format!("Relation OID {} not found", rel_oid))?; - - Ok(Change { - table: relation.name.clone(), - schema: relation.namespace.clone(), - operation: Operation::Truncate, - row_data: None, - }) - } - - fn build_row_data(&self, relation: &RelationInfo, tuple: &TupleData) -> Result { - let mut columns = HashMap::new(); - - for (i, value) in tuple.values.iter().enumerate() { - if i >= relation.columns.len() { - return Err(format!( - "Tuple has more values than relation has columns: {} vs {}", - tuple.values.len(), - relation.columns.len() - )); - } - - let col = &relation.columns[i]; - let cache_value = match value { - ColumnValue::Null => CacheValue::Null, - ColumnValue::UnchangedToasted => CacheValue::UnchangedToasted, - ColumnValue::Text(s) => CacheValue::String(s.clone()), - ColumnValue::Binary(b) => CacheValue::Binary(b.clone()), - }; - - columns.insert(col.name.clone(), cache_value); - } - - Ok(RowData { - columns, - key_values: None, - }) - } - - fn extract_key_values( - &self, - relation: &RelationInfo, - key_tuple: &TupleData, - ) -> Result, String> { - let mut key_values = HashMap::new(); - - for (i, value) in key_tuple.values.iter().enumerate() { - if i >= relation.columns.len() { - break; // Key might have fewer columns - } - - let col = &relation.columns[i]; - if col.flags & 1 != 0 { - // This column is part of the key - let cache_value = match value { - ColumnValue::Null => CacheValue::Null, - ColumnValue::UnchangedToasted => CacheValue::UnchangedToasted, - ColumnValue::Text(s) => CacheValue::String(s.clone()), - ColumnValue::Binary(b) => CacheValue::Binary(b.clone()), - }; - - key_values.insert(col.name.clone(), cache_value); - } - } - - Ok(key_values) - } - - /// Emit a buffered transaction atomically - fn emit_transaction(&mut self, xid: u32, commit_lsn: u64, end_lsn: u64, timestamp: i64) -> Result<(), String> { - if let Some(buf) = self.txn_buffers.remove(&xid) { - let event = CacheInvalidationEvent { - xid, - commit_lsn, - end_lsn, - timestamp, - changes: buf.changes, - }; - - self.tx.send(event).ok(); // ignore if no receivers - } - - Ok(()) - } - - /// Parse raw replication data bytes and process messages - /// This is the main entry point from pgwire-replication - pub async fn process_raw(&mut self, data: &[u8]) -> Result<(), String> { - let mut buf = bytes::BytesMut::from(&data[..]); - - while !buf.is_empty() { - if let Some(msg) = self.parser.parse_message(&mut buf)? { - self.process_message(msg).await?; - } - } - - Ok(()) - } - - /// Get statistics about buffered transactions - pub fn buffered_transaction_count(&self) -> usize { - self.txn_buffers.len() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_emitter_creation() { - let (tx, _rx) = broadcast::channel(100); - let _emitter = CacheInvalidationEmitter::new(tx); - } - - #[test] - fn test_transaction_buffering() { - let (tx, _rx) = broadcast::channel(100); - let emitter = CacheInvalidationEmitter::new(tx); - - // Changes should be buffered per transaction until commit - assert_eq!(emitter.buffered_transaction_count(), 0); - } -} diff --git a/maybe_sql_integration/example.rs b/maybe_sql_integration/example.rs deleted file mode 100644 index 9cf4d91..0000000 --- a/maybe_sql_integration/example.rs +++ /dev/null @@ -1,259 +0,0 @@ -/// Example integration of pgwire-replication with cache invalidation -/// -/// This shows how to: -/// 1. Connect to PostgreSQL logical replication stream -/// 2. Parse logical messages -/// 3. Buffer changes per transaction -/// 4. Emit all changes atomically at commit via Tokio broadcast -/// 5. Subscribe to events for client notification - -use tokio::sync::broadcast; -use std::time::Duration; - -// These would be your actual modules -mod pg_wal_parser; -mod cache_invalidation; - -use cache_invalidation::{CacheInvalidationEvent, CacheInvalidationEmitter}; - -/// Example configuration -pub struct ReplicationConfig { - pub pg_host: String, - pub pg_port: u16, - pub pg_user: String, - pub pg_password: String, - pub pg_database: String, - pub replication_slot: String, - pub publication: String, - pub start_lsn: String, // e.g. "0/0" or "0/16B6C50" -} - -/// Main replication stream handler -pub struct WalStreamListener { - config: ReplicationConfig, - tx: broadcast::Sender, -} - -impl WalStreamListener { - pub fn new( - config: ReplicationConfig, - tx: broadcast::Sender, - ) -> Self { - Self { config, tx } - } - - /// Start the replication stream listener - /// In production, this would use pgwire-replication::ReplicationClient - pub async fn start(self) -> Result<(), Box> { - // This is a sketch - actual implementation would use pgwire-replication - - // 1. Create replication client - // let mut client = pgwire_replication::ReplicationClient::connect( - // pgwire_replication::ReplicationConfig { - // host: self.config.pg_host.into(), - // port: self.config.pg_port, - // user: self.config.pg_user.into(), - // password: self.config.pg_password.into(), - // database: self.config.pg_database.into(), - // slot: self.config.replication_slot.into(), - // publication: self.config.publication.into(), - // start_lsn: pgwire_replication::Lsn::parse(&self.config.start_lsn)?, - // status_interval: Duration::from_secs(10), - // idle_wakeup_interval: Duration::from_secs(10), - // ..Default::default() - // }, - // ).await?; - - // 2. Create emitter with transaction buffering - let mut emitter = CacheInvalidationEmitter::new(self.tx); - - // 3. Process stream - // loop { - // match client.recv().await? { - // Some(pgwire_replication::ReplicationEvent::XLogData { data, wal_end, .. }) => { - // // Parse and buffer changes per transaction - // emitter.process_raw(&data).await?; - // - // // Update LSN checkpoint after processing - // client.update_applied_lsn(wal_end); - // } - // Some(pgwire_replication::ReplicationEvent::KeepAlive { .. }) => { - // // Server keepalive, continue - // } - // Some(pgwire_replication::ReplicationEvent::StoppedAt { .. }) => { - // // Stream ended gracefully - // break; - // } - // Some(_) => {} - // None => break, - // } - // } - - Ok(()) - } -} - -/// Example client that subscribes to cache invalidation events -/// All changes in a transaction are emitted atomically on commit -pub async fn example_subscriber( - mut rx: broadcast::Receiver, -) -> Result<(), Box> { - loop { - match rx.recv().await { - Ok(event) => { - println!( - "Transaction committed: XID={} LSN={:016X}-{:016X} timestamp={} changes={}", - event.xid, event.commit_lsn, event.end_lsn, event.timestamp, event.changes.len() - ); - - // All changes in this transaction are now available atomically - for change in &event.changes { - match &change.operation { - cache_invalidation::Operation::Insert => { - println!(" INSERT {}.{}", change.schema, change.table); - } - cache_invalidation::Operation::Update => { - println!(" UPDATE {}.{}", change.schema, change.table); - } - cache_invalidation::Operation::Delete => { - println!(" DELETE {}.{}", change.schema, change.table); - } - cache_invalidation::Operation::Truncate => { - println!(" TRUNCATE {}.{}", change.schema, change.table); - } - } - - if let Some(row) = &change.row_data { - if let Some(keys) = &row.key_values { - print!(" Keys: "); - for (k, v) in keys { - print!("{}={:?} ", k, v); - } - println!(); - } - } - } - - // In a real system, you would: - // 1. Invalidate all cache entries for this transaction atomically - // 2. Notify WebSocket clients (single transaction message) - // 3. Store event for audit/replication trail - // 4. Update Prometheus metrics - // 5. Persist LSN for crash recovery - } - Err(broadcast::error::RecvError::Lagged(_)) => { - println!("Warning: Cache invalidation queue lagged, reloading cache from DB"); - } - Err(broadcast::error::RecvError::Closed) => { - println!("Cache invalidation stream closed"); - break; - } - } - } - - Ok(()) -} - -/// Example showing transaction atomicity -/// Demonstrates that a multi-table transaction is delivered as one event -async fn example_transaction_atomicity() { - println!("Example: Multi-table transaction"); - println!(); - println!("BEGIN"); - println!(" INSERT INTO users (id, name) VALUES (1, 'alice')"); - println!(" INSERT INTO posts (id, user_id, title) VALUES (100, 1, 'Hello')"); - println!(" INSERT INTO comments (id, post_id, text) VALUES (1000, 100, 'Nice!')"); - println!("COMMIT LSN=0/1234567 XID=5000"); - println!(); - println!("Result: Single CacheInvalidationEvent with 3 changes"); - println!(" - Change[0]: users.INSERT"); - println!(" - Change[1]: posts.INSERT"); - println!(" - Change[2]: comments.INSERT"); - println!(); - println!("Subscribers receive all 3 changes atomically,"); - println!("so they can update cache without partial state."); -} - -#[tokio::main] -async fn main() -> Result<(), Box> { - // In a real application, you might: - - // 1. Create broadcast channel for cache events (large capacity for bursty writes) - let (tx, rx) = broadcast::channel::(10000); - - // 2. Spawn WAL stream listener - let config = ReplicationConfig { - pg_host: "127.0.0.1".to_string(), - pg_port: 5432, - pg_user: "postgres".to_string(), - pg_password: "postgres".to_string(), - pg_database: "mydb".to_string(), - replication_slot: "my_slot".to_string(), - publication: "my_publication".to_string(), - start_lsn: "0/0".to_string(), - }; - - let listener = WalStreamListener::new(config, tx); - - tokio::spawn(async move { - if let Err(e) = listener.start().await { - eprintln!("Replication stream error: {}", e); - } - }); - - // 3. Spawn multiple subscribers (e.g., for different cache layers) - let rx_cache = tx.subscribe(); - tokio::spawn(async move { - if let Err(e) = example_subscriber(rx_cache).await { - eprintln!("Cache subscriber error: {}", e); - } - }); - - let rx_analytics = tx.subscribe(); - tokio::spawn(async move { - // Another subscriber could log for analytics - let mut rx = rx_analytics; - loop { - if let Ok(event) = rx.recv().await { - // Track metrics, e.g., changes per table per second - eprintln!("Metric: {} changes in XID {}", event.changes.len(), event.xid); - } else { - break; - } - } - }); - - // 4. Your application continues in the background - // The cache invalidation events flow continuously - - // Keep main alive - example_transaction_atomicity().await; - tokio::signal::ctrl_c().await?; - println!("Shutting down"); - - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn test_broadcast_channel() { - let (tx, mut rx) = broadcast::channel::(10); - - let event = CacheInvalidationEvent { - xid: 1000, - commit_lsn: 0x123456, - end_lsn: 0x654321, - timestamp: 1_000_000, - changes: vec![], - }; - - tx.send(event.clone()).unwrap(); - - let received = rx.recv().await.unwrap(); - assert_eq!(received.xid, 1000); - assert_eq!(received.changes.len(), 0); - } -} diff --git a/maybe_sql_integration/pg_wal_parser.rs b/maybe_sql_integration/pg_wal_parser.rs deleted file mode 100644 index 8c3d12f..0000000 --- a/maybe_sql_integration/pg_wal_parser.rs +++ /dev/null @@ -1,478 +0,0 @@ -use bytes::{Buf, BytesMut}; -use std::collections::HashMap; - -/// PostgreSQL logical replication message parser -/// Handles pgoutput protocol messages from replication streams - -#[derive(Debug, Clone)] -pub struct ColumnInfo { - pub flags: u8, // 1 if part of key, 0 otherwise - pub name: String, - pub type_oid: u32, - pub type_modifier: i32, -} - -#[derive(Debug, Clone)] -pub struct RelationInfo { - pub oid: u32, - pub namespace: String, - pub name: String, - pub replica_identity: u8, - pub columns: Vec, -} - -#[derive(Debug, Clone)] -pub enum ColumnValue { - Null, - UnchangedToasted, - Text(String), - Binary(Vec), -} - -#[derive(Debug)] -pub struct TupleData { - pub values: Vec, -} - -#[derive(Debug)] -pub enum LogicalMessage { - Begin { - final_lsn: u64, - timestamp: i64, - xid: u32, - }, - Relation(RelationInfo), - Insert { - xid: u32, - rel_oid: u32, - new_tuple: TupleData, - }, - Update { - xid: u32, - rel_oid: u32, - key_tuple: Option, - old_tuple: Option, - new_tuple: TupleData, - }, - Delete { - xid: u32, - rel_oid: u32, - key_tuple: Option, - old_tuple: Option, - }, - Commit { - flags: u8, - commit_lsn: u64, - end_lsn: u64, - timestamp: i64, - }, - Type { - xid: u32, - type_oid: u32, - namespace: String, - type_name: String, - }, - Truncate { - xid: u32, - num_relations: u32, - options: u8, - rel_oids: Vec, - }, - Message { - xid: u32, - flags: u8, - lsn: u64, - prefix: String, - content: Vec, - }, -} - -pub struct MessageParser { - /// Map of relation OID -> RelationInfo for decoding tuples - relations: HashMap, - /// Map of type OID -> type name for potential future use - types: HashMap, -} - -impl MessageParser { - pub fn new() -> Self { - Self { - relations: HashMap::new(), - types: HashMap::new(), - } - } - - /// Parse a single logical replication message from bytes - /// Assumes input starts with the message type byte - pub fn parse_message(&mut self, input: &mut BytesMut) -> Result, String> { - if input.is_empty() { - return Ok(None); - } - - let msg_type = input[0] as char; - - match msg_type { - 'B' => self.parse_begin(input), - 'I' => self.parse_insert(input), - 'U' => self.parse_update(input), - 'D' => self.parse_delete(input), - 'C' => self.parse_commit(input), - 'R' => { - self.parse_relation(input)?; - Ok(None) // Relation messages are stored internally - } - 'Y' => self.parse_type(input), - 'T' => self.parse_truncate(input), - 'M' => self.parse_message_msg(input), - 't' => Err("Tuple data outside of Insert/Update/Delete (unexpected)".to_string()), - _ => Err(format!("Unknown message type: {} (byte: {})", msg_type, input[0])), - } - } - - fn read_string(buf: &mut BytesMut) -> Result { - // PostgreSQL strings are null-terminated - let mut vec = Vec::new(); - loop { - if buf.is_empty() { - return Err("EOF while reading string".to_string()); - } - let byte = buf.get_u8(); - if byte == 0 { - break; - } - vec.push(byte); - } - String::from_utf8(vec).map_err(|e| format!("Invalid UTF-8 in string: {}", e)) - } - - fn read_i64(buf: &mut BytesMut) -> Result { - if buf.len() < 8 { - return Err("Not enough bytes for i64".to_string()); - } - Ok(buf.get_i64()) - } - - fn read_u64(buf: &mut BytesMut) -> Result { - if buf.len() < 8 { - return Err("Not enough bytes for u64".to_string()); - } - Ok(buf.get_u64()) - } - - fn read_i32(buf: &mut BytesMut) -> Result { - if buf.len() < 4 { - return Err("Not enough bytes for i32".to_string()); - } - Ok(buf.get_i32()) - } - - fn read_u32(buf: &mut BytesMut) -> Result { - if buf.len() < 4 { - return Err("Not enough bytes for u32".to_string()); - } - Ok(buf.get_u32()) - } - - fn read_i16(buf: &mut BytesMut) -> Result { - if buf.len() < 2 { - return Err("Not enough bytes for i16".to_string()); - } - Ok(buf.get_i16()) - } - - fn read_u8(buf: &mut BytesMut) -> Result { - if buf.is_empty() { - return Err("Not enough bytes for u8".to_string()); - } - Ok(buf.get_u8()) - } - - fn parse_begin(&mut self, buf: &mut BytesMut) -> Result, String> { - buf.get_u8(); // 'B' - let final_lsn = Self::read_u64(buf)?; - let timestamp = Self::read_i64(buf)?; - let xid = Self::read_u32(buf)?; - - Ok(Some(LogicalMessage::Begin { - final_lsn, - timestamp, - xid, - })) - } - - fn parse_commit(&mut self, buf: &mut BytesMut) -> Result, String> { - buf.get_u8(); // 'C' - let flags = Self::read_u8(buf)?; - let commit_lsn = Self::read_u64(buf)?; - let end_lsn = Self::read_u64(buf)?; - let timestamp = Self::read_i64(buf)?; - - Ok(Some(LogicalMessage::Commit { - flags, - commit_lsn, - end_lsn, - timestamp, - })) - } - - fn parse_relation(&mut self, buf: &mut BytesMut) -> Result<(), String> { - buf.get_u8(); // 'R' - // xid only present for streamed transactions (version 2+), skip for now - let rel_oid = Self::read_u32(buf)?; - let namespace = Self::read_string(buf)?; - let name = Self::read_string(buf)?; - let replica_identity = Self::read_u8(buf)?; - let num_columns = Self::read_i16(buf)? as usize; - - let mut columns = Vec::with_capacity(num_columns); - for _ in 0..num_columns { - let flags = Self::read_u8(buf)?; - let col_name = Self::read_string(buf)?; - let type_oid = Self::read_u32(buf)?; - let type_modifier = Self::read_i32(buf)?; - - columns.push(ColumnInfo { - flags, - name: col_name, - type_oid, - type_modifier, - }); - } - - let relation = RelationInfo { - oid: rel_oid, - namespace, - name, - replica_identity, - columns, - }; - - self.relations.insert(rel_oid, relation); - Ok(()) - } - - fn parse_insert(&mut self, buf: &mut BytesMut) -> Result, String> { - buf.get_u8(); // 'I' - let xid = Self::read_u32(buf)?; - let rel_oid = Self::read_u32(buf)?; - - // Next byte must be 'N' for new tuple - let marker = Self::read_u8(buf)?; - if marker != b'N' { - return Err(format!("Expected 'N' in INSERT, got: {}", marker as char)); - } - - let new_tuple = self.parse_tuple_data(buf, rel_oid)?; - - Ok(Some(LogicalMessage::Insert { - xid, - rel_oid, - new_tuple, - })) - } - - fn parse_update(&mut self, buf: &mut BytesMut) -> Result, String> { - buf.get_u8(); // 'U' - let xid = Self::read_u32(buf)?; - let rel_oid = Self::read_u32(buf)?; - - let mut key_tuple = None; - let mut old_tuple = None; - - // Optional K or O markers - if !buf.is_empty() { - match buf[0] as char { - 'K' => { - buf.get_u8(); // consume 'K' - key_tuple = Some(self.parse_tuple_data(buf, rel_oid)?); - } - 'O' => { - buf.get_u8(); // consume 'O' - old_tuple = Some(self.parse_tuple_data(buf, rel_oid)?); - } - _ => {} // neither K nor O, proceed to N - } - } - - let marker = Self::read_u8(buf)?; - if marker != b'N' { - return Err(format!("Expected 'N' in UPDATE, got: {}", marker as char)); - } - - let new_tuple = self.parse_tuple_data(buf, rel_oid)?; - - Ok(Some(LogicalMessage::Update { - xid, - rel_oid, - key_tuple, - old_tuple, - new_tuple, - })) - } - - fn parse_delete(&mut self, buf: &mut BytesMut) -> Result, String> { - buf.get_u8(); // 'D' - let xid = Self::read_u32(buf)?; - let rel_oid = Self::read_u32(buf)?; - - let mut key_tuple = None; - let mut old_tuple = None; - - // Either K or O (but not both) - if !buf.is_empty() { - match buf[0] as char { - 'K' => { - buf.get_u8(); // consume 'K' - key_tuple = Some(self.parse_tuple_data(buf, rel_oid)?); - } - 'O' => { - buf.get_u8(); // consume 'O' - old_tuple = Some(self.parse_tuple_data(buf, rel_oid)?); - } - _ => return Err(format!("Expected K or O in DELETE, got: {}", buf[0] as char)), - } - } - - Ok(Some(LogicalMessage::Delete { - xid, - rel_oid, - key_tuple, - old_tuple, - })) - } - - fn parse_type(&mut self, buf: &mut BytesMut) -> Result, String> { - buf.get_u8(); // 'Y' - let type_oid = Self::read_u32(buf)?; - let namespace = Self::read_string(buf)?; - let type_name = Self::read_string(buf)?; - let xid = 0; // Type doesn't have xid in base protocol - - self.types - .insert(type_oid, (namespace.clone(), type_name.clone())); - - Ok(Some(LogicalMessage::Type { - xid, - type_oid, - namespace, - type_name, - })) - } - - fn parse_truncate(&mut self, buf: &mut BytesMut) -> Result, String> { - buf.get_u8(); // 'T' - let xid = Self::read_u32(buf)?; - let num_relations = Self::read_u32(buf)?; - let options = Self::read_u8(buf)?; - - let mut rel_oids = Vec::with_capacity(num_relations as usize); - for _ in 0..num_relations { - rel_oids.push(Self::read_u32(buf)?); - } - - Ok(Some(LogicalMessage::Truncate { - xid, - num_relations, - options, - rel_oids, - })) - } - - fn parse_message_msg(&mut self, buf: &mut BytesMut) -> Result, String> { - buf.get_u8(); // 'M' - let xid = Self::read_u32(buf)?; - let flags = Self::read_u8(buf)?; - let lsn = Self::read_u64(buf)?; - let prefix = Self::read_string(buf)?; - let content_len = Self::read_i32(buf)? as usize; - - if buf.len() < content_len { - return Err(format!( - "Not enough bytes for message content: {} < {}", - buf.len(), - content_len - )); - } - - let content = buf.split_to(content_len).to_vec(); - - Ok(Some(LogicalMessage::Message { - xid, - flags, - lsn, - prefix, - content, - })) - } - - fn parse_tuple_data(&self, buf: &mut BytesMut, rel_oid: u32) -> Result { - let num_columns = Self::read_i16(buf)? as usize; - let mut values = Vec::with_capacity(num_columns); - - let relation = self - .relations - .get(&rel_oid) - .ok_or_else(|| format!("Unknown relation OID: {}", rel_oid))?; - - if num_columns != relation.columns.len() { - return Err(format!( - "Column count mismatch: {} vs {}", - num_columns, - relation.columns.len() - )); - } - - for _ in 0..num_columns { - let marker = Self::read_u8(buf)?; - - let value = match marker as char { - 'n' => ColumnValue::Null, - 'u' => ColumnValue::UnchangedToasted, - 't' => { - let len = Self::read_i32(buf)? as usize; - if buf.len() < len { - return Err("Not enough bytes for text column".to_string()); - } - let data = buf.split_to(len).to_vec(); - let text = String::from_utf8(data) - .map_err(|e| format!("Invalid UTF-8 in column: {}", e))?; - ColumnValue::Text(text) - } - 'b' => { - let len = Self::read_i32(buf)? as usize; - if buf.len() < len { - return Err("Not enough bytes for binary column".to_string()); - } - let data = buf.split_to(len).to_vec(); - ColumnValue::Binary(data) - } - _ => return Err(format!("Unknown column marker: {}", marker as char)), - }; - - values.push(value); - } - - Ok(TupleData { values }) - } - - /// Get a relation by OID (used for cache invalidation mapping) - pub fn get_relation(&self, oid: u32) -> Option<&RelationInfo> { - self.relations.get(&oid) - } - - /// Get all cached relations - pub fn relations(&self) -> &HashMap { - &self.relations - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_message_parser_creation() { - let parser = MessageParser::new(); - assert!(parser.relations.is_empty()); - } -} diff --git a/src/server.rs b/src/server.rs index 9fd7b8b..9de0fdb 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,8 +1,8 @@ use crate::connection::{ConnectionManager, SubscriptionRegistry}; use crate::context::{ComponentTree, RenderContext}; +use crate::query::CacheKey; 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}; @@ -38,7 +38,11 @@ use uuid::Uuid; /// Ok((html, keys)) /// })); /// ``` -pub type RenderFn = Arc BoxFuture<'static, anyhow::Result<(String, Vec)>> + Send + Sync>; +pub type RenderFn = Arc< + dyn Fn(RenderContext) -> BoxFuture<'static, anyhow::Result<(String, Vec)>> + + Send + + Sync, +>; // --------------------------------------------------------------------------- // AppState @@ -109,7 +113,9 @@ pub fn spawn_fanout(state: SharedState) { let conn_ids = state.subscriptions.get_connections_for_key(&key); for conn_id in conn_ids { - let signals_json = state.connections.get_signals(&conn_id) + let signals_json = state + .connections + .get_signals(&conn_id) .unwrap_or_else(|| "{}".to_string()); let Some(sse_tx) = state.connections.get_sender(&conn_id) else { continue; @@ -129,12 +135,7 @@ pub fn spawn_fanout(state: SharedState) { .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 mut cx = RenderContext::new(conn_id, signals, tree, state.store.clone()); let render_fn = state.render_fn.clone(); let state_ref = state.clone(); @@ -154,11 +155,19 @@ pub fn spawn_fanout(state: SharedState) { .collect::>() .join("\n"); - let event = Event::default() + let html_event = Event::default() .event("datastar-patch-elements") .data(data_payload); - if sse_tx.send(event).await.is_err() { + if sse_tx.send(html_event).await.is_err() { + tracing::debug!("fanout: SSE channel closed for {}", conn_id); + } + + let signals_event = Event::default() + .event("datastar-patch-signals") + .data("signals ".to_owned() + &signals_json); + + if sse_tx.send(signals_event).await.is_err() { tracing::debug!("fanout: SSE channel closed for {}", conn_id); } } @@ -209,7 +218,9 @@ pub async fn sse_handler( println!("SSE Handler called"); // Trigger first render immediately using the pre-seeded signals. - let signals_json = state.connections.get_signals(&conn_id) + let signals_json = state + .connections + .get_signals(&conn_id) .unwrap_or_else(|| "{}".to_string()); let render_fn = state.render_fn.clone(); @@ -247,12 +258,9 @@ pub async fn sse_handler( Err(e) => tracing::error!("sse_handler: initial render failed: {}", e), } }); - let stream = ReceiverStream::new(rx); - let event_stream = tokio_stream::StreamExt::map(stream, |event| { - Ok::(event) - }); + let event_stream = tokio_stream::StreamExt::map(stream, |event| Ok::(event)); Sse::new(event_stream).keep_alive(KeepAlive::default()) } @@ -294,14 +302,22 @@ pub struct ActionParams { /// StatusCode::ACCEPTED /// } /// ``` -pub fn ingest_signals( - state: &AppState, - conn_id: Uuid, - body: serde_json::Value, -) -> SignalStore { +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 +} + +pub fn patch_signal(state: &AppState, conn_id: &Uuid, key: &str, value: serde_json::Value) { + let current = state + .connections + .get_signals(conn_id) + .unwrap_or_else(|| "{}".to_string()); + let mut store = SignalStore::from_json(serde_json::from_str(¤t).unwrap_or_default()); + store.insert(key.to_string(), value); + state + .connections + .update_signals(conn_id, store.to_json_string()); +} diff --git a/src/sse.rs b/src/sse.rs index 5725708..1d9accc 100644 --- a/src/sse.rs +++ b/src/sse.rs @@ -30,6 +30,7 @@ pub fn format_patch_elements(html: &str) -> String { event } + /// Format a `datastar-patch-signals` SSE event. /// /// `signals_js` should be a JS object literal string, e.g. `{foo: '', bar: 0}`.