sub(feat): include primary key in change msg

- It's the PK that the user really needs and that might not
be the rowid
- If the table schema includes WITHOUT ROWID, the first/only
column making up the PK is returned. That's hokey. So now
we return None for rowid in that case
- Needed to add some introspection logic to find which
column(s) make up the PK and whether the table was created
using WITHOUT ROWID

Author: pmorris-dev

Cargo.lock

@@ -23,6 +23,8 @@ tokio = { version = "1.49.0", 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"
+regex = "1.12.3"
+sqlx = { version = "0.8.6", features = ["sqlite", "runtime-tokio"], default-features = false }
 # Required for preupdate_hook - SQLite must be compiled with SQLITE_ENABLE_PREUPDATE_HOOK
 libsqlite3-sys = { version = "0.30", features = ["preupdate_hook"] }
 

crates/sqlx-sqlite-observer/Cargo.toml

@@ -116,6 +116,65 @@ This ensures subscribers **only receive notifications for committed changes**.
    * **`ObservableSqliteDatabase`**: Wrapper for `SqliteDatabase` with observation
    * **`ObservableWriteGuard`**: Write guard with hooks registered
 
+### `TableInfo`
+
+Schema information for observed tables (used internally, also exported).
+
+   * `pk_columns: Vec<usize>` - Column indices forming the primary key
+   * `without_rowid: bool` - Whether the table uses WITHOUT ROWID
+
+## Primary Key Extraction
+
+The `primary_key` field on `TableChange` always contains the actual primary key
+value(s) for the affected row:
+
+```rust
+let change = rx.recv().await?;
+
+// Single-column PK (e.g., INTEGER PRIMARY KEY)
+if let Some(ColumnValue::Integer(id)) = change.primary_key.first() {
+    println!("Changed row id: {}", id);
+}
+
+// Composite PK - values are in declaration order
+for (i, pk_value) in change.primary_key.iter().enumerate() {
+    println!("PK column {}: {:?}", i, pk_value);
+}
+```
+
+**Why `primary_key` instead of just `rowid`?**
+
+SQLite's internal `rowid` works well for tables with `INTEGER PRIMARY KEY`, but
+has limitations:
+
+   * **Text or UUID primary keys**: The `rowid` is an internal integer, not your
+     actual key
+   * **Composite primary keys**: The `rowid` doesn't represent your multi-column
+     key
+   * **WITHOUT ROWID tables**: The `rowid` from the preupdate hook is unreliable
+
+The `primary_key` field extracts the actual primary key values from the captured
+column data, giving you meaningful identifiers regardless of table structure.
+
+### WITHOUT ROWID Tables
+
+For tables created with `WITHOUT ROWID`, the `rowid` field in `TableChange` will
+be `None`:
+
+```rust
+let change = rx.recv().await?;
+
+if change.rowid.is_none() {
+    // This is a WITHOUT ROWID table
+    // Use primary_key instead
+    println!("PK: {:?}", change.primary_key);
+}
+```
+
+This is because SQLite's preupdate hook provides the first PRIMARY KEY column
+(coerced to i64) as the "rowid" for WITHOUT ROWID tables, which may not be
+meaningful/correct for non-integer or composite primary keys.
+
 ## Examples
 
 > **Coming in Phase 2** - Full working examples will be added in a subsequent PR.

crates/sqlx-sqlite-observer/README.md

@@ -30,19 +30,18 @@
 //! ```
 //!
 //! Changes captured by the preupdate hook are buffered until the transaction
-//! completes. On commit, buffered changes are published to subscribers. On
-//! rollback (or implicit rollback due to an error), they are discarded without
-//! notification.
+//! (explicit or implicit) completes. On commit, buffered changes are published
+//! to subscribers. On rollback, they are discarded without notification.
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use std::time::Instant;
 
 use parking_lot::{Mutex, RwLock};
 use tokio::sync::broadcast;
-use tracing::{debug, trace};
+use tracing::{debug, error, trace};
 
-use crate::change::{ChangeOperation, TableChange};
+use crate::change::{ChangeOperation, ColumnValue, TableChange, TableInfo};
 use crate::hooks::{PreUpdateEvent, SqliteValue};
 
 /// Transaction-aware observation broker.
@@ -54,6 +53,7 @@ pub struct ObservationBroker {
    buffer: Mutex<Vec<PreUpdateEvent>>,
    change_tx: broadcast::Sender<TableChange>,
    observed_tables: RwLock<HashSet<String>>,
+   table_info: RwLock<HashMap<String, TableInfo>>,
    capture_values: bool,
 }
 
@@ -65,6 +65,7 @@ impl ObservationBroker {
          buffer: Mutex::new(Vec::new()),
          change_tx,
          observed_tables: RwLock::new(HashSet::new()),
+         table_info: RwLock::new(HashMap::new()),
          capture_values,
       })
    }
