merge db-watch into db-live-query

Author: yujonglee

Cargo.lock

@@ -88,7 +88,6 @@ hypr-db-core2 = { path = "crates/db-core2", package = "db-core2" }
 hypr-db-live-query = { path = "crates/db-live-query", package = "db-live-query" }
 hypr-db-parser = { path = "crates/db-parser", package = "db-parser" }
 hypr-db-user = { path = "crates/db-user", package = "db-user" }
-hypr-db-watch = { path = "crates/db-watch", package = "db-watch" }
 hypr-denoise = { path = "crates/denoise", package = "denoise" }
 hypr-detect = { path = "crates/detect", package = "detect" }
 hypr-device-monitor = { path = "crates/device-monitor", package = "device-monitor" }

Cargo.toml

@@ -48,3 +48,9 @@
 - If the pool policy changes between single-connection and multi-connection, the hook path should remain correct without upper-layer changes.
 - In-memory databases should still be treated carefully; `max_connections(1)` is the safe default unless shared-memory behavior is explicitly intended.
 - If code needs to answer "what tables does this SQL depend on?" or "which subscribers should rerun?", it belongs above this crate.
+
+## Test Ownership
+
+- Put tests here when the behavior is about database opening, connection policy, migration failure handling, cloudsync wiring, or raw table-change hook behavior.
+- Prefer temp-database integration tests here over higher-level plugin tests when verifying pooled connection semantics.
+- Do not test subscription reruns, dependency extraction, transport delivery, or Tauri command behavior here.

crates/db-core2/AGENTS.md

@@ -455,6 +455,76 @@ mod tests {
         assert!(matches!(error, DbOpenError::Migration(message) if message == "nope"));
     }
 
