fix up pg replication example

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

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-on:click="@post('/action/post_message')"
-                    data-attr=(format!("{{disabled: {}.trim() === ''}}", new_msg.val))
                     { "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 {
+            body
-                div
+                data-signals=(signals_json)
-                    data-signals=(signals_json)
+                data-init=(format!("@get('/sse?conn={conn_id}')"))
-                    data-on:load=(format!("@get('/sse?conn={conn_id}')"))
+            {
-                {
+                (inner)
-                    (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<String>,
-}
-
 async fn post_message_handler(
     State(state): State<(SharedState, RenderState)>,
-    axum::extract::Query(params): axum::extract::Query<ActionParams>,
     Json(body): Json<serde_json::Value>,
 ) -> 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(())
-}
+}

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

maybe_sql_integration/README.md

@@ -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<dyn std::error::Error>> {
-    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<dyn std::error::Error>>(())
-    });
-
-    // 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)

maybe_sql_integration/cache_invalidation.rs

@@ -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<Change>,
-}
-
-#[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<RowData>,
-}
-
-#[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<String, CacheValue>,
-    /// Primary key values if available
-    pub key_values: Option<HashMap<String, CacheValue>>,
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub enum CacheValue {
-    Null,
-    String(String),
-    Binary(Vec<u8>),
-    UnchangedToasted,
-}
-
-/// Transaction buffer for a single transaction
-struct TransactionBuffer {
-    xid: u32,
-    changes: Vec<Change>,
-}
-
-pub struct CacheInvalidationEmitter {
-    parser: MessageParser,
-    tx: broadcast::Sender<CacheInvalidationEvent>,
-    /// Buffers for active transactions, keyed by XID
-    txn_buffers: HashMap<u32, TransactionBuffer>,
-    /// Track the current transaction being processed
-    current_xid: Option<u32>,
-}
-
-impl CacheInvalidationEmitter {
-    pub fn new(tx: broadcast::Sender<CacheInvalidationEvent>) -> 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<Change, String> {
-        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<TupleData>,
-    ) -> Result<Change, String> {
-        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<TupleData>,
-    ) -> Result<Change, String> {
-        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<Change, String> {
-        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<RowData, String> {
-        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<HashMap<String, CacheValue>, 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);
-    }
-}

maybe_sql_integration/example.rs

@@ -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<CacheInvalidationEvent>,
-}
-
-impl WalStreamListener {
-    pub fn new(
-        config: ReplicationConfig,
-        tx: broadcast::Sender<CacheInvalidationEvent>,
-    ) -> Self {
-        Self { config, tx }
-    }
-
-    /// Start the replication stream listener
-    /// In production, this would use pgwire-replication::ReplicationClient
-    pub async fn start(self) -> Result<(), Box<dyn std::error::Error>> {
-        // 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<CacheInvalidationEvent>,
-) -> Result<(), Box<dyn std::error::Error>> {
-    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<dyn std::error::Error>> {
-    // In a real application, you might:
-    
-    // 1. Create broadcast channel for cache events (large capacity for bursty writes)
-    let (tx, rx) = broadcast::channel::<CacheInvalidationEvent>(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::<CacheInvalidationEvent>(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);
-    }
-}

maybe_sql_integration/pg_wal_parser.rs

@@ -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<ColumnInfo>,
-}
-
-#[derive(Debug, Clone)]
-pub enum ColumnValue {
-    Null,
-    UnchangedToasted,
-    Text(String),
-    Binary(Vec<u8>),
-}
-
-#[derive(Debug)]
-pub struct TupleData {
-    pub values: Vec<ColumnValue>,
-}
-
-#[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<TupleData>,
-        old_tuple: Option<TupleData>,
-        new_tuple: TupleData,
-    },
-    Delete {
-        xid: u32,
-        rel_oid: u32,
-        key_tuple: Option<TupleData>,
-        old_tuple: Option<TupleData>,
-    },
-    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<u32>,
-    },
-    Message {
-        xid: u32,
-        flags: u8,
-        lsn: u64,
-        prefix: String,
-        content: Vec<u8>,
-    },
-}
-
-pub struct MessageParser {
-    /// Map of relation OID -> RelationInfo for decoding tuples
-    relations: HashMap<u32, RelationInfo>,
-    /// Map of type OID -> type name for potential future use
-    types: HashMap<u32, (String, String)>,
-}
-
-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<Option<LogicalMessage>, 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<String, String> {
-        // 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<i64, String> {
-        if buf.len() < 8 {
-            return Err("Not enough bytes for i64".to_string());
-        }
-        Ok(buf.get_i64())
-    }
-
-    fn read_u64(buf: &mut BytesMut) -> Result<u64, String> {
-        if buf.len() < 8 {
-            return Err("Not enough bytes for u64".to_string());
-        }
-        Ok(buf.get_u64())
-    }
-
-    fn read_i32(buf: &mut BytesMut) -> Result<i32, String> {
-        if buf.len() < 4 {
-            return Err("Not enough bytes for i32".to_string());
-        }
-        Ok(buf.get_i32())
-    }
-
-    fn read_u32(buf: &mut BytesMut) -> Result<u32, String> {
-        if buf.len() < 4 {
-            return Err("Not enough bytes for u32".to_string());
-        }
-        Ok(buf.get_u32())
-    }
-
-    fn read_i16(buf: &mut BytesMut) -> Result<i16, String> {
-        if buf.len() < 2 {
-            return Err("Not enough bytes for i16".to_string());
-        }
-        Ok(buf.get_i16())
-    }
-
-    fn read_u8(buf: &mut BytesMut) -> Result<u8, String> {
-        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<Option<LogicalMessage>, 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<Option<LogicalMessage>, 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<Option<LogicalMessage>, 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<Option<LogicalMessage>, 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<Option<LogicalMessage>, 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<Option<LogicalMessage>, 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<Option<LogicalMessage>, 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<Option<LogicalMessage>, 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<TupleData, String> {
-        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<u32, RelationInfo> {
-        &self.relations
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_message_parser_creation() {
-        let parser = MessageParser::new();
-        assert!(parser.relations.is_empty());
-    }
-}

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<dyn Fn(RenderContext) -> BoxFuture<'static, anyhow::Result<(String, Vec<CacheKey>)>> + Send + Sync>;
+pub type RenderFn = Arc<
+    dyn Fn(RenderContext) -> BoxFuture<'static, anyhow::Result<(String, Vec<CacheKey>)>>
+        + 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(
+                let mut cx = RenderContext::new(conn_id, signals, tree, state.store.clone());
-                    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::<Vec<_>>()
                                 .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| {
+    let event_stream = tokio_stream::StreamExt::map(stream, |event| Ok::<Event, Infallible>(event));
-        Ok::<Event, Infallible>(event)
-    });
 
     Sse::new(event_stream).keep_alive(KeepAlive::default())
 }
@@ -294,14 +302,22 @@ pub struct ActionParams {
 ///     StatusCode::ACCEPTED
 /// }
 /// ```
-pub fn ingest_signals(
+pub fn ingest_signals(state: &AppState, conn_id: Uuid, body: serde_json::Value) -> SignalStore {
-    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
-}
+}
+
+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(&current).unwrap_or_default());
+    store.insert(key.to_string(), value);
+    state
+        .connections
+        .update_signals(conn_id, store.to_json_string());
+}

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}`.