@@ -74,20 +75,25 @@ impl ObservationBroker {
       self.observed_tables.read().contains(table)
    }
 
-   /// Registers tables for observation.
+   /// Registers a table for observation with its schema information.
    ///
    /// Only changes to observed tables will be buffered and published.
-   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);
-      }
+   /// The `TableInfo` is required to correctly extract primary key values
+   /// and determine whether the rowid is meaningful for the table.
+   pub fn observe_table(&self, table: &str, info: TableInfo) {
+      trace!(
+         table = %table,
+         pk_columns = ?info.pk_columns,
+         without_rowid = info.without_rowid,
+         "Observing table with schema info"
+      );
+      self.observed_tables.write().insert(table.to_string());
+      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()
    }
 
    /// Returns a list of all observed tables.
@@ -125,8 +131,14 @@ impl ObservationBroker {
       debug!(count = events.len(), "Flushing buffered changes on commit");
 
       for event in events {
-         let table_change = self.event_to_change(event);
-         let _ = self.change_tx.send(table_change);
+         match self.event_to_change(event) {
+            Ok(table_change) => {
+               let _ = self.change_tx.send(table_change);
+            }
+            Err(e) => {
+               error!(error = %e, "Failed to convert event to change");
+            }
+         }
       }
    }
 
@@ -155,13 +167,22 @@ impl ObservationBroker {
    }
 
    /// Converts a PreUpdateEvent to a TableChange for broadcast.
-   fn event_to_change(&self, event: PreUpdateEvent) -> TableChange {
-      let rowid = match event.operation {
-         ChangeOperation::Insert => Some(event.new_rowid),
-         ChangeOperation::Delete => Some(event.old_rowid),
-         ChangeOperation::Update => Some(event.new_rowid),
+   fn event_to_change(&self, event: PreUpdateEvent) -> crate::Result<TableChange> {
+      let table_info = self.table_info.read().get(&event.table).cloned();
+
+      // For WITHOUT ROWID tables, the rowid from preupdate hook is not meaningful
+      let rowid = match &table_info {
+         Some(info) if info.without_rowid => None,
+         _ => match event.operation {
+            ChangeOperation::Insert => Some(event.new_rowid),
+            ChangeOperation::Delete => Some(event.old_rowid),
+            ChangeOperation::Update => Some(event.new_rowid),
+         },
       };
 
+      // Extract primary key values from the appropriate column values
+      let primary_key = self.extract_primary_key(&event, table_info.as_ref())?;
+
       let (old_values, new_values) = if self.capture_values {
          (
             event.old_values.map(Self::values_to_vec),
@@ -171,14 +192,59 @@ impl ObservationBroker {
          (None, None)
       };
 
-      TableChange {
+      Ok(TableChange {
          table: event.table,
          operation: Some(event.operation),
          rowid,
+         primary_key,
          old_values,
          new_values,
          timestamp: Instant::now(),
+      })
+   }
+
+   /// Extracts primary key values from the event based on table schema.
+   ///
+   /// Returns an error if the schema has drifted (e.g., table was altered)
+   /// and PK column indices are out of bounds.
+   fn extract_primary_key(
+      &self,
+      event: &PreUpdateEvent,
+      table_info: Option<&TableInfo>,
+   ) -> crate::Result<Vec<ColumnValue>> {
+      let Some(info) = table_info else {
+         return Ok(Vec::new());
+      };
+
+      if info.pk_columns.is_empty() {
+         return Ok(Vec::new());
+      }
+
+      // For DELETE, use old values; for INSERT/UPDATE, use new values
+      let values = match event.operation {
+         ChangeOperation::Delete => event.old_values.as_ref(),
+         ChangeOperation::Insert | ChangeOperation::Update => event.new_values.as_ref(),
+      };
+
+      let Some(values) = values else {
+         return Ok(Vec::new());
+      };
+
+      // Extract values at the PK column indices, erroring if any index is out of bounds
+      let mut pk_values = Vec::with_capacity(info.pk_columns.len());
+      for &idx in &info.pk_columns {
+         match values.get(idx) {
+            Some(v) => pk_values.push(v.clone().into()),
+            None => {
+               return Err(crate::Error::SchemaMismatch {
+                  table: event.table.clone(),
+                  expected: info.pk_columns.len(),
+                  actual: values.len(),
+               });
+            }
+         }
       }
+      Ok(pk_values)
    }
 
    /// Converts SqliteValue vec to ColumnValue vec for TableChange.

crates/sqlx-sqlite-observer/src/broker.rs

@@ -2,6 +2,32 @@ use std::time::Instant;
 
 use crate::hooks::SqliteValue;
 
+/// Schema information for an observed table.
+///
+/// Used to extract primary key values and determine if rowid is meaningful.
+#[derive(Debug, Clone, Default)]
+pub struct TableInfo {
+   /// Column indices that form the primary key, in declaration order.
+   /// For `INTEGER PRIMARY KEY` tables, this contains the single PK column index.
+   /// For composite PKs, indices are ordered as declared in the schema.
+   pub pk_columns: Vec<usize>,
+   /// True if the table was created with `WITHOUT ROWID`.
+   /// For such tables, the preupdate hook's rowid parameter contains the first column
+   /// of the PRIMARY KEY (coerced to i64), which may not be meaningful/correct for
+   /// non-integer or composite primary keys.
+   pub without_rowid: bool,
+}
+
+impl TableInfo {
+   /// Creates a new TableInfo with the given PK column indices.
+   pub fn new(pk_columns: Vec<usize>, without_rowid: bool) -> Self {
+      Self {
+         pk_columns,
+         without_rowid,
+      }
+   }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum ChangeOperation {
    Insert,
@@ -83,7 +109,14 @@ impl ColumnValue {
 pub struct TableChange {
    pub table: String,
    pub operation: Option<ChangeOperation>,
+   /// The SQLite internal rowid. This is `None` for WITHOUT ROWID tables
+   /// since the preupdate hook's rowid parameter is not meaningful for them.
    pub rowid: Option<i64>,
+   /// The primary key value(s) for the affected row.
+   /// For composite primary keys, values are ordered by their declaration order.
+   /// For DELETE operations, this contains the old PK values.
+   /// For INSERT/UPDATE operations, this contains the new PK values.
+   pub primary_key: Vec<ColumnValue>,
    /// Column values before the change (for UPDATE and DELETE).
    /// Values are ordered by column index as defined in the table schema.
    pub old_values: Option<Vec<ColumnValue>>,

crates/sqlx-sqlite-observer/src/change.rs

@@ -6,4 +6,18 @@ pub enum Error {
    /// Failed to register SQLite hooks.
    #[error("Hook registration failed: {0}")]
    HookRegistration(String),
+
+   /// SQLx database error.
+   #[error("Database error: {0}")]
+   Sqlx(#[from] sqlx::Error),
+
+   /// Schema mismatch - table schema changed while observing.
+   #[error(
+      "Schema mismatch for table '{table}': expected {expected} PK columns, but only {actual} values available"
+   )]
+   SchemaMismatch {
+      table: String,
+      expected: usize,
+      actual: usize,
+   },
 }

crates/sqlx-sqlite-observer/src/error.rs

@@ -106,7 +106,7 @@ struct HookContext {
 ///
 /// # Example
 ///
-/// ```rust
+/// ```no_run
 /// use sqlx_sqlite_observer::is_preupdate_hook_enabled;
 ///
 /// if !is_preupdate_hook_enabled() {

crates/sqlx-sqlite-observer/src/hooks.rs

@@ -27,9 +27,10 @@ pub mod broker;
 pub mod change;
 pub mod error;
 pub mod hooks;
+pub mod schema;
 
 pub use broker::ObservationBroker;
-pub use change::{ChangeOperation, ColumnValue, TableChange};
+pub use change::{ChangeOperation, ColumnValue, TableChange, TableInfo};
 pub use error::Error;
 pub use hooks::{SqliteValue, is_preupdate_hook_enabled};