Merge pull request #33 from pmorris-dev/sqlite-observer-phase-2

sub(feat): implement the base API and logic

Author: jjhafer

Cargo.lock

@@ -20,6 +20,7 @@ bundled = ["libsqlite3-sys/bundled"]
 
 [dependencies]
 tokio = { version = "1.49.0", features = ["sync"] }
+tokio-stream = { version = "0.1", features = ["sync"] }
 thiserror = "2.0.17"
 tracing = { version = "0.1.44", default-features = false, features = ["std", "release_max_level_off"] }
 parking_lot = "0.12.3"
@@ -30,3 +31,6 @@ libsqlite3-sys = { version = "0.30.1", features = ["preupdate_hook"] }
 
 [dev-dependencies]
 tokio = { version = "1.49.0", features = ["full", "macros"] }
+futures = "0.3.31"
+tempfile = "3.24.0"
+tracing-subscriber = "0.3.22"

crates/sqlx-sqlite-observer/Cargo.toml

@@ -98,14 +98,14 @@ This ensures subscribers **only receive notifications for committed changes**.
    * **`ChangeOperation`**: Insert, Update, or Delete
    * **`ColumnValue`**: Typed column value (Null, Integer, Real, Text, Blob)
    * **`ObserverConfig`**: Configuration for table filtering and channel
-     capacity <!-- COMING SOON -->
+     capacity
 
-### Observer Types <!-- COMING SOON -->
+### Observer Types
 
    * **`SqliteObserver`**: Main observer for `SqlitePool` connections
    * **`ObservableConnection`**: Connection wrapper with hooks registered
 
-### Stream Types <!-- COMING SOON -->
+### Stream Types
 
    * **`TableChangeStream`**: Async stream of table changes
    * **`TableChangeStreamExt`**: Extension trait for converting receivers to
@@ -177,19 +177,110 @@ meaningful/correct for non-integer or composite primary keys.
 
 ## Examples
 
-> **Coming in Phase 2** - Full working examples will be added in a subsequent PR.
-
 ### Basic Usage
 
-<!-- TODO: Add basic example showing SqliteObserver usage -->
+```rust,no_run
+use sqlx::SqlitePool;
+use sqlx_sqlite_observer::{SqliteObserver, ObserverConfig, ColumnValue};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let pool = SqlitePool::connect("sqlite:mydb.db").await?;
+    let observer = SqliteObserver::new(pool, ObserverConfig::default());
+
+    // Subscribe to changes on specific tables
+    let mut rx = observer.subscribe(["users"]);
+
+    // Spawn a task to handle notifications
+    tokio::spawn(async move {
+        while let Ok(change) = rx.recv().await {
+            println!(
+                "Table {} row {} was {:?}",
+                change.table,
+                change.rowid.unwrap_or(-1),
+                change.operation
+            );
+            if let Some(ColumnValue::Integer(id)) = change.primary_key.first() {
+                println!("  PK: {}", id);
+            }
+        }
+    });
+
+    // Use the observer to execute queries
+    let mut conn = observer.acquire().await?;
+    sqlx::query("INSERT INTO users (name) VALUES (?)")
+        .bind("Alice")
+        .execute(&mut **conn)
+        .await?;
+
+    Ok(())
+}
+```
 
 ### Stream API
 
-<!-- TODO: Add stream example showing TableChangeStream usage -->
+```rust,no_run
+use futures::StreamExt;
+use sqlx::SqlitePool;
+use sqlx_sqlite_observer::{SqliteObserver, ObserverConfig};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let pool = SqlitePool::connect("sqlite:mydb.db").await?;
+    let config = ObserverConfig::new().with_tables(["users", "posts"]);
+    let observer = SqliteObserver::new(pool, config);
+
+    let mut stream = observer.subscribe_stream(["users"]);
+
+    while let Some(change) = stream.next().await {
+        println!(
+            "Table {} row {} was {:?}",
+            change.table,
+            change.rowid.unwrap_or(-1),
+            change.operation
+        );
+    }
+
+    Ok(())
+}
+```
 
 ### Value Capture
 
-<!-- TODO: Add example showing old/new column value access -->
+```rust,no_run
+use sqlx::SqlitePool;
+use sqlx_sqlite_observer::{SqliteObserver, ObserverConfig, ColumnValue};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let pool = SqlitePool::connect("sqlite:mydb.db").await?;
+    let config = ObserverConfig::new().with_tables(["users"]);
+    let observer = SqliteObserver::new(pool, config);
+
+    let mut rx = observer.subscribe(["users"]);
+    let change = rx.recv().await?;
+
+    // Access old/new column values
+    if let Some(old) = &change.old_values {
+        println!("Old values: {:?}", old);
+    }
+    if let Some(new) = &change.new_values {
+        println!("New values: {:?}", new);
+    }
+
+    // Disable value capture for lower memory usage
+    let config = ObserverConfig::new()
+        .with_tables(["users"])
+        .with_capture_values(false);
+    let observer = SqliteObserver::new(
+        SqlitePool::connect("sqlite:mydb.db").await?,
+        config,
+    );
+    // old_values and new_values will be None
+
+    Ok(())
+}
+```
 
 ### SQLx SQLite Connection Manager Integration
 
@@ -204,8 +295,6 @@ buffered. All changes in a transaction are delivered at once on commit. If your
 transaction contains more mutating statements than this capacity, **messages
 will be dropped**.
 
-<!-- COMING SOON -->
-
 ```rust
 let config = ObserverConfig::new()
     .with_tables(["users", "posts"])
@@ -217,8 +306,6 @@ let config = ObserverConfig::new()
 By default, `TableChange` includes `old_values` and `new_values` with the actual
 column data. Disable this for lower memory usage if you only need row IDs:
 
-<!-- COMING SOON -->
-
 ```rust
 let config = ObserverConfig::new()
     .with_tables(["users"])

crates/sqlx-sqlite-observer/README.md

@@ -91,6 +91,40 @@ impl ObservationBroker {
       self.table_info.write().insert(table.to_string(), info);
    }
 
+   /// Registers multiple tables for observation without schema info.
+   ///
+   /// This is a two-phase registration: tables are marked for observation immediately,
+   /// but primary key extraction will return empty `Vec` until [`set_table_info`] is
+   /// called for each table. This is useful when you want to register tables before
+   /// their schema is known (e.g., before the first connection is acquired).
+   ///
+   /// **Prefer [`observe_table`] when schema info is available**, as it atomically
+   /// registers the table and sets schema info in one call.
+   ///
+   /// [`set_table_info`]: Self::set_table_info
+   /// [`observe_table`]: Self::observe_table
+   pub fn observe_tables<I, S>(&self, tables: I)
+   where
+      I: IntoIterator<Item = S>,
+      S: AsRef<str>,
+   {
+      let mut observed = self.observed_tables.write();
+      for table in tables {
+         let table_name = table.as_ref().to_string();
+         trace!(table = %table_name, "Observing table");
+         observed.insert(table_name);
+      }
+   }
+
+   /// Sets the schema information for an observed table.
+   ///
+   /// This information is used to extract primary key values and determine
+   /// whether the rowid is meaningful for the table.
+   pub fn set_table_info(&self, table: &str, info: TableInfo) {
+      trace!(table = %table, pk_columns = ?info.pk_columns, without_rowid = info.without_rowid, "Setting table info");
+      self.table_info.write().insert(table.to_string(), info);
+   }
+
    /// Gets the schema information for an observed table.
    pub fn get_table_info(&self, table: &str) -> Option<TableInfo> {
       self.table_info.read().get(table).cloned()