+    #[tokio::test]
+    async fn open_with_migrate_returns_recreate_failed_when_retry_also_fails() {
+        let tmp = tempfile::tempdir().unwrap();
+        let db_path = tmp.path().join("app.db");
+        let attempts = AtomicUsize::new(0);
+
+        let error = Db3::open_with_migrate(
+            DbOpenOptions {
+                storage: DbStorage::Local(&db_path),
+                cloudsync: false,
+                journal_mode_wal: true,
+                foreign_keys: true,
+                max_connections: Some(1),
+                migration_failure_policy: MigrationFailurePolicy::Recreate,
+            },
+            |pool| {
+                let n = attempts.fetch_add(1, Ordering::SeqCst);
+                Box::pin(async move {
+                    let table_name = if n == 0 {
+                        "first_attempt"
+                    } else {
+                        "second_attempt"
+                    };
+                    let sql = format!("CREATE TABLE {table_name} (id TEXT PRIMARY KEY NOT NULL)");
+                    sqlx::query(&sql).execute(pool).await.unwrap();
+                    Err::<(), &'static str>("still broken")
+                })
+            },
+        )
+        .await
+        .unwrap_err();
+
+        assert_eq!(attempts.load(Ordering::SeqCst), 2);
+        assert!(
+            matches!(error, DbOpenError::RecreateFailed(message) if message == "migration failed: still broken")
+        );
+    }
+
+    #[tokio::test]
+    async fn open_with_migrate_applies_requested_pragmas() {
+        let tmp = tempfile::tempdir().unwrap();
+        let db_path = tmp.path().join("app.db");
+
+        let db = Db3::open_with_migrate(
+            DbOpenOptions {
+                storage: DbStorage::Local(&db_path),
+                cloudsync: false,
+                journal_mode_wal: true,
+                foreign_keys: true,
+                max_connections: Some(1),
+                migration_failure_policy: MigrationFailurePolicy::Fail,
+            },
+            |_pool| Box::pin(async { Ok::<(), sqlx::Error>(()) }),
+        )
+        .await
+        .unwrap();
+
+        let foreign_keys: i64 = sqlx::query_scalar("PRAGMA foreign_keys")
+            .fetch_one(db.pool().as_ref())
+            .await
+            .unwrap();
+        let journal_mode: String = sqlx::query_scalar("PRAGMA journal_mode")
+            .fetch_one(db.pool().as_ref())
+            .await
+            .unwrap();
+
+        assert_eq!(foreign_keys, 1);
+        assert_eq!(journal_mode.to_lowercase(), "wal");
+    }
+
     #[tokio::test]
     async fn emits_table_changes_for_local_writes() {
         let db = Db3::connect_memory_plain().await.unwrap();
@@ -479,6 +549,44 @@ mod tests {
         assert_eq!(change.kind, TableChangeKind::Insert);
     }
 
+    #[tokio::test]
+    async fn emits_update_and_delete_table_changes() {
+        let db = Db3::connect_memory_plain().await.unwrap();
+        sqlx::query("CREATE TABLE test_events (id TEXT PRIMARY KEY NOT NULL, value TEXT NOT NULL)")
+            .execute(db.pool().as_ref())
+            .await
+            .unwrap();
+        sqlx::query("INSERT INTO test_events (id, value) VALUES ('a', 'before')")
+            .execute(db.pool().as_ref())
+            .await
+            .unwrap();
+
+        let mut changes = db.subscribe_table_changes();
+
+        sqlx::query("UPDATE test_events SET value = 'after' WHERE id = 'a'")
+            .execute(db.pool().as_ref())
+            .await
+            .unwrap();
+        sqlx::query("DELETE FROM test_events WHERE id = 'a'")
+            .execute(db.pool().as_ref())
+            .await
+            .unwrap();
+
+        let update = tokio::time::timeout(std::time::Duration::from_secs(1), changes.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let delete = tokio::time::timeout(std::time::Duration::from_secs(1), changes.recv())
+            .await
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(update.table, "test_events");
+        assert_eq!(update.kind, TableChangeKind::Update);
+        assert_eq!(delete.table, "test_events");
+        assert_eq!(delete.kind, TableChangeKind::Delete);
+    }
+
     #[tokio::test]
     async fn emits_table_changes_across_multiple_connections() {
         let dir = tempfile::tempdir().unwrap();

crates/db-core2/src/lib.rs

@@ -4,7 +4,7 @@
 
 - `db-live-query` is the reusable live-query service layer.
 - It executes SQL, analyzes query dependencies, tracks subscriptions, reruns affected queries, and serializes rows for sinks.
-- It consumes raw table-change signals from `db-core2` and pure watch indexing from `db-watch`.
+- It consumes raw table-change signals from `db-core2` and maintains its own watch indexing (`watch.rs`).
 
 ## This Crate Owns
 
@@ -36,7 +36,7 @@
 
 ## Dependency Direction
 
-- This crate may depend on `db-core2` and `db-watch`.
+- This crate may depend on `db-core2`.
 - `plugins/db` may depend on this crate.
 - This crate must not depend on Tauri or app-specific UI/runtime layers.
 
@@ -46,3 +46,9 @@
 - Raw SQLite hook installation and pooled connection setup belong below this crate in `db-core2`.
 - Tauri channels, app bootstrap, and JS-facing event types belong above this crate in `plugins/db`.
 - If this layer ever becomes app-specific, it should be split again rather than letting transport or product logic leak inward.
+
+## Test Ownership
+
+- Put tests here when the behavior is about dependency extraction, reactive vs non-reactive classification, rerun targeting, unsubscribe semantics, stale-sink cleanup, or JSON row serialization.
+- These tests may use a real temp database plus a fake sink, but they should not depend on Tauri transport types.
+- Higher layers should not duplicate this crate's rerun and invalidation tests unless they are specifically proving adapter integration.

crates/db-live-query/AGENTS.md

@@ -5,7 +5,6 @@ edition = "2024"
 
 [dependencies]
 hypr-db-core2 = { workspace = true }
-hypr-db-watch = { workspace = true }
 serde_json = { workspace = true }
 sqlx = { workspace = true, features = ["runtime-tokio", "sqlite", "sqlite-unbundled"] }
 thiserror = { workspace = true }

crates/db-live-query/Cargo.toml

@@ -0,0 +1,11 @@
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    #[error(transparent)]
+    Sqlx(#[from] sqlx::Error),
+    #[error("subscription not found: {0}")]
+    SubscriptionNotFound(String),
+    #[error("failed to send query event: {0}")]
+    Sink(String),
+}
+
+pub type Result<T> = std::result::Result<T, Error>;

crates/db-live-query/src/error.rs

@@ -216,6 +216,18 @@ mod tests {
         assert_eq!(tables, HashSet::from(["daily_notes".to_string()]));
     }
 
+    #[tokio::test]
+    async fn bracket_quoted_alias_query() {
+        let db = test_db().await;
+        let tables = extract_tables(
+            db.pool(),
+            "SELECT [dn].id FROM [daily_notes] AS [dn] WHERE [dn].date = '2026-04-11'",
+        )
+        .await
+        .unwrap();
+        assert_eq!(tables, HashSet::from(["daily_notes".to_string()]));
+    }
+
     #[tokio::test]
     async fn schema_qualified_query() {
         let db = test_db().await;