Merge pull request #37 from pmorris-dev/add-typescript-api-for-observer

feat: add observer TS API and commands (#412486)

Author: jjhafer

.gitignore

@@ -1,5 +1,5 @@
 # Rust
-/target/
+**/target/
 
 # Node
 node_modules/
@@ -18,4 +18,7 @@ dist-js/
 Thumbs.db
 
 # Tauri
-/gen/
+**/gen/
+
+# Examples
+examples/**/Cargo.lock

Cargo.lock

@@ -5,11 +5,12 @@ members = [
    "crates/sqlx-sqlite-observer",
    "crates/sqlx-sqlite-toolkit",
 ]
+exclude = ["examples/observer-demo/src-tauri"]
 
 [package]
 name = "tauri-plugin-sqlite"
-version = "0.1.0"
-description = "A Tauri plugin for SQLite database access with connection management"
+version = "0.2.0"
+description = "A Tauri plugin for SQLite with connection pooling, builder-pattern queries, transactions, and reactive change notifications"
 license = "MIT"
 edition = "2024"
 rust-version = "1.89"
@@ -20,7 +21,6 @@ tauri = "2.9.3"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.145"
 thiserror = "2.0.17"
-log = "0.4.28"
 time = "0.3.44"
 tokio = { version = "1.48.0", features = ["rt", "sync", "time"] }
 indexmap = { version = "2.12.1", features = ["serde"] }
@@ -34,8 +34,14 @@ sqlx = { version = "0.8.6", features = ["sqlite", "json", "time", "runtime-tokio
 # Connection manager
 sqlx-sqlite-conn-mgr = { path = "crates/sqlx-sqlite-conn-mgr" }
 
-# High-level toolkit (builders, transactions, decoding)
-sqlx-sqlite-toolkit = { path = "crates/sqlx-sqlite-toolkit" }
+# Toolkit (high-level API — builders, transactions, decoding)
+sqlx-sqlite-toolkit = { path = "crates/sqlx-sqlite-toolkit", features = ["observer"] }
+
+# Observer types (for payload conversion)
+sqlx-sqlite-observer = { path = "crates/sqlx-sqlite-observer", features = ["conn-mgr"] }
+
+# Async stream support for observer subscriptions
+futures = "0.3.31"
 
 [build-dependencies]
 tauri-plugin = { version = "2.5.1", features = ["build"] }

Cargo.toml

@@ -397,6 +397,66 @@ await db.executeTransaction([
    * Attachments are connection-scoped and don't persist across queries
    * Main database is always accessible without a schema prefix
 
+### Change Notifications
+
+Subscribe to real-time change notifications when rows are inserted, updated, or
+deleted. Changes are only published after transactions commit — you never see
+partial or rolled-back data.
+
+```typescript
+// 1. Enable observation for specific tables
+await db.observe(['users', 'posts']);
+
+// 2. Subscribe to changes
+const subscription = await db.subscribe(['users'], (event) => {
+   if (event.event === 'change') {
+      const { table, operation, primaryKey, newValues, oldValues } = event.data;
+
+      console.info(`${operation} on ${table}, row key:`, primaryKey);
+
+      if (operation === 'insert' || operation === 'update') {
+         console.info('New values:', newValues);
+      }
+      if (operation === 'update' || operation === 'delete') {
+         console.info('Old values:', oldValues);
+      }
+   } else if (event.event === 'lagged') {
+      // Consumer fell behind — some notifications were missed
+      console.warn(`Missed ${event.data.count} notifications`);
+   }
+});
+
+// 3. Changes are now streamed to the callback
+await db.execute('INSERT INTO users (name) VALUES ($1)', ['Alice']);
+// callback fires: { event: 'change', data: { table: 'users', operation: 'insert', ... } }
+
+// 4. Unsubscribe when done
+await subscription.unsubscribe();
+
+// 5. Disable observation entirely (also aborts all active subscriptions)
+await db.unobserve();
+```
+
+**Configuration:**
+
+```typescript
+await db.observe(['users'], {
+   channelCapacity: 512,  // default: 256 — at least the number of writes in your largest transaction
+   captureValues: false,  // default: true — disable to reduce memory per notification
+});
+```
+
+**Important:**
+
+   * Call `observe()` before `subscribe()` — subscribing without observation returns
+     an error
+   * Multiple subscriptions can be active on the same database, each filtering by
+     different tables
+   * `lagged` events indicate the broadcast channel filled up before the
+     subscriber could read — increase `channelCapacity`
+   * Column values (`oldValues`, `newValues`) are typed as `ColumnValue` — a tagged
+     union of `null`, `integer`, `real`, `text`, or `blob` (base64-encoded)
+
 ### Error Handling
 
 ```typescript
@@ -419,6 +479,8 @@ Common error codes:
    * `IO_ERROR` - File system error
    * `MIGRATION_ERROR` - Migration failed
    * `MULTIPLE_ROWS_RETURNED` - `fetchOne()` returned multiple rows
+   * `OBSERVATION_NOT_ENABLED` - Called `subscribe()` before `observe()`
+   * `OBSERVER_ERROR` - Error from the observer subsystem
 
 ### Closing and Removing
 
@@ -449,6 +511,9 @@ await db.remove();           // Close and DELETE database file(s) - irreversible
 | `fetchOne<T>(query, values?)` | Execute SELECT, return single row or `undefined` |
 | `close()` | Close connection, returns `true` if was loaded |
 | `remove()` | Close and delete database file(s), returns `true` if was loaded |
+| `observe(tables, config?)` | Enable change observation for tables |
+| `subscribe(tables, onEvent)` | Subscribe to change notifications, returns `Subscription` |
+| `unobserve()` | Disable observation and abort all subscriptions |
 
 ### Builder Methods
 
@@ -469,6 +534,12 @@ return builders that are directly awaitable and support method chaining:
 | `commit()` | Commit transaction and release write lock |
 | `rollback()` | Rollback transaction and release write lock |
 
+### Subscription Methods
+
+| Method | Description |
+| ------ | ----------- |
+| `unsubscribe()` | Stop receiving change notifications, returns `true` if was active |
+
 ### Types
 
 ```typescript
@@ -492,6 +563,33 @@ interface SqliteError {
    code: string;
    message: string;
 }
+
+interface ObserverConfig {
+   channelCapacity?: number;  // default: 256
+   captureValues?: boolean;   // default: true
+}
+
+type ChangeOperation = 'insert' | 'update' | 'delete';
+
+type ColumnValue =
+   | { type: 'null' }
+   | { type: 'integer'; value: number }
+   | { type: 'real'; value: number }
+   | { type: 'text'; value: string }
+   | { type: 'blob'; value: string };  // base64-encoded
+
+interface TableChange {
+   table: string;
+   operation?: ChangeOperation;
+   rowid?: number;
+   primaryKey: ColumnValue[];
+   oldValues?: ColumnValue[];   // update, delete
+   newValues?: ColumnValue[];   // insert, update
+}
+
+type TableChangeEvent =
+   | { event: 'change'; data: TableChange }
+   | { event: 'lagged'; data: { count: number } };
 ```
 
 ## Rust-Only API

README.md

@@ -12,6 +12,10 @@ fn main() {
       "close_all",
       "remove",
       "get_migration_events",
+      "observe",
+      "subscribe",
+      "unsubscribe",
+      "unobserve",
    ])
    .build();
 }

api-iife.js

@@ -9,7 +9,7 @@
 //! Use [`is_preupdate_hook_enabled()`] to check at runtime whether the linked
 //! SQLite library supports this feature.
 
-use std::ffi::{CStr, CString, c_int, c_void};
+use std::ffi::{CStr, CString, c_char, c_int, c_void};
 use std::panic::catch_unwind;
 use std::ptr;
 use std::sync::Arc;
@@ -61,7 +61,7 @@ impl SqliteValue {
                SqliteValue::Null
             } else {
                // SAFETY: SQLite guarantees text is valid UTF-8 with a null terminator
-               let cstr = unsafe { CStr::from_ptr(text_ptr as *const i8) };
+               let cstr = unsafe { CStr::from_ptr(text_ptr as *const c_char) };
                SqliteValue::Text(cstr.to_string_lossy().into_owned())
             }
          }
@@ -214,8 +214,8 @@ unsafe extern "C" fn preupdate_callback(
    user_data: *mut c_void,
    db: *mut sqlite3,
    op: c_int,
-   _database: *const i8,
-   table: *const i8,
+   _database: *const c_char,
+   table: *const c_char,
    old_rowid: i64,
    new_rowid: i64,
 ) {

build.rs

@@ -5,6 +5,12 @@ version = "0.8.6"
 license = "MIT"
 edition = "2024"
 rust-version = "1.89"
+authors = ["Jeremy Thomerson"]
+description = "High-level SQLite API built on sqlx with builder-pattern queries, transactions, and JSON decoding"
+repository = "https://github.com/silvermine/tauri-plugin-sqlite"
+readme = "README.md"
+keywords = ["sqlite", "sqlx", "database", "transactions", "async"]
+categories = ["database", "asynchronous"]
 
 [features]
 default = []

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

@@ -292,15 +292,25 @@ impl DatabaseWrapper {
    /// Close the database connection.
    ///
    /// Checkpoints the WAL and closes all connection pools.
-   pub async fn close(self) -> Result<(), Error> {
+   /// If observation is enabled, it is disabled first to unregister SQLite hooks
+   /// and allow the write connection to close cleanly.
+   pub async fn close(mut self) -> Result<(), Error> {
+      #[cfg(feature = "observer")]
+      self.disable_observation();
+
       self.inner.close().await?;
       Ok(())
    }
 
    /// Close the database connection and remove all database files.
    ///
    /// Removes the main database file, WAL, and SHM files.
-   pub async fn remove(self) -> Result<(), Error> {
+   /// If observation is enabled, it is disabled first to unregister SQLite hooks
+   /// and allow the write connection to close cleanly.
+   pub async fn remove(mut self) -> Result<(), Error> {
+      #[cfg(feature = "observer")]
+      self.disable_observation();
+
       self.inner.remove().await?;
       Ok(())
    }
@@ -310,9 +320,14 @@ impl DatabaseWrapper {
    /// After calling this, write operations will be tracked and subscribers
    /// can receive change notifications.
    ///
+   /// If observation is already enabled, the previous observer is disabled first.
+   /// This drops the old broadcast broker, causing existing subscriber streams to
+   /// terminate. Callers must re-subscribe after re-enabling observation.
+   ///
    /// Requires the `observer` feature.
    #[cfg(feature = "observer")]
    pub fn enable_observation(&mut self, config: ObserverConfig) {
+      self.disable_observation();
       self.observer = Some(ObservableSqliteDatabase::new(
          Arc::clone(&self.inner),
          config,

crates/sqlx-sqlite-toolkit/Cargo.toml

@@ -17,6 +17,7 @@ module.exports = [
          '**/*.d.ts',
          '**/commitlint.config.cjs',
          'vitest.config.ts',
+         'examples/**',
       ],
    },
    ...config,