diff --git a/database/Cargo.lock b/database/Cargo.lock index 508d36d31..7a5338091 100644 --- a/database/Cargo.lock +++ b/database/Cargo.lock @@ -593,10 +593,12 @@ dependencies = [ "chrono", "clap", "deadpool-postgres", + "futures-util", "iii-console-ui", "iii-helpers", "iii-sdk", "mysql_async", + "notify", "postgres-types", "r2d2", "r2d2_sqlite", @@ -814,6 +816,15 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "fsevent-sys" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2" +dependencies = [ + "libc", +] + [[package]] name = "funty" version = "2.0.0" @@ -1323,6 +1334,26 @@ dependencies = [ "serde_core", ] +[[package]] +name = "inotify" +version = "0.11.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "153be1941a183ec9ccd095ddbe17a8b8d435ef6c76e9e02451b933c3999af2c8" +dependencies = [ + "bitflags", + "inotify-sys", + "libc", +] + +[[package]] +name = "inotify-sys" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c033f80b2c113cdf91ab7a33faa9cbc014726dcad99880c8609af2a370edf37d" +dependencies = [ + "libc", +] + [[package]] name = "ipnet" version = "2.12.0" @@ -1391,6 +1422,26 @@ dependencies = [ "indexmap", ] +[[package]] +name = "kqueue" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "273c0752728918e0ac4976f2b275b6fefb9ecd400585dec929419f3844cd87b5" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087" +dependencies = [ + "bitflags", + "libc", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -1539,6 +1590,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" dependencies = [ "libc", + "log", "wasi 0.11.1+wasi-snapshot-preview1", "windows-sys 0.61.2", ] @@ -1586,6 +1638,7 @@ dependencies = [ "base64 0.21.7", "bindgen", "bitflags", + "bitvec", "btoi", "byteorder", "bytes", @@ -1620,6 +1673,33 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "notify" +version = "8.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3" +dependencies = [ + "bitflags", + "fsevent-sys", + "inotify", + "kqueue", + "libc", + "log", + "mio", + "notify-types", + "walkdir", + "windows-sys 0.60.2", +] + +[[package]] +name = "notify-types" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a" +dependencies = [ + "bitflags", +] + [[package]] name = "ntapi" version = "0.4.3" @@ -2386,6 +2466,15 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + [[package]] name = "saturating" version = "0.1.0" @@ -3286,6 +3375,16 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + [[package]] name = "want" version = "0.3.1" @@ -3504,6 +3603,15 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "winapi-x86_64-pc-windows-gnu" version = "0.4.0" diff --git a/database/Cargo.toml b/database/Cargo.toml index 48bdeb1df..f2c7084f4 100644 --- a/database/Cargo.toml +++ b/database/Cargo.toml @@ -47,7 +47,7 @@ tokio-postgres-rustls = "0.13" rust_decimal = { version = "1", features = ["db-tokio-postgres"] } # MySQL — `rustls-tls` adds rustls-backed TLS without pulling in OpenSSL. -mysql_async = { version = "0.34", default-features = false, features = ["minimal-rust", "rustls-tls"] } +mysql_async = { version = "0.34", default-features = false, features = ["minimal-rust", "rustls-tls", "binlog"] } # TLS (shared between postgres and mysql) rustls = { version = "0.23", default-features = false, features = ["aws_lc_rs", "tls12", "logging"] } @@ -60,6 +60,11 @@ rustls-webpki = { version = "0.103", default-features = false, features = ["std" rusqlite = { version = "0.31", features = ["bundled", "chrono", "column_decltype"] } r2d2 = "0.8" r2d2_sqlite = "0.24" +# Filesystem watch for sqlite native capture (inotify wake-up; the changelog +# table is the source of truth, fs events only decide when to look). +notify = "8" +# StreamExt for the mysql binlog stream (already in the tree via mysql_async). +futures-util = { version = "0.3", default-features = false, features = ["std", "async-await"] } [dev-dependencies] tempfile = "3" diff --git a/database/config.yaml.example b/database/config.yaml.example index 318344420..68d3d6416 100644 --- a/database/config.yaml.example +++ b/database/config.yaml.example @@ -16,3 +16,20 @@ databases: max: 10 idle_timeout_ms: 30000 acquire_timeout_ms: 5000 + # Native change capture: `database::row-changed` fires for committed writes + # from ANY client (psql, other processes), not just this worker. Bindings + # must name a table. Default (`capture: statements`) only reports writes + # made through this worker. + # - postgres: triggers + LISTEN/NOTIFY (role needs DDL on watched tables) + # - sqlite (file-backed only): triggers + changelog, fs-watch wake-up + # - mysql: binlog replication stream; nothing installed, but the user + # needs: GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* + # analytics: + # url: postgres://user:pass@localhost:5432/analytics + # capture: native + # events: + # url: sqlite:./data/events.db + # capture: native + # orders: + # url: mysql://user:pass@localhost:3306/orders + # capture: native diff --git a/database/src/config.rs b/database/src/config.rs index 951f98383..357e8dd11 100644 --- a/database/src/config.rs +++ b/database/src/config.rs @@ -67,6 +67,14 @@ pub struct DatabaseConfig { pub pool: PoolConfig, #[serde(default)] pub tls: TlsConfig, + /// How `database::row-changed` events are captured for this database. + /// `statements` (default) classifies the SQL this worker executes; + /// `native` makes writes from ANY client — psql, other processes — fire + /// too. Postgres: triggers + LISTEN/NOTIFY. File-backed sqlite: a + /// trigger-fed changelog drained on filesystem wake-up. MySQL: the + /// binlog replication stream (needs REPLICATION SLAVE + CLIENT). + #[serde(default, skip_serializing_if = "CaptureMode::is_statements")] + pub capture: CaptureMode, /// Populated by [`WorkerConfig::finalize`] from the URL scheme. /// Do not construct `DatabaseConfig` directly without calling /// `finalize` — the default `Sqlite` value will silently mismatch @@ -149,6 +157,28 @@ pub enum DriverKind { Sqlite, } +/// How row-change events are captured for one database. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize, Serialize, JsonSchema)] +#[serde(rename_all = "lowercase")] +pub enum CaptureMode { + /// Classify the SQL this worker executes. Writes from other clients are + /// invisible. Works on every driver. The default. + #[default] + Statements, + /// Capture writes from any client, including other processes. + /// Table-scoped bindings only. Postgres: triggers + LISTEN/NOTIFY on a + /// dedicated connection (needs DDL rights). File-backed sqlite: + /// triggers + changelog table + filesystem watch. MySQL: the binlog + /// replication stream (needs replication grants, nothing installed). + Native, +} + +impl CaptureMode { + pub fn is_statements(&self) -> bool { + *self == CaptureMode::Statements + } +} + #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema)] pub struct PoolConfig { /// Maximum number of open connections in the pool. @@ -197,6 +227,7 @@ impl WorkerConfig { url: DEFAULT_SQLITE_URL.to_string(), pool: PoolConfig::default(), tls: TlsConfig::default(), + capture: CaptureMode::default(), driver: DriverKind::default(), }, )]), @@ -267,6 +298,42 @@ impl WorkerConfig { redact_url(&db.url) ) })?; + if db.capture == CaptureMode::Native { + match db.driver { + // Postgres captures via LISTEN/NOTIFY; server-side + // prerequisites are checked at binding registration, + // where failures are actionable. + DriverKind::Postgres => {} + DriverKind::Mysql => { + // Binlog events are filtered to the url's schema; a + // url without one would capture every database on + // the server — table names and change volumes from + // schemas this handle was never meant to see. + let has_schema = url::Url::parse(&db.url) + .map(|u| !u.path().trim_start_matches('/').is_empty()) + .unwrap_or(false); + if !has_schema { + return Err(format!( + "db `{name}`: `capture: native` on mysql requires the \ + url to name a database (mysql://host/dbname) — binlog \ + events are filtered to that schema" + )); + } + } + DriverKind::Sqlite => { + // A `:memory:` database exists per connection — a + // watcher connection would open a different database + // and hear nothing, ever. + if db.url.contains(":memory:") { + return Err(format!( + "db `{name}`: `capture: native` requires a file-backed \ + sqlite database; `:memory:` is per-connection and \ + cannot be observed" + )); + } + } + } + } } Ok(cfg) } @@ -326,7 +393,7 @@ pub fn validate_sql_identifier(s: &str) -> Result<(), String> { Ok(()) } -fn detect_driver(url: &str) -> Option { +pub(crate) fn detect_driver(url: &str) -> Option { let lower = url.to_ascii_lowercase(); if lower.starts_with("postgres://") || lower.starts_with("postgresql://") { Some(DriverKind::Postgres) @@ -408,6 +475,45 @@ mod tests { assert!(matches!(back.databases["p"].driver, DriverKind::Sqlite)); } + #[test] + fn capture_native_allows_all_drivers_except_memory_sqlite() { + for url in [ + "postgres://u@h/db", + "sqlite:./data/iii.db", + "mysql://u@h/db", + ] { + let c = cfg(&format!( + "databases:\n p:\n url: {url}\n capture: native\n" + )); + assert_eq!(c.databases["p"].capture, CaptureMode::Native, "{url}"); + } + + // A per-connection `:memory:` database cannot be observed. + let err = WorkerConfig::from_yaml( + "databases:\n p:\n url: \"sqlite::memory:\"\n capture: native\n", + ) + .unwrap_err(); + assert!(err.contains("file-backed"), "got: {err}"); + + // A schema-less mysql url would capture every database on the server. + let err = WorkerConfig::from_yaml( + "databases:\n p:\n url: mysql://u@h\n capture: native\n", + ) + .unwrap_err(); + assert!(err.contains("name a database"), "got: {err}"); + let err = WorkerConfig::from_yaml( + "databases:\n p:\n url: mysql://u@h/\n capture: native\n", + ) + .unwrap_err(); + assert!(err.contains("name a database"), "got: {err}"); + + // Default stays statements and stays out of the serialized form — + // existing configs round-trip byte-identical. + let d = cfg("databases:\n p:\n url: postgres://u@h/db\n"); + assert_eq!(d.databases["p"].capture, CaptureMode::Statements); + assert!(d.to_json()["databases"]["p"].get("capture").is_none()); + } + #[test] fn json_schema_is_object_with_databases_property() { let schema = WorkerConfig::json_schema(); diff --git a/database/src/configuration.rs b/database/src/configuration.rs index 6a94422fd..2f49d8870 100644 --- a/database/src/configuration.rs +++ b/database/src/configuration.rs @@ -81,8 +81,16 @@ pub async fn build_pools(cfg: &WorkerConfig) -> Result, St Ok(pools) } -pub async fn apply_config(state: &AppState, cfg: WorkerConfig) -> Result<(), String> { +pub async fn apply_config( + state: &AppState, + cfg: WorkerConfig, + listeners: Option<&crate::triggers::NativeListeners>, +) -> Result<(), String> { let new_pools = build_pools(&cfg).await?; + // A failed pool build leaves everything untouched, listeners included. + if let Some(listeners) = listeners { + listeners.sync(&cfg); + } // Swap pools and the config snapshot inside one critical section (pools // lock first, then config) so a concurrent reader never observes new // pools paired with the old config or vice-versa. A failed build above @@ -113,7 +121,11 @@ pub struct OnConfigChangeResponse { } /// Register the internal config-change handler and bind a `configuration` trigger. -pub fn register_config_trigger(iii: &IIIClient, state: AppState) -> Result<(), Error> { +pub fn register_config_trigger( + iii: &IIIClient, + state: AppState, + listeners: Option>, +) -> Result<(), Error> { let st = state.clone(); let engine = iii.clone(); iii.register_function( @@ -121,8 +133,9 @@ pub fn register_config_trigger(iii: &IIIClient, state: AppState) -> Result<(), E RegisterFunction::new_async(move |_event: OnConfigChangeEvent| { let st = st.clone(); let engine = engine.clone(); + let listeners = listeners.clone(); async move { - on_config_change(&engine, &st).await; + on_config_change(&engine, &st, listeners.as_deref()).await; Ok::(OnConfigChangeResponse { ok: true }) } }) @@ -150,7 +163,11 @@ pub fn register_config_trigger(iii: &IIIClient, state: AppState) -> Result<(), E /// `payload.new_value` would let any caller replace the live connection pools /// (e.g. point them at an attacker-controlled database) without updating /// persisted state. Re-fetch the stored value via `configuration::get` instead. -async fn on_config_change(iii: &IIIClient, state: &AppState) { +async fn on_config_change( + iii: &IIIClient, + state: &AppState, + listeners: Option<&crate::triggers::NativeListeners>, +) { let cfg = match fetch_config(iii).await { Ok(cfg) => cfg, Err(e) => { @@ -161,7 +178,7 @@ async fn on_config_change(iii: &IIIClient, state: &AppState) { return; } }; - match apply_config(state, cfg).await { + match apply_config(state, cfg, listeners).await { Ok(()) => tracing::info!("database pools reloaded after configuration change"), Err(e) => tracing::error!( error = %e, diff --git a/database/src/handlers/mod.rs b/database/src/handlers/mod.rs index 80e7ef470..783fad377 100644 --- a/database/src/handlers/mod.rs +++ b/database/src/handlers/mod.rs @@ -38,6 +38,7 @@ pub mod prepare; pub mod query; pub mod rollback_transaction; pub mod run_statement; +pub mod test_connection; pub mod transaction; pub mod transaction_execute; pub mod transaction_query; @@ -84,6 +85,18 @@ pub struct AppState { } impl AppState { + /// Whether `db` announces its writes via SQL classification. A + /// `capture: native` database hears its own commits through NOTIFY like + /// every other client — classifying here too would fire everything twice. + async fn classifies_own_writes(&self, db: &str) -> bool { + self.config + .read() + .await + .databases + .get(db) + .is_none_or(|d| d.capture.is_statements()) + } + /// Announce a committed change, if anything is listening. Every mutating /// handler ends with this; the bus decides whether the statement changed /// rows and who cares. @@ -95,12 +108,14 @@ impl AppState { returning: Option<&[serde_json::Map]>, ) { if let Some(bus) = &self.row_changes { - bus.emit(db, sql, affected_rows, returning).await; + if self.classifies_own_writes(db).await { + bus.emit(db, sql, affected_rows, returning).await; + } } } /// Buffer a change made inside an interactive transaction until its commit. - pub fn stage_row_change( + pub async fn stage_row_change( &self, transaction_id: &str, db: &str, @@ -109,7 +124,9 @@ impl AppState { returning: Option<&[serde_json::Map]>, ) { if let Some(bus) = &self.row_changes { - bus.stage(transaction_id, db, sql, affected_rows, returning); + if self.classifies_own_writes(db).await { + bus.stage(transaction_id, db, sql, affected_rows, returning); + } } } diff --git a/database/src/handlers/rollback_transaction.rs b/database/src/handlers/rollback_transaction.rs index 7e0a230ea..026fcb27e 100644 --- a/database/src/handlers/rollback_transaction.rs +++ b/database/src/handlers/rollback_transaction.rs @@ -170,7 +170,8 @@ mod tests { }); wait_until_taken(&st).await; - st.stage_row_change(&id, "primary", "INSERT INTO t VALUES (1)", 1, None); + st.stage_row_change(&id, "primary", "INSERT INTO t VALUES (1)", 1, None) + .await; assert_eq!(bus.pending_count(&id), 1); drop(lock); @@ -188,7 +189,8 @@ mod tests { .await .unwrap(); let id = begin.transaction.id; - st.stage_row_change(&id, "primary", "INSERT INTO t VALUES (1)", 1, None); + st.stage_row_change(&id, "primary", "INSERT INTO t VALUES (1)", 1, None) + .await; let lock = st.transactions.lock(&id).await.unwrap(); let task_state = st.clone(); diff --git a/database/src/handlers/test_connection.rs b/database/src/handlers/test_connection.rs new file mode 100644 index 000000000..407e71db9 --- /dev/null +++ b/database/src/handlers/test_connection.rs @@ -0,0 +1,299 @@ +//! `database::testConnection` — probe a CANDIDATE database config. +//! +//! Exists for the console's configuration form: the operator edits a url, +//! presses "test connection", and learns whether it works *before* saving. +//! The probe therefore takes the url/tls straight from the request instead +//! of a configured handle, opens one throwaway connection outside every +//! pool, runs the smallest possible query, and reports the outcome as data +//! (`ok: false` is a normal response, not a handler error). +//! +//! Error texts are returned verbatim except that the url's credentials are +//! scrubbed. That is safe here where it would not be elsewhere: everything +//! echoed back originates from the caller's own request, so nothing +//! cross-tenant can leak — but a driver error that quotes the url must not +//! turn a stored password into log/UI text. + +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use std::time::{Duration, Instant}; + +use crate::config::{detect_driver, DriverKind, TlsConfig}; +use crate::pool::tls::make_pg_connector; +use crate::transaction::driver_system; + +const DEFAULT_TIMEOUT_MS: u64 = 5_000; +const MAX_TIMEOUT_MS: u64 = 30_000; + +#[derive(Debug, Deserialize, JsonSchema)] +pub struct TestConnectionReq { + /// Connection url to probe (`postgres://…`, `mysql://…`, `sqlite:…`). + pub url: String, + /// TLS settings to probe with. Defaults like a configured database + /// (mode `require`) when omitted. + #[serde(default)] + pub tls: Option, + /// Overall budget for the attempt. Default 5000, capped at 30000. + #[serde(default)] + pub timeout_ms: Option, +} + +#[derive(Debug, Serialize, JsonSchema)] +pub struct TestConnectionResp { + /// Whether a connection was established and answered a query. + pub ok: bool, + /// "postgres" | "mysql" | "sqlite" | "unknown". + pub driver: String, + /// Wall time of the whole attempt. + pub latency_ms: u64, + /// Server version string, when the probe got far enough to ask. + #[serde(skip_serializing_if = "Option::is_none")] + pub server_version: Option, + /// Why the probe failed (credentials scrubbed). Absent on success. + #[serde(skip_serializing_if = "Option::is_none")] + pub message: Option, +} + +pub async fn handle(req: TestConnectionReq) -> Result { + let started = Instant::now(); + let Some(driver) = detect_driver(&req.url) else { + return Ok(TestConnectionResp { + ok: false, + driver: "unknown".into(), + latency_ms: 0, + server_version: None, + message: Some( + "unknown url scheme — expected sqlite:, postgres:// / postgresql://, or mysql://" + .into(), + ), + }); + }; + let tls = req.tls.clone().unwrap_or_default(); + let budget = Duration::from_millis( + req.timeout_ms + .unwrap_or(DEFAULT_TIMEOUT_MS) + .min(MAX_TIMEOUT_MS), + ); + + let outcome = tokio::time::timeout(budget, probe(driver, &req.url, &tls)).await; + let latency_ms = started.elapsed().as_millis() as u64; + let (ok, server_version, message) = match outcome { + Ok(Ok(version)) => (true, Some(version), None), + Ok(Err(e)) => (false, None, Some(scrub_credentials(&e, &req.url))), + Err(_) => ( + false, + None, + Some(format!("timed out after {}ms", budget.as_millis())), + ), + }; + Ok(TestConnectionResp { + ok, + driver: driver_system(driver).to_string(), + latency_ms, + server_version, + message, + }) +} + +async fn probe(driver: DriverKind, url: &str, tls: &TlsConfig) -> Result { + match driver { + DriverKind::Postgres => probe_postgres(url, tls).await, + DriverKind::Mysql => probe_mysql(url, tls).await, + DriverKind::Sqlite => probe_sqlite(url).await, + } +} + +async fn probe_postgres(url: &str, tls: &TlsConfig) -> Result { + async fn ping(client: tokio_postgres::Client) -> Result { + let row = client + .query_one("SELECT version()", &[]) + .await + .map_err(|e| e.to_string())?; + Ok(row.get::<_, String>(0)) + } + match make_pg_connector(tls).map_err(|e| format!("{e:?}"))? { + Some(connector) => { + let (client, conn) = tokio_postgres::connect(url, connector) + .await + .map_err(|e| e.to_string())?; + tokio::spawn(async move { + let _ = conn.await; + }); + ping(client).await + } + None => { + let (client, conn) = tokio_postgres::connect(url, tokio_postgres::NoTls) + .await + .map_err(|e| e.to_string())?; + tokio::spawn(async move { + let _ = conn.await; + }); + ping(client).await + } + } +} + +async fn probe_mysql(url: &str, tls: &TlsConfig) -> Result { + use mysql_async::prelude::Queryable as _; + let opts = crate::triggers::mysql_binlog::build_opts(url, tls)?; + let mut conn = mysql_async::Conn::new(opts) + .await + .map_err(|e| e.to_string())?; + let version: Option = conn + .query_first("SELECT VERSION()") + .await + .map_err(|e| e.to_string())?; + let _ = conn.disconnect().await; + version.ok_or_else(|| "server returned no version row".into()) +} + +async fn probe_sqlite(url: &str) -> Result { + let url = url.to_string(); + tokio::task::spawn_blocking(move || { + let version = |conn: &rusqlite::Connection| -> Result { + conn.query_row("SELECT sqlite_version()", [], |r| r.get::<_, String>(0)) + .map(|v| format!("SQLite {v}")) + .map_err(|e| e.to_string()) + }; + if url.contains(":memory:") { + let conn = rusqlite::Connection::open_in_memory().map_err(|e| e.to_string())?; + return version(&conn); + } + let Some(path) = crate::triggers::sqlite_watch::sqlite_file_path(&url) else { + return Err("unreadable sqlite url".into()); + }; + // Open WITHOUT the create flag: a probe must not leave a database + // file behind. A missing file is still useful news — the pool + // creates it (and any parent dirs) when the configuration is saved. + if !path.exists() { + return Err(format!( + "file {} does not exist yet — it is created automatically when this \ + configuration is saved", + path.display() + )); + } + let conn = rusqlite::Connection::open_with_flags( + &path, + rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE, + ) + .map_err(|e| e.to_string())?; + version(&conn) + }) + .await + .map_err(|e| format!("probe task failed: {e}"))? +} + +/// Strip the url's username/password out of an error text — drivers quote +/// the connection string in some failure modes. +fn scrub_credentials(message: &str, url: &str) -> String { + let mut out = message.to_string(); + if let Ok(parsed) = url::Url::parse(url) { + if let Some(password) = parsed.password() { + if !password.is_empty() { + out = out.replace(password, "***"); + } + } + if !parsed.username().is_empty() { + out = out.replace(&format!("{}:", parsed.username()), "***:"); + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + fn req(url: &str) -> TestConnectionReq { + TestConnectionReq { + url: url.into(), + tls: None, + timeout_ms: None, + } + } + + #[tokio::test(flavor = "multi_thread")] + async fn sqlite_memory_probes_ok_with_a_version() { + let resp = handle(req("sqlite::memory:")).await.unwrap(); + assert!(resp.ok, "{:?}", resp.message); + assert_eq!(resp.driver, "sqlite"); + assert!(resp.server_version.unwrap().starts_with("SQLite ")); + } + + #[tokio::test(flavor = "multi_thread")] + async fn sqlite_existing_file_probes_ok_and_missing_file_explains() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("probe.db"); + rusqlite::Connection::open(&path).unwrap(); + + let resp = handle(req(&format!("sqlite:{}", path.display()))) + .await + .unwrap(); + assert!(resp.ok, "{:?}", resp.message); + + let missing = dir.path().join("not-yet.db"); + let resp = handle(req(&format!("sqlite:{}", missing.display()))) + .await + .unwrap(); + assert!(!resp.ok); + assert!(resp.message.unwrap().contains("does not exist yet")); + // The probe must not have created it. + assert!(!missing.exists()); + } + + #[tokio::test(flavor = "multi_thread")] + async fn unknown_scheme_reports_without_erroring() { + let resp = handle(req("mongodb://nope")).await.unwrap(); + assert!(!resp.ok); + assert_eq!(resp.driver, "unknown"); + assert!(resp.message.unwrap().contains("unknown url scheme")); + } + + #[tokio::test(flavor = "multi_thread")] + async fn refused_postgres_fails_fast_and_scrubs_credentials() { + // Port 1 refuses instantly; the error must carry neither the + // password nor the username from the probed url. + let mut r = req("postgres://leaky_user:leaky_pass@127.0.0.1:1/db"); + r.tls = Some(TlsConfig { + mode: crate::config::TlsMode::Disable, + ..Default::default() + }); + let resp = handle(r).await.unwrap(); + assert!(!resp.ok); + assert_eq!(resp.driver, "postgres"); + let message = resp.message.unwrap(); + assert!(!message.contains("leaky_pass"), "{message}"); + assert!(!message.contains("leaky_user:"), "{message}"); + } + + #[test] + fn scrub_replaces_userinfo_everywhere() { + let scrubbed = scrub_credentials( + "connect to postgres://u:sekret@h failed for u:sekret", + "postgres://u:sekret@h/db", + ); + assert!(!scrubbed.contains("sekret"), "{scrubbed}"); + } + + /// Live probes, gated like the pool tests. + #[tokio::test(flavor = "multi_thread")] + async fn live_postgres_and_mysql_probe_ok_when_configured() { + for (env, prefix) in [ + ("TEST_POSTGRES_URL", "postgres"), + ("TEST_MYSQL_URL", "mysql"), + ] { + let Some(url) = std::env::var(env).ok() else { + eprintln!("skipping: {env} not set"); + continue; + }; + let mut r = req(&url); + r.tls = Some(TlsConfig { + mode: crate::config::TlsMode::Disable, + ..Default::default() + }); + let resp = handle(r).await.unwrap(); + assert!(resp.ok, "{env}: {:?}", resp.message); + assert_eq!(resp.driver, prefix); + assert!(resp.server_version.is_some()); + } + } +} diff --git a/database/src/handlers/transaction_execute.rs b/database/src/handlers/transaction_execute.rs index ef0724fdb..698f76c66 100644 --- a/database/src/handlers/transaction_execute.rs +++ b/database/src/handlers/transaction_execute.rs @@ -120,13 +120,15 @@ pub async fn handle(state: &AppState, req: TxExecuteReq) -> Result Result<()> { "starting" ); + // Identify as `database` in the console's workers view instead of the + // SDK's hostname:pid fallback. III_WORKER_NAME still wins — that is the + // managed-spawn identity contract (the engine exports it for workers it + // owns), and hand-run instances (workers-dev) can use it to tag + // themselves per worktree. + let mut metadata = iii_sdk::runtime::WorkerMetadata::default(); + if std::env::var("III_WORKER_NAME").map_or(true, |v| v.is_empty()) { + metadata.name = database::worker_name().to_string(); + } + metadata.description = Some( + "SQL for PostgreSQL, MySQL, and SQLite: queries, statements, interactive \ + transactions, and database::row-changed triggers (statements or native capture)." + .to_string(), + ); + // Arc-wrapped for `ui::register` (the console-ui crate clones the client // into its hot-reload watcher task); everything else auto-derefs. let iii = Arc::new(register_worker( &cli.url, InitOptions { + metadata: Some(metadata), otel: Some(OtelConfig::default()), ..Default::default() }, @@ -110,6 +127,12 @@ async fn main() -> Result<()> { iii.clone(), ROW_CHANGE_DISPATCH_TIMEOUT_MS, )); + // One LISTEN task per `capture: native` postgres database, live from + // startup — external writes must be heard before any binding registers. + let native_listeners = Arc::new(database::triggers::NativeListeners::new( + row_changes.clone(), + )); + native_listeners.sync(&cfg); let state = AppState { pools: Arc::new(RwLock::new(pools)), config: Arc::new(RwLock::new(cfg)), @@ -296,6 +319,21 @@ async fn main() -> Result<()> { .description("Rollback and finalize an interactive transaction."), ); } + { + iii.register_function( + "database::testConnection", + RegisterFunction::new_async(move |req: TestConnectionReq| async move { + test_connection::handle(req) + .await + .map_err(iii_sdk::errors::Error::from) + }) + .description( + "Probe a candidate database config (url + optional tls) with one \ + throwaway connection, without touching configured pools. Reports \ + ok/driver/latency/server version; failures are data, not errors.", + ), + ); + } { let st = state.clone(); iii.register_function( @@ -327,13 +365,14 @@ async fn main() -> Result<()> { database::triggers::RowChangedHandler { bus: row_changes.clone(), config: state.config.clone(), + pools: state.pools.clone(), }, ) .trigger_request_format::() .call_request_format::(), ); - configuration::register_config_trigger(&iii, state.clone()) + configuration::register_config_trigger(&iii, state.clone(), Some(native_listeners.clone())) .context("registering configuration change trigger")?; // Injectable console UI (function-trigger renderer) — after the @@ -341,7 +380,7 @@ async fn main() -> Result<()> { database::ui::register(&iii); tracing::info!( - "database worker registered 13 functions and 1 trigger type, waiting for invocations" + "database worker registered 14 functions and 1 trigger type, waiting for invocations" ); wait_for_shutdown_signal().await?; tracing::info!("database worker shutting down"); diff --git a/database/src/triggers/bus.rs b/database/src/triggers/bus.rs index 54bd54820..894af3a60 100644 --- a/database/src/triggers/bus.rs +++ b/database/src/triggers/bus.rs @@ -227,6 +227,12 @@ impl RowChangeBus { self.lock().pending.get(transaction_id).map_or(0, Vec::len) } + /// Announce an already-shaped event — the native capture path, where the + /// database told us table/op/count and there is no SQL to classify. + pub async fn emit_event(&self, event: RowChangedEvent) { + self.fan_out(event).await; + } + /// Announce a committed change. `sql` is classified here so callers stay /// one line; a statement that changes no rows fires nothing. pub async fn emit( @@ -328,7 +334,7 @@ impl RowChangeBus { } } -fn now_ms() -> i64 { +pub(crate) fn now_ms() -> i64 { use std::time::{SystemTime, UNIX_EPOCH}; SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/database/src/triggers/handler.rs b/database/src/triggers/handler.rs index 8f962f89e..66260751c 100644 --- a/database/src/triggers/handler.rs +++ b/database/src/triggers/handler.rs @@ -5,6 +5,7 @@ //! emitting. Registration fails loudly for a database that is not configured — //! a binding on a typo'd handle would otherwise sit there listening to nothing. +use std::collections::HashMap; use std::sync::Arc; use async_trait::async_trait; @@ -12,18 +13,44 @@ use iii_sdk::errors::Error; use iii_sdk::trigger::{TriggerConfig, TriggerHandler}; use super::bus::{RowChangeBus, RowChangedConfig}; -use crate::config::WorkerConfig; +use super::native; +use crate::config::{CaptureMode, WorkerConfig}; +use crate::pool::Pool; pub struct RowChangedHandler { pub bus: Arc, /// Live configuration, swapped together with the pools on hot reload. pub config: Arc>, + /// Live pools — a `capture: native` binding installs its database + /// triggers through the bound database's pool at registration time. + pub pools: Arc>>, } fn config_error(message: String) -> Error { Error::Handler(serde_json::json!({ "code": "CONFIG_ERROR", "message": message }).to_string()) } +/// Pick the guidance appended to a postgres trigger-install failure. The +/// hint must match the actual failure: dressing a `does not exist` error in +/// privilege advice sent a real operator chasing grants when the problem +/// was table-name casing (native bindings quote the name verbatim, and +/// quoted postgres identifiers are case-sensitive). +fn pg_install_hint(error_text: &str, table: &str) -> String { + if error_text.contains("does not exist") { + format!( + ". Note: the binding's table name is quoted verbatim into DDL and \ + quoted postgres identifiers are case-sensitive — `{table}` must \ + match the table's actual spelling" + ) + } else if error_text.contains("permission denied") || error_text.contains("must be owner") { + ". The configured role needs TRIGGER privilege on the table (or ownership) \ + and CREATE on its schema" + .to_string() + } else { + String::new() + } +} + #[async_trait] impl TriggerHandler for RowChangedHandler { async fn register_trigger(&self, config: TriggerConfig) -> Result<(), Error> { @@ -31,7 +58,7 @@ impl TriggerHandler for RowChangedHandler { .map_err(|e| config_error(format!("row-changed config: {e}")))?; let live = self.config.read().await; - if !live.databases.contains_key(&cfg.db) { + let Some(db_cfg) = live.databases.get(&cfg.db) else { let mut known = live.databases.keys().cloned().collect::>(); known.sort(); return Err(config_error(format!( @@ -39,9 +66,14 @@ impl TriggerHandler for RowChangedHandler { cfg.db, known.join(", ") ))); - } + }; + let native = db_cfg.capture == CaptureMode::Native; drop(live); + if native { + self.install_native_triggers(&cfg).await?; + } + let table = cfg.table.clone(); self.bus.register( config.id.clone(), @@ -59,12 +91,108 @@ impl TriggerHandler for RowChangedHandler { } async fn unregister_trigger(&self, config: TriggerConfig) -> Result<(), Error> { + // ponytail: native-capture triggers stay installed on unregister — + // an orphan pg_notify per write statement is near-free and idempotent + // to reinstall; add DDL teardown when someone actually needs it. self.bus.unregister(&config.id); tracing::info!(instance = %config.id, "row-changed trigger unregistered"); Ok(()) } } +impl RowChangedHandler { + /// Install the NOTIFY function and per-table triggers for a native + /// binding. Fails loudly — a binding whose DDL did not land would sit + /// there hearing nothing, which is the failure mode this worker refuses. + async fn install_native_triggers(&self, cfg: &RowChangedConfig) -> Result<(), Error> { + let Some(table) = cfg.table.as_deref() else { + return Err(config_error(format!( + "db `{}` uses `capture: native`, which requires this binding to \ + name a `table` — per-table database triggers are what make \ + external writes visible", + cfg.db + ))); + }; + let pool = self.pools.read().await.get(&cfg.db).cloned(); + match pool { + Some(Pool::Postgres(pg)) => { + let sql = native::install_sql(table).map_err(config_error)?; + let client = pg.acquire().await.map_err(|e| { + config_error(format!("db `{}`: acquiring connection: {e}", cfg.db)) + })?; + client.batch_execute(&sql).await.map_err(|e| { + let text = e.to_string(); + config_error(format!( + "installing native capture triggers on `{table}`: {text}{}", + pg_install_hint(&text, table) + )) + })?; + } + Some(Pool::Sqlite(sq)) => { + let sql = super::sqlite_watch::install_sql(table).map_err(config_error)?; + let conn = sq.acquire().await.map_err(|e| { + config_error(format!("db `{}`: acquiring connection: {e}", cfg.db)) + })?; + let table_for_err = table.to_string(); + tokio::task::spawn_blocking(move || conn.with(|c| c.execute_batch(&sql))) + .await + .map_err(|e| config_error(format!("sqlite DDL join: {e}")))? + .map_err(|e| { + config_error(format!( + "installing native capture triggers on `{table_for_err}`: {e}" + )) + })?; + } + Some(Pool::Mysql(my)) => { + // Binlog capture installs nothing — but a binding on a server + // that cannot be streamed would sit silent forever. Verify + // the prerequisites here, where the failure is actionable. + use mysql_async::prelude::Queryable as _; + let mut conn = my.acquire().await.map_err(|e| { + config_error(format!("db `{}`: acquiring connection: {e}", cfg.db)) + })?; + let settings: Option<(i64, String)> = conn + .query_first("SELECT @@log_bin, @@binlog_format") + .await + .map_err(|e| config_error(format!("db `{}`: {e}", cfg.db)))?; + match settings { + Some((1, format)) if format.eq_ignore_ascii_case("ROW") => {} + Some((1, format)) => { + return Err(config_error(format!( + "db `{}`: binlog_format is {format}; native capture needs ROW \ + (SET GLOBAL binlog_format = 'ROW', the 8.x default)", + cfg.db + ))); + } + _ => { + return Err(config_error(format!( + "db `{}`: the server runs without a binary log (log_bin=OFF); \ + native capture reads the binlog and cannot work here", + cfg.db + ))); + } + } + // Doubles as the privilege probe: needs REPLICATION CLIENT, + // and the stream itself needs REPLICATION SLAVE. + super::mysql_binlog::binlog_position(&mut conn) + .await + .map_err(|e| config_error(format!("db `{}`: {e}", cfg.db)))?; + } + None => { + // Every driver supports native capture, so reaching this arm + // means exactly one thing: config and pools drifted, which + // apply_config forbids. + return Err(config_error(format!( + "db `{}`: no pool available for native capture", + cfg.db + ))); + } + } + tracing::info!(db = %cfg.db, table = %table, "native capture triggers installed"); + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; @@ -78,17 +206,67 @@ mod tests { } } - #[tokio::test] - async fn registration_uses_the_live_database_config() { - let config = Arc::new(tokio::sync::RwLock::new(WorkerConfig::default())); - let bus = Arc::new(RowChangeBus::new( - Arc::new(iii_sdk::IIIClient::new("ws://127.0.0.1:9")), - 100, - )); + fn handler_with( + config: WorkerConfig, + ) -> (RowChangedHandler, Arc>) { + let config = Arc::new(tokio::sync::RwLock::new(config)); let handler = RowChangedHandler { - bus, + bus: Arc::new(RowChangeBus::new( + Arc::new(iii_sdk::IIIClient::new("ws://127.0.0.1:9")), + 100, + )), config: config.clone(), + pools: Arc::new(tokio::sync::RwLock::new(HashMap::new())), }; + (handler, config) + } + + #[test] + fn pg_install_hint_matches_the_failure_shape() { + // The bug this pins: a `does not exist` failure wrapped in privilege + // advice reads as a grants problem and hides the real cause (casing). + let hint = pg_install_hint( + r#"relation "III_TRIGGER_TEST" does not exist"#, + "III_TRIGGER_TEST", + ); + assert!(hint.contains("case-sensitive"), "{hint}"); + assert!(!hint.contains("TRIGGER privilege"), "{hint}"); + + let hint = pg_install_hint("permission denied for table orders", "orders"); + assert!(hint.contains("TRIGGER privilege"), "{hint}"); + let hint = pg_install_hint("must be owner of relation orders", "orders"); + assert!(hint.contains("TRIGGER privilege"), "{hint}"); + + // Anything else gets the raw error only — no guessed guidance. + assert_eq!(pg_install_hint("connection reset by peer", "orders"), ""); + } + + #[tokio::test] + async fn native_bindings_must_name_a_table() { + let cfg = WorkerConfig::from_yaml( + "databases:\n p:\n url: postgres://u@h/db\n capture: native\n", + ) + .unwrap(); + let (handler, _) = handler_with(cfg); + + let err = handler + .register_trigger(trigger("i1", "p")) + .await + .unwrap_err(); + assert!(err.to_string().contains("name a `table`"), "{err}"); + + // With a table but no live pool the DDL cannot land; registration + // still fails loudly instead of listening to nothing. + let mut with_table = trigger("i2", "p"); + with_table.config = serde_json::json!({ "db": "p", "table": "orders" }); + let err = handler.register_trigger(with_table).await.unwrap_err(); + assert!(err.to_string().contains("no pool"), "{err}"); + assert_eq!(handler.bus.subscriber_count(), 0); + } + + #[tokio::test] + async fn registration_uses_the_live_database_config() { + let (handler, config) = handler_with(WorkerConfig::default()); handler .register_trigger(trigger("initial", "primary")) diff --git a/database/src/triggers/mod.rs b/database/src/triggers/mod.rs index 7a105d931..51c7550d2 100644 --- a/database/src/triggers/mod.rs +++ b/database/src/triggers/mod.rs @@ -5,16 +5,32 @@ //! events, so a coordinator had to be told out-of-band (a state key written //! alongside the row) or poll. //! -//! This is deliberately NOT change data capture. It reports mutations THIS -//! worker performed, on commit, by classifying the SQL it was given. A write -//! applied by psql, another worker, or a database-side trigger is invisible -//! here. That covers the case it exists for — agents whose only writer is this -//! worker — and nothing more, which is why it needs no logical replication, no -//! driver-specific setup, and works identically on SQLite, Postgres and MySQL. +//! Two capture modes, chosen per database in the worker config: +//! +//! * `statements` (default): report mutations THIS worker performed, on +//! commit, by classifying the SQL it was given. A write applied by psql, +//! another worker, or a database-side trigger is invisible. No database +//! setup, works identically on SQLite, Postgres and MySQL. +//! * `native`: committed writes from ANY client fire events; table-scoped +//! bindings only. Postgres (`native.rs`): triggers + LISTEN/NOTIFY on a +//! dedicated connection. File-backed sqlite (`sqlite_watch.rs`): triggers → +//! changelog table → fs-watch drain. MySQL (`mysql_binlog.rs`): the binlog +//! replication stream — nothing installed in the schema at all. +//! +//! Native delivery is at-most-once by contract: events raised while a +//! listener is down — including the instants around a hot reload that +//! restarts it, or an interactive transaction that outlives a reload of its +//! database's config and commits on the OLD database — can be lost. It is a +//! doorbell, not a ledger; subscribers needing a gapless view reconcile on +//! their own schedule. pub mod bus; pub mod handler; +pub mod mysql_binlog; +pub mod native; pub mod sql; +pub mod sqlite_watch; pub use bus::{RowChangeBus, RowChangedConfig, RowChangedEvent, ROW_CHANGED_TYPE}; pub use handler::RowChangedHandler; +pub use native::NativeListeners; diff --git a/database/src/triggers/mysql_binlog.rs b/database/src/triggers/mysql_binlog.rs new file mode 100644 index 000000000..94d732fbe --- /dev/null +++ b/database/src/triggers/mysql_binlog.rs @@ -0,0 +1,339 @@ +//! Native change capture for MySQL: the binlog replication stream. +//! +//! MySQL has no LISTEN/NOTIFY, but it has something stronger — the binary +//! log every replica reads. The worker connects as a replica (one dedicated +//! connection per `capture: native` database), starts at the server's +//! current position, and decodes row events into `database::row-changed` +//! events. Nothing is installed in the user's schema: no triggers, no +//! changelog table, no DDL at binding registration. +//! +//! Semantics match the other drivers: only committed writes appear in the +//! binlog (row events are flushed at commit), so commit gating is free; a +//! reconnect re-snapshots the position, so events raised while the stream +//! was down are lost — at-most-once, a doorbell not a ledger. +//! +//! Server prerequisites, checked loudly at binding registration: +//! `log_bin=ON`, `binlog_format=ROW` (both 8.x defaults), and the +//! `REPLICATION SLAVE, REPLICATION CLIENT` global grants for the worker's +//! user. + +use std::sync::Arc; +use std::time::Duration; + +use futures_util::StreamExt; +use mysql_async::binlog::events::{EventData, RowsEventData}; +use mysql_async::prelude::Queryable; +use mysql_async::{BinlogStreamRequest, Conn, Opts, OptsBuilder}; + +use super::bus::{now_ms, RowChangeBus, RowChangedEvent}; +use super::sql::Op; +use crate::config::TlsConfig; +use crate::pool::tls::make_mysql_ssl_opts; + +/// Statements that report the current binlog file/position. 8.2 renamed the +/// classic one; try newest first, fall back on "unknown statement". +const POSITION_STATEMENTS: [&str; 2] = ["SHOW BINARY LOG STATUS", "SHOW MASTER STATUS"]; + +/// The GRANT hint quoted in every privilege-shaped failure. Kept in one +/// place so registration errors and stream errors say the same thing. +pub(crate) const GRANT_HINT: &str = + "the worker's user needs: GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ''@'%'"; + +/// A server id for COM_REGISTER_SLAVE that stays clear of the low range +/// operators typically hand-assign to real replicas. Collisions only matter +/// between simultaneous replicas of the same server: the handle name keeps +/// two `capture: native` databases in ONE worker apart (same-id replicas +/// evict each other and reconnect-thrash forever), and the pid keeps +/// concurrent workers on one host apart. +fn server_id(db_name: &str) -> u32 { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + db_name.hash(&mut hasher); + std::process::id().hash(&mut hasher); + 1_000_000_000 + (hasher.finish() % 1_000_000) as u32 +} + +pub(crate) fn build_opts(url: &str, tls: &TlsConfig) -> Result { + let base = Opts::from_url(url).map_err(|_| "invalid mysql url".to_string())?; + let mut builder = OptsBuilder::from_opts(base); + if let Some(ssl) = make_mysql_ssl_opts(tls).map_err(|e| format!("{e:?}"))? { + builder = builder.ssl_opts(ssl); + } + Ok(builder.into()) +} + +/// The current binlog (file, position), or an actionable error. Requires +/// REPLICATION CLIENT — this doubles as the registration-time privilege +/// probe. +pub(crate) async fn binlog_position(conn: &mut Conn) -> Result<(String, u64), String> { + let mut last_err = String::new(); + for sql in POSITION_STATEMENTS { + match conn.query_first::(sql).await { + Ok(Some(row)) => { + let file: Option = row.get(0); + let pos: Option = row.get(1); + match (file, pos) { + (Some(file), Some(pos)) => return Ok((file, pos)), + _ => return Err(format!("`{sql}` returned an unreadable row")), + } + } + Ok(None) => { + return Err( + "the server reports no binlog position — is log_bin enabled?".to_string(), + ) + } + Err(e) => { + let msg = e.to_string(); + // 8.2 removed SHOW MASTER STATUS' predecessor and older + // servers don't know the new form — try the other spelling. + if msg.contains("error in your SQL syntax") || msg.contains("Unknown") { + last_err = msg; + continue; + } + return Err(format!("{msg}; {GRANT_HINT}")); + } + } + } + Err(format!("{last_err}; {GRANT_HINT}")) +} + +fn op_of(rows: &RowsEventData<'_>) -> Op { + match rows { + RowsEventData::WriteRowsEvent(_) | RowsEventData::WriteRowsEventV1(_) => Op::Insert, + RowsEventData::UpdateRowsEvent(_) + | RowsEventData::UpdateRowsEventV1(_) + | RowsEventData::PartialUpdateRowsEvent(_) => Op::Update, + RowsEventData::DeleteRowsEvent(_) | RowsEventData::DeleteRowsEventV1(_) => Op::Delete, + } +} + +/// Keep the stream alive forever, reconnecting with capped backoff. Events +/// flow into `bus` via an unbounded channel so the decode loop never blocks +/// on subscriber dispatch. +pub(crate) async fn run_binlog( + db_name: String, + url: String, + tls: TlsConfig, + bus: Arc, +) { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); + // The forwarder dies with this task: aborting run_binlog drops `tx`, + // recv() yields None, and the spawned task returns. + let bus_for_forwarder = Arc::clone(&bus); + tokio::spawn(async move { + while let Some(event) = rx.recv().await { + bus_for_forwarder.emit_event(event).await; + } + }); + + let mut delay = Duration::from_secs(1); + loop { + match stream_once(&db_name, &url, &tls, &tx).await { + Ok(()) => { + tracing::warn!(db = %db_name, "binlog stream ended; reconnecting"); + delay = Duration::from_secs(1); + } + Err(e) => { + tracing::warn!(db = %db_name, error = %e, "binlog capture error; retrying"); + } + } + tokio::time::sleep(delay).await; + delay = (delay * 2).min(Duration::from_secs(30)); + } +} + +/// One replica session: snapshot position, stream, decode, emit. +async fn stream_once( + db_name: &str, + url: &str, + tls: &TlsConfig, + out: &tokio::sync::mpsc::UnboundedSender, +) -> Result<(), String> { + let opts = build_opts(url, tls)?; + // Events for OTHER databases on the same server are not this handle's + // business — a binding names a db handle, and the handle names a schema. + let schema = opts.db_name().map(str::to_string); + let mut conn = Conn::new(opts).await.map_err(|e| e.to_string())?; + let (file, pos) = binlog_position(&mut conn).await?; + let mut stream = conn + .get_binlog_stream( + BinlogStreamRequest::new(server_id(db_name)) + .with_filename(file.as_bytes()) + .with_pos(pos), + ) + .await + .map_err(|e| format!("{e}; {GRANT_HINT}"))?; + tracing::info!(db = %db_name, file = %file, pos, "native capture streaming binlog"); + + // One statement's rows can arrive chunked across several events; merge + // ADJACENT same-(table, op) row events and flush on any other event — + // every transaction ends with a non-rows event (Xid), so nothing is + // held past its commit. TableMapEvent is a flush point too: in row + // format every STATEMENT re-maps its table before its rows events, + // while the chunks of one statement share a single map — so flushing + // there yields exactly one event per statement (matching postgres) + // without breaking chunk merging. + let mut pending: Option<(String, Op, u64)> = None; + let flush = |pending: &mut Option<(String, Op, u64)>| { + if let Some((table, op, n)) = pending.take() { + let _ = out.send(RowChangedEvent { + db: db_name.to_string(), + table: Some(table), + op, + affected_rows: n, + returning: None, + at: now_ms(), + }); + } + }; + + while let Some(event) = stream.next().await { + let event = event.map_err(|e| e.to_string())?; + let data = match event.read_data() { + Ok(Some(data)) => data, + // Undecodable/unknown events still delimit statements. + _ => { + flush(&mut pending); + continue; + } + }; + match data { + EventData::RowsEvent(rows) => { + let op = op_of(&rows); + let Some(tme) = stream.get_tme(rows.table_id()) else { + // No table map — cannot attribute; drop rather than lie. + flush(&mut pending); + continue; + }; + if schema.as_deref().is_some_and(|s| tme.database_name() != s) { + flush(&mut pending); + continue; + } + let table = tme.table_name().to_string(); + let n = rows.rows(tme).count() as u64; + if n == 0 { + continue; + } + match &mut pending { + Some((last_table, last_op, total)) + if *last_table == table && *last_op == op => + { + *total += n; + } + slot => { + flush(slot); + *slot = Some((table, op, n)); + } + } + } + // A table map opens the next statement's rows — statement boundary. + EventData::TableMapEvent(_) => flush(&mut pending), + // Anything else (Xid, Query, Rotate, Gtid, …) ends a statement. + _ => flush(&mut pending), + } + } + flush(&mut pending); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn server_id_stays_in_range_and_separates_handles() { + let a = server_id("primary"); + let b = server_id("analytics"); + for id in [a, b] { + assert!((1_000_000_000..1_001_000_000).contains(&id)); + } + // Two native handles in one worker must register as DIFFERENT + // replicas — the server evicts duplicate server ids. + assert_ne!(a, b); + // Deterministic within a process: reconnects keep their identity. + assert_eq!(a, server_id("primary")); + } + + /// The cross-client claim, mysql edition: writes from a plain client + /// connection arrive through the replica stream. Requires + /// TEST_MYSQL_URL and replication grants for that user; fails (not + /// skips) without the grants — the error names the exact GRANT. + #[tokio::test(flavor = "multi_thread")] + async fn binlog_capture_hears_writes_from_another_connection() { + let Some(url) = std::env::var("TEST_MYSQL_URL").ok() else { + eprintln!("skipping: TEST_MYSQL_URL not set"); + return; + }; + let tls = TlsConfig { + mode: crate::config::TlsMode::Disable, + ..Default::default() + }; + + let table = format!("iii_binlog_capture_{}", std::process::id()); + let writer_pool = mysql_async::Pool::new(url.as_str()); + let mut writer = writer_pool.get_conn().await.unwrap(); + writer + .query_drop(format!("DROP TABLE IF EXISTS {table}")) + .await + .unwrap(); + writer + .query_drop(format!("CREATE TABLE {table} (id INT PRIMARY KEY, n INT)")) + .await + .unwrap(); + + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let streamer = { + let url = url.clone(); + let db = "primary".to_string(); + tokio::spawn(async move { + if let Err(e) = stream_once(&db, &url, &tls, &tx).await { + panic!("stream_once failed: {e}"); + } + }) + }; + // Give the replica session a moment to snapshot + attach. + tokio::time::sleep(Duration::from_millis(500)).await; + + writer + .query_drop(format!("INSERT INTO {table} VALUES (1, 10), (2, 20)")) + .await + .unwrap(); + writer + .query_drop(format!("UPDATE {table} SET n = n + 1")) + .await + .unwrap(); + writer + .query_drop(format!("DELETE FROM {table} WHERE id = 1")) + .await + .unwrap(); + + let mut events = Vec::new(); + while events.len() < 3 { + let event = tokio::time::timeout(Duration::from_secs(10), rx.recv()) + .await + .expect("binlog event within 10s") + .expect("stream alive"); + // The server may interleave writes from other databases/tests; + // keep only our table's events. + if event.table.as_deref() == Some(table.as_str()) { + events.push(event); + } + } + assert_eq!(events[0].op, Op::Insert); + assert_eq!(events[0].affected_rows, 2); + assert_eq!(events[0].db, "primary"); + assert!(events[0].returning.is_none()); + assert_eq!(events[1].op, Op::Update); + assert_eq!(events[1].affected_rows, 2); + assert_eq!(events[2].op, Op::Delete); + assert_eq!(events[2].affected_rows, 1); + + streamer.abort(); + writer + .query_drop(format!("DROP TABLE {table}")) + .await + .unwrap(); + drop(writer); + let _ = writer_pool.disconnect().await; + } +} diff --git a/database/src/triggers/native.rs b/database/src/triggers/native.rs new file mode 100644 index 000000000..e365229c7 --- /dev/null +++ b/database/src/triggers/native.rs @@ -0,0 +1,551 @@ +//! Native change capture for Postgres: database triggers + LISTEN/NOTIFY. +//! +//! The statements path (`bus.rs`) hears only what this worker executes. A +//! database configured with `capture: native` instead installs an AFTER +//! trigger per bound table that `pg_notify`s a small JSON payload, and the +//! worker holds one dedicated (non-pooled) connection per database doing +//! LISTEN. Any client's committed write — psql, another worker, another +//! process — fires the same `database::row-changed` event. +//! +//! Delivery is NOTIFY's: commit-gated (nothing fires for rolled-back +//! transactions) but at-most-once — notifications raised while the listener +//! connection is down are lost. Subscribers that cannot tolerate a gap must +//! reconcile on their own schedule; this is a doorbell, not a ledger. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use serde::Deserialize; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio_postgres::{AsyncMessage, Client, Connection, NoTls, Socket}; + +use super::bus::{now_ms, RowChangeBus, RowChangedEvent}; +use super::sql::Op; +use crate::config::{CaptureMode, TlsConfig, WorkerConfig}; +use crate::pool::tls::make_pg_connector; + +/// The NOTIFY channel every iii trigger function raises on. +pub(crate) const CHANNEL: &str = "iii_row_changed"; + +/// The DDL that makes one table announce its changes: a shared trigger +/// function (idempotent to reinstall) plus three statement-level triggers. +/// Statement-level with transition tables gives a real row count without a +/// NOTIFY per row; `IF n > 0` keeps the existing "no rows, no event" rule. +pub(crate) fn install_sql(table: &str) -> Result { + let target = quote_table(table)?; + Ok(format!( + r#"CREATE OR REPLACE FUNCTION iii_row_changed_notify() RETURNS trigger +LANGUAGE plpgsql AS $iii$ +DECLARE n bigint := 0; +BEGIN + IF TG_OP = 'DELETE' THEN + SELECT count(*) INTO n FROM old_rows; + ELSE + SELECT count(*) INTO n FROM new_rows; + END IF; + IF n > 0 THEN + PERFORM pg_notify('{CHANNEL}', json_build_object( + 'table', TG_TABLE_SCHEMA || '.' || TG_TABLE_NAME, + 'op', lower(TG_OP), + 'n', n)::text); + END IF; + RETURN NULL; +END +$iii$; +DROP TRIGGER IF EXISTS iii_row_changed_ins ON {target}; +CREATE TRIGGER iii_row_changed_ins AFTER INSERT ON {target} + REFERENCING NEW TABLE AS new_rows FOR EACH STATEMENT + EXECUTE FUNCTION iii_row_changed_notify(); +DROP TRIGGER IF EXISTS iii_row_changed_upd ON {target}; +CREATE TRIGGER iii_row_changed_upd AFTER UPDATE ON {target} + REFERENCING NEW TABLE AS new_rows FOR EACH STATEMENT + EXECUTE FUNCTION iii_row_changed_notify(); +DROP TRIGGER IF EXISTS iii_row_changed_del ON {target}; +CREATE TRIGGER iii_row_changed_del AFTER DELETE ON {target} + REFERENCING OLD TABLE AS old_rows FOR EACH STATEMENT + EXECUTE FUNCTION iii_row_changed_notify(); +"# + )) +} + +/// Quote a `table` or `schema.table` reference so it is only ever an +/// identifier — binding config is a trust boundary and this string lands in +/// DDL. A name that does not exist fails loudly at CREATE TRIGGER. +/// Shared with the sqlite watcher: `"…"` quoting is valid in both dialects. +pub(crate) fn quote_table(t: &str) -> Result { + let t = t.trim(); + if t.is_empty() { + return Err("table name is empty".into()); + } + let parts: Vec<&str> = t.split('.').collect(); + if parts.len() > 2 { + return Err(format!("table `{t}` must be `table` or `schema.table`")); + } + Ok(parts + .iter() + .map(|p| { + let p = p.trim(); + // Accept an already-quoted part without double-wrapping it. + let bare = p + .strip_prefix('"') + .and_then(|s| s.strip_suffix('"')) + .unwrap_or(p); + format!("\"{}\"", bare.replace('"', "\"\"")) + }) + .collect::>() + .join(".")) +} + +/// What the trigger function sends. `op` reuses the wire enum, so +/// `lower(TG_OP)` maps directly onto insert/update/delete. +#[derive(Deserialize)] +struct Payload { + table: String, + op: Op, + n: u64, +} + +/// A NOTIFY payload as a bus event, or None (with a warning) for payloads +/// this worker did not shape — someone else may NOTIFY on our channel. +pub(crate) fn parse_notification(db: &str, payload: &str) -> Option { + let p: Payload = match serde_json::from_str(payload) { + Ok(p) => p, + Err(e) => { + tracing::warn!(db = %db, error = %e, "unparseable payload on {CHANNEL}; dropped"); + return None; + } + }; + Some(RowChangedEvent { + db: db.to_string(), + table: Some(p.table), + op: p.op, + affected_rows: p.n, + returning: None, + at: now_ms(), + }) +} + +enum TaskHandle { + /// Postgres LISTEN / mysql binlog task — a tokio task, aborted on removal. + Async(tokio::task::JoinHandle<()>), + /// Sqlite watcher — a dedicated OS thread (rusqlite is sync and the + /// connection must stay put); told to stop via flag plus a wake poke so + /// it exits within one drain instead of one fallback tick — a stopped + /// watcher lingering next to its replacement would double-drain and + /// double-GC the same changelog. + Thread { + stop: Arc, + wake: std::sync::mpsc::Sender<()>, + }, +} + +impl TaskHandle { + fn stop(&self) { + match self { + TaskHandle::Async(handle) => handle.abort(), + TaskHandle::Thread { stop, wake } => { + stop.store(true, std::sync::atomic::Ordering::Relaxed); + let _ = wake.send(()); + } + } + } +} + +struct ListenerTask { + /// Serialized DatabaseConfig; a reload that changes url/tls restarts the + /// listener, one that leaves the db untouched does not. + fingerprint: String, + handle: TaskHandle, +} + +/// One capture task per `capture: native` database — a LISTEN connection for +/// postgres, a changelog watcher thread for sqlite — reconciled against the +/// live config at startup and on every hot reload. +pub struct NativeListeners { + bus: Arc, + tasks: Mutex>, +} + +impl NativeListeners { + pub fn new(bus: Arc) -> Self { + Self { + bus, + tasks: Mutex::new(HashMap::new()), + } + } + + /// Start missing listeners, stop removed ones, restart changed ones. + /// Must run inside a tokio runtime. + pub fn sync(&self, cfg: &WorkerConfig) { + let desired: HashMap = cfg + .databases + .iter() + .filter(|(_, db)| db.capture == CaptureMode::Native) + .map(|(name, db)| { + let fingerprint = serde_json::to_string(db).unwrap_or_default(); + (name.clone(), (db.clone(), fingerprint)) + }) + .collect(); + + let mut tasks = self.tasks.lock().unwrap_or_else(|e| e.into_inner()); + tasks.retain(|name, task| { + let keep = desired + .get(name) + .is_some_and(|(_, fp)| *fp == task.fingerprint); + if !keep { + task.handle.stop(); + tracing::info!(db = %name, "native capture listener stopped"); + } + keep + }); + for (name, (db, fingerprint)) in desired { + if tasks.contains_key(&name) { + continue; + } + let Some(handle) = self.spawn(&name, &db) else { + continue; + }; + tasks.insert( + name, + ListenerTask { + fingerprint, + handle, + }, + ); + } + } + + fn spawn(&self, name: &str, db: &crate::config::DatabaseConfig) -> Option { + match db.driver { + crate::config::DriverKind::Postgres => { + Some(TaskHandle::Async(tokio::spawn(run_listener( + name.to_string(), + db.url.clone(), + db.tls.clone(), + Arc::clone(&self.bus), + )))) + } + crate::config::DriverKind::Sqlite => { + let Some(path) = super::sqlite_watch::sqlite_file_path(&db.url) else { + // Config validation rejects `:memory:`; reaching this + // means drift — fail visible, not silent. + tracing::warn!(db = %name, "native capture needs a file-backed sqlite url"); + return None; + }; + let stop = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let (wake_tx, wake_rx) = std::sync::mpsc::channel::<()>(); + let bus = Arc::clone(&self.bus); + let rt = tokio::runtime::Handle::current(); + let db_name = name.to_string(); + let thread_stop = Arc::clone(&stop); + let thread_wake = wake_tx.clone(); + if let Err(e) = std::thread::Builder::new() + .name(format!("sqlite-capture-{name}")) + .spawn(move || { + super::sqlite_watch::run_watcher( + &db_name, + &path, + &thread_stop, + thread_wake, + &wake_rx, + |event| { + // Bridge sync → async: the watcher thread parks + // on the runtime while the bus fans out. + rt.block_on(bus.emit_event(event)); + true + }, + ); + }) + { + // A database with no watcher hears nothing — that must + // never happen silently. + tracing::warn!(db = %name, error = %e, "sqlite capture watcher thread failed to spawn"); + return None; + } + Some(TaskHandle::Thread { + stop, + wake: wake_tx, + }) + } + crate::config::DriverKind::Mysql => Some(TaskHandle::Async(tokio::spawn( + super::mysql_binlog::run_binlog( + name.to_string(), + db.url.clone(), + db.tls.clone(), + Arc::clone(&self.bus), + ), + ))), + } + } + + #[cfg(test)] + pub(crate) fn task_count(&self) -> usize { + self.tasks.lock().unwrap_or_else(|e| e.into_inner()).len() + } +} + +/// Hold a LISTEN connection open forever, reconnecting with capped backoff. +/// Events raised while disconnected are lost — see the module doc. +async fn run_listener(db: String, url: String, tls: TlsConfig, bus: Arc) { + let mut delay = Duration::from_secs(1); + loop { + match listen_once(&db, &url, &tls, &bus).await { + Ok(()) => { + tracing::warn!(db = %db, "native capture connection closed; reconnecting"); + delay = Duration::from_secs(1); + } + Err(e) => { + tracing::warn!(db = %db, error = %e, "native capture listener error; retrying"); + } + } + tokio::time::sleep(delay).await; + delay = (delay * 2).min(Duration::from_secs(30)); + } +} + +async fn listen_once( + db: &str, + url: &str, + tls: &TlsConfig, + bus: &RowChangeBus, +) -> Result<(), String> { + // Dedicated connection, never the pool: LISTEN state is per-session, and + // a pooled session's notifications would go to whoever holds the object. + match make_pg_connector(tls).map_err(|e| format!("{e:?}"))? { + Some(connector) => { + let (client, conn) = tokio_postgres::connect(url, connector) + .await + .map_err(|e| e.to_string())?; + session(db, client, conn, bus).await + } + None => { + let (client, conn) = tokio_postgres::connect(url, NoTls) + .await + .map_err(|e| e.to_string())?; + session(db, client, conn, bus).await + } + } +} + +/// Drive one connection: issue LISTEN, then pump messages until the server +/// closes. `poll_message` both performs the connection's I/O and yields +/// notifications, so this single loop is the whole event pump. +async fn session( + db: &str, + client: Client, + mut conn: Connection, + bus: &RowChangeBus, +) -> Result<(), String> +where + S: AsyncRead + AsyncWrite + Unpin, +{ + let listen = client.batch_execute("LISTEN iii_row_changed"); + tokio::pin!(listen); + let mut listening = false; + loop { + tokio::select! { + r = &mut listen, if !listening => { + r.map_err(|e| e.to_string())?; + listening = true; + tracing::info!(db = %db, channel = CHANNEL, "native capture listening"); + } + msg = std::future::poll_fn(|cx| conn.poll_message(cx)) => match msg { + None => return Ok(()), + Some(Err(e)) => return Err(e.to_string()), + Some(Ok(AsyncMessage::Notification(n))) => { + if n.channel() == CHANNEL { + if let Some(event) = parse_notification(db, n.payload()) { + bus.emit_event(event).await; + } + } + } + // Notices and any future message kinds (enum is non_exhaustive). + Some(Ok(_)) => {} + }, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn install_sql_quotes_identifiers_and_rejects_garbage() { + let sql = install_sql("orders").unwrap(); + assert!(sql.contains("AFTER INSERT ON \"orders\"")); + assert!(sql.contains("AFTER UPDATE ON \"orders\"")); + assert!(sql.contains("AFTER DELETE ON \"orders\"")); + + let sql = install_sql("public.orders").unwrap(); + assert!(sql.contains("ON \"public\".\"orders\"")); + + // Already-quoted input is not double-wrapped. + let sql = install_sql("\"Orders\"").unwrap(); + assert!(sql.contains("ON \"Orders\"")); + + assert!(install_sql(" ").is_err()); + assert!(install_sql("a.b.c").is_err()); + } + + #[test] + fn install_sql_neutralizes_injection_attempts() { + // The binding config is a trust boundary; a hostile table name must + // come out as one (weird) quoted identifier, never as loose SQL. + let evil = r#"orders"; DROP TABLE users; --"#; + let sql = install_sql(evil).unwrap(); + assert!(sql.contains(r#"ON "orders""; DROP TABLE users; --""#)); + assert!(!sql.contains(r#"ON orders"#)); + } + + #[test] + fn parse_notification_maps_payloads_and_drops_foreign_ones() { + let ev = parse_notification( + "primary", + r#"{"table":"public.orders","op":"insert","n":3}"#, + ) + .unwrap(); + assert_eq!(ev.db, "primary"); + assert_eq!(ev.table.as_deref(), Some("public.orders")); + assert_eq!(ev.op, Op::Insert); + assert_eq!(ev.affected_rows, 3); + assert!(ev.returning.is_none()); + + // Someone else NOTIFYing on our channel must not become an event. + assert!(parse_notification("primary", "not json").is_none()); + assert!(parse_notification("primary", r#"{"table":"t","op":"vacuum","n":1}"#).is_none()); + assert!(parse_notification("primary", r#"{"op":"insert","n":1}"#).is_none()); + } + + #[tokio::test] + async fn sync_reconciles_listener_tasks_with_config() { + let bus = Arc::new(RowChangeBus::new( + Arc::new(iii_sdk::IIIClient::new("ws://127.0.0.1:9")), + 100, + )); + let listeners = NativeListeners::new(bus); + + let native = |url: &str| { + crate::config::WorkerConfig::from_yaml(&format!( + "databases:\n p:\n url: {url}\n capture: native\n tls:\n mode: disable\n" + )) + .unwrap() + }; + + // Port 1 refuses connections; the task just retries in background. + listeners.sync(&native("postgres://u@127.0.0.1:1/db")); + assert_eq!(listeners.task_count(), 1); + + // Same config → same task, not a restart. + listeners.sync(&native("postgres://u@127.0.0.1:1/db")); + assert_eq!(listeners.task_count(), 1); + + // Changed url → replaced. Removed → stopped. + listeners.sync(&native("postgres://u@127.0.0.1:2/db")); + assert_eq!(listeners.task_count(), 1); + listeners.sync(&crate::config::WorkerConfig::default()); + assert_eq!(listeners.task_count(), 0); + } + + /// The claim this feature exists for: a write from a *different + /// connection* (stand-in for a different process) raises a notification + /// the worker can parse. Requires TEST_POSTGRES_URL, like the pool tests. + #[tokio::test(flavor = "multi_thread")] + async fn native_capture_hears_writes_from_another_connection() { + let Some(url) = std::env::var("TEST_POSTGRES_URL").ok() else { + eprintln!("skipping: TEST_POSTGRES_URL not set"); + return; + }; + + let (listener, mut conn) = tokio_postgres::connect(&url, NoTls).await.unwrap(); + let (writer, writer_conn) = tokio_postgres::connect(&url, NoTls).await.unwrap(); + tokio::spawn(async move { + let _ = writer_conn.await; + }); + + /// Await a client call while pumping its connection — client futures + /// only resolve while someone polls the connection (`session` does + /// this for the real listener). Notices are consumed and dropped. + async fn drive( + conn: &mut Connection, + fut: impl std::future::Future, + ) -> T { + tokio::pin!(fut); + loop { + tokio::select! { + r = &mut fut => return r, + msg = std::future::poll_fn(|cx| conn.poll_message(cx)) => { + msg.expect("connection open").expect("no protocol error"); + } + } + } + } + + // Table + triggers, installed the way the handler installs them. + let table = format!("iii_native_capture_{}", std::process::id()); + drive(&mut conn, async { + listener + .batch_execute(&format!( + "DROP TABLE IF EXISTS {table}; CREATE TABLE {table} (id int, n int);" + )) + .await + .unwrap(); + listener + .batch_execute(&install_sql(&table).unwrap()) + .await + .unwrap(); + listener + .batch_execute("LISTEN iii_row_changed") + .await + .unwrap(); + }) + .await; + + // The "other process" writes: 2 inserts, 1 update, 1 delete. + writer + .batch_execute(&format!( + "INSERT INTO {table} VALUES (1, 10), (2, 20); \ + UPDATE {table} SET n = 5; \ + DELETE FROM {table} WHERE id = 1; \ + UPDATE {table} SET n = 9 WHERE id = 999;" // 0 rows → no event + )) + .await + .unwrap(); + + let mut events = Vec::new(); + while events.len() < 3 { + let msg = tokio::time::timeout( + Duration::from_secs(5), + std::future::poll_fn(|cx| conn.poll_message(cx)), + ) + .await + .expect("notification within 5s") + .expect("connection open") + .expect("no protocol error"); + if let AsyncMessage::Notification(n) = msg { + assert_eq!(n.channel(), CHANNEL); + events.push(parse_notification("primary", n.payload()).unwrap()); + } + } + + assert_eq!(events[0].op, Op::Insert); + assert_eq!(events[0].affected_rows, 2); + assert_eq!(events[1].op, Op::Update); + assert_eq!(events[1].affected_rows, 2); + assert_eq!(events[2].op, Op::Delete); + assert_eq!(events[2].affected_rows, 1); + for ev in &events { + assert!(crate::triggers::sql::same_table( + ev.table.as_deref().unwrap(), + &table + )); + } + + let _ = drive( + &mut conn, + listener.batch_execute(&format!("DROP TABLE {table}")), + ) + .await; + } +} diff --git a/database/src/triggers/sqlite_watch.rs b/database/src/triggers/sqlite_watch.rs new file mode 100644 index 000000000..c1c6799ee --- /dev/null +++ b/database/src/triggers/sqlite_watch.rs @@ -0,0 +1,526 @@ +//! Native change capture for file-backed SQLite: triggers + changelog + watch. +//! +//! SQLite is embedded — there is no server to broadcast "someone wrote". +//! What does exist: SQL triggers fire for ANY process's writes, and their +//! inserts into a changelog table commit atomically with the write itself +//! (a rollback removes them — commit gating is free and exact). Delivery is +//! this worker draining that changelog. Wake-up is event-driven: an fs watch +//! on the database file (inotify & co. via `notify`) plus `PRAGMA +//! data_version` — which changes iff another connection committed — as the +//! cheap confirm gate, with a slow fallback tick so a missed fs event +//! degrades latency instead of dropping anything. The changelog is the +//! source of truth; the watch only decides when to look. +//! +//! Boot behavior matches the postgres path: the cursor starts at the current +//! changelog head, so writes made while no worker was running are not +//! replayed. At-most-once, a doorbell not a ledger. + +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc; +use std::time::Duration; + +use notify::{RecursiveMode, Watcher}; +use rusqlite::Connection; + +use super::bus::{now_ms, RowChangedEvent}; +use super::native::quote_table; +use super::sql::Op; + +/// The changelog every captured table's triggers append to. +pub(crate) const CHANGELOG: &str = "_iii_row_changes"; + +/// How long the watcher sleeps when no fs event arrives. Pure insurance — +/// on a filesystem where the watch misses events (NFS, some overlayfs), +/// capture latency degrades to this instead of failing. +const FALLBACK_TICK: Duration = Duration::from_secs(2); + +/// The database file behind a `sqlite:` url, or None for `:memory:` forms +/// (which config validation already rejects for native capture). +pub(crate) fn sqlite_file_path(url: &str) -> Option { + let path = url.strip_prefix("sqlite:").unwrap_or(url); + let path = path.strip_prefix("file:").unwrap_or(path); + let path = path.split('?').next().unwrap_or(path); + if path.is_empty() || path.contains(":memory:") { + return None; + } + Some(PathBuf::from(path)) +} + +/// DDL for one captured table: the shared changelog plus three row-level +/// triggers (SQLite has no statement-level triggers or transition tables). +/// Idempotent to reinstall; trigger names embed the table because SQLite +/// trigger names are schema-global, not per-table like postgres. +/// +/// Two properties matter here: +/// * The whole script runs inside ONE explicit transaction. `execute_batch` +/// autocommits per statement, so without it a reinstall (second binding on +/// the same table) would open a window with the triggers dropped — an +/// external write landing there would be lost silently. +/// * The trigger-name suffix and the recorded `tbl` value are lowercased. +/// SQLite resolves identifiers case-insensitively, so bindings spelled in +/// different cases must converge on the SAME trigger set and the same +/// changelog spelling — never a second set double-logging every write. +pub(crate) fn install_sql(table: &str) -> Result { + let target = quote_table(table)?; + let spelling = table.trim().to_lowercase(); + let name = |suffix: &str| { + format!( + "\"iii_row_changed_{suffix}_{}\"", + spelling.replace('"', "\"\"") + ) + }; + let ins = name("ins"); + let upd = name("upd"); + let del = name("del"); + Ok(format!( + r#"BEGIN IMMEDIATE; +CREATE TABLE IF NOT EXISTS {CHANGELOG} ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + tbl TEXT NOT NULL, + op TEXT NOT NULL +); +DROP TRIGGER IF EXISTS {ins}; +CREATE TRIGGER {ins} AFTER INSERT ON {target} FOR EACH ROW +BEGIN INSERT INTO {CHANGELOG} (tbl, op) VALUES ('{tbl}', 'insert'); END; +DROP TRIGGER IF EXISTS {upd}; +CREATE TRIGGER {upd} AFTER UPDATE ON {target} FOR EACH ROW +BEGIN INSERT INTO {CHANGELOG} (tbl, op) VALUES ('{tbl}', 'update'); END; +DROP TRIGGER IF EXISTS {del}; +CREATE TRIGGER {del} AFTER DELETE ON {target} FOR EACH ROW +BEGIN INSERT INTO {CHANGELOG} (tbl, op) VALUES ('{tbl}', 'delete'); END; +COMMIT; +"#, + tbl = spelling.replace('\'', "''"), + )) +} + +/// One coalesced run of changelog rows: (table, op, row count). +type Run = (String, Op, u64); + +/// Collapse per-row changelog entries into per-run events: a 1000-row UPDATE +/// is one event with `affected_rows: 1000`, not a thousand events. Order is +/// preserved; only adjacent same-(table, op) rows merge. +pub(crate) fn coalesce(rows: Vec<(String, Op)>) -> Vec { + let mut out: Vec = Vec::new(); + for (tbl, op) in rows { + match out.last_mut() { + Some((last_tbl, last_op, n)) if *last_tbl == tbl && *last_op == op => *n += 1, + _ => out.push((tbl, op, 1)), + } + } + out +} + +fn parse_op(s: &str) -> Option { + match s { + "insert" => Some(Op::Insert), + "update" => Some(Op::Update), + "delete" => Some(Op::Delete), + _ => None, + } +} + +/// Read everything past `cursor`, in id order. Returns the coalesced runs +/// and the new cursor. A missing changelog table (no binding installed DDL +/// yet) is an empty result, not an error. +fn drain(conn: &Connection, cursor: i64) -> rusqlite::Result<(Vec, i64)> { + let mut stmt = match conn.prepare(&format!( + "SELECT id, tbl, op FROM {CHANGELOG} WHERE id > ?1 ORDER BY id" + )) { + Ok(stmt) => stmt, + Err(e) if e.to_string().contains("no such table") => return Ok((Vec::new(), cursor)), + Err(e) => return Err(e), + }; + let mut rows = stmt.query([cursor])?; + let mut latest = cursor; + let mut raw: Vec<(String, Op)> = Vec::new(); + while let Some(row) = rows.next()? { + latest = row.get(0)?; + let tbl: String = row.get(1)?; + let op: String = row.get(2)?; + // Unknown ops (a future schema writing richer rows) are skipped, not + // fatal — the cursor still advances past them. + if let Some(op) = parse_op(&op) { + raw.push((tbl, op)); + } + } + Ok((coalesce(raw), latest)) +} + +/// The watcher thread body: one dedicated connection (the pool cannot serve +/// this — `data_version` is per-connection), an fs watch for wake-up, drain +/// on every wake. Each event goes to `on_event`; returning `false` from it +/// ends the watcher (the bus side is gone). Returns when `stop` is set — +/// promptly, because the spawner holds the `wake` sender and pokes it after +/// setting the flag; without that poke a stopped watcher would linger for up +/// to [`FALLBACK_TICK`], overlapping its replacement on the same changelog. +pub(crate) fn run_watcher( + db_name: &str, + path: &Path, + stop: &AtomicBool, + wake_tx: mpsc::Sender<()>, + wake_rx: &mpsc::Receiver<()>, + mut on_event: impl FnMut(RowChangedEvent) -> bool, +) { + let conn = match Connection::open(path) { + Ok(c) => c, + Err(e) => { + tracing::warn!(db = %db_name, error = %e, "sqlite watcher could not open database"); + return; + } + }; + let _ = conn.busy_timeout(Duration::from_secs(5)); + // The changelog may not exist yet (first binding not registered); create + // it here too so MAX(id) and data_version have something to run against. + if let Err(e) = conn.execute_batch(&format!( + "CREATE TABLE IF NOT EXISTS {CHANGELOG} (id INTEGER PRIMARY KEY AUTOINCREMENT, tbl TEXT NOT NULL, op TEXT NOT NULL)" + )) { + tracing::warn!(db = %db_name, error = %e, "sqlite watcher could not ensure changelog"); + return; + } + + // Skip history: only changes committed from now on are announced. + let mut cursor: i64 = conn + .query_row( + &format!("SELECT COALESCE(MAX(id), 0) FROM {CHANGELOG}"), + [], + |r| r.get(0), + ) + .unwrap_or(0); + + // Watch the parent directory, filtered to this database's files — the + // -wal/-journal siblings appear and disappear (checkpoints), so watching + // the paths themselves would race their recreation. + let file_prefix = path.file_name().map(|n| n.to_string_lossy().into_owned()); + let mut watcher = { + let wake = wake_tx.clone(); + let prefix = file_prefix.clone(); + notify::recommended_watcher(move |res: Result| { + let Ok(event) = res else { return }; + let relevant = event.paths.iter().any(|p| match (&prefix, p.file_name()) { + (Some(prefix), Some(name)) => name.to_string_lossy().starts_with(prefix.as_str()), + _ => true, + }); + if relevant { + let _ = wake.send(()); + } + }) + .ok() + }; + let watch_dir = path.parent().filter(|p| !p.as_os_str().is_empty()); + let watching = match (&mut watcher, watch_dir) { + (Some(w), Some(dir)) => w.watch(dir, RecursiveMode::NonRecursive).is_ok(), + (Some(w), None) => w.watch(Path::new("."), RecursiveMode::NonRecursive).is_ok(), + _ => false, + }; + if !watching { + tracing::warn!( + db = %db_name, + "sqlite watcher running without fs events; falling back to {}s polling", + FALLBACK_TICK.as_secs() + ); + } + tracing::info!(db = %db_name, path = %path.display(), fs_events = watching, "native capture watching"); + + let mut data_version: i64 = pragma_data_version(&conn).unwrap_or(0); + let mut first = true; + while !stop.load(Ordering::Relaxed) { + if !first { + // Block until something happens (or the fallback tick), then + // collapse any burst of fs events into one drain pass. + let _ = wake_rx.recv_timeout(FALLBACK_TICK); + while wake_rx.try_recv().is_ok() {} + } + first = false; + if stop.load(Ordering::Relaxed) { + break; + } + + // `data_version` changes iff some OTHER connection committed — + // exactly the writes this watcher exists to see. Unchanged → the fs + // event was noise (reads, -shm traffic) and the drain is skipped. + let version = match pragma_data_version(&conn) { + Ok(v) => v, + Err(e) => { + tracing::warn!(db = %db_name, error = %e, "sqlite watcher data_version failed"); + continue; + } + }; + if version == data_version { + continue; + } + data_version = version; + + match drain(&conn, cursor) { + Ok((runs, new_cursor)) => { + for (tbl, op, n) in runs { + let event = RowChangedEvent { + db: db_name.to_string(), + table: Some(tbl), + op, + affected_rows: n, + returning: None, + at: now_ms(), + }; + if !on_event(event) { + return; // bus side gone — shutting down + } + } + if new_cursor != cursor { + cursor = new_cursor; + // ponytail: GC assumes this worker is the only watcher of + // this file; two workers watching one db would starve each + // other. Per-watcher cursor rows if that ever exists. + let _ = + conn.execute(&format!("DELETE FROM {CHANGELOG} WHERE id <= ?1"), [cursor]); + } + } + Err(e) => { + tracing::warn!(db = %db_name, error = %e, "sqlite watcher drain failed"); + } + } + } +} + +fn pragma_data_version(conn: &Connection) -> rusqlite::Result { + conn.query_row("PRAGMA data_version", [], |r| r.get(0)) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + + #[test] + fn sqlite_file_path_strips_schemes_and_rejects_memory() { + assert_eq!( + sqlite_file_path("sqlite:./data/iii.db"), + Some(PathBuf::from("./data/iii.db")) + ); + assert_eq!( + sqlite_file_path("sqlite:file:./x.db?mode=rwc"), + Some(PathBuf::from("./x.db")) + ); + assert_eq!(sqlite_file_path("sqlite::memory:"), None); + assert_eq!(sqlite_file_path("sqlite:file::memory:?cache=shared"), None); + } + + #[test] + fn install_sql_quotes_and_embeds_table_names() { + let sql = install_sql("orders").unwrap(); + assert!(sql.contains("AFTER INSERT ON \"orders\"")); + assert!(sql.contains("\"iii_row_changed_del_orders\"")); + assert!(sql.contains("VALUES ('orders', 'update')")); + // Reinstall must be atomic: execute_batch autocommits per statement, + // so the script carries its own transaction. + assert!(sql.starts_with("BEGIN IMMEDIATE;")); + assert!(sql.trim_end().ends_with("COMMIT;")); + assert!(install_sql(" ").is_err()); + + // Hostile names stay inside identifier quotes and string literals + // (the recorded spelling is lowercased along with everything else). + let evil = "t'; DROP TABLE x; --"; + let sql = install_sql(evil).unwrap(); + assert!(sql.contains("VALUES ('t''; drop table x; --', 'insert')")); + } + + /// Bindings spelled in different cases must converge on ONE trigger set + /// writing ONE changelog row per change — never a second set that + /// double-logs every write. (SQLite resolves identifiers + /// case-insensitively, and install_sql normalizes the embedded spelling; + /// this pins both halves against a real database.) + #[test] + fn differently_cased_reinstall_never_double_logs() { + let conn = Connection::open_in_memory().unwrap(); + conn.execute_batch("CREATE TABLE items (id INTEGER PRIMARY KEY, n INT)") + .unwrap(); + conn.execute_batch(&install_sql("items").unwrap()).unwrap(); + conn.execute_batch(&install_sql("ITEMS").unwrap()).unwrap(); + + let triggers: i64 = conn + .query_row( + "SELECT count(*) FROM sqlite_master WHERE type = 'trigger'", + [], + |r| r.get(0), + ) + .unwrap(); + assert_eq!(triggers, 3, "reinstall must replace, not accumulate"); + + conn.execute_batch("INSERT INTO items (n) VALUES (1)") + .unwrap(); + let rows: Vec<(String, String)> = conn + .prepare(&format!("SELECT tbl, op FROM {CHANGELOG}")) + .unwrap() + .query_map([], |r| Ok((r.get(0)?, r.get(1)?))) + .unwrap() + .collect::>() + .unwrap(); + assert_eq!(rows, vec![("items".to_string(), "insert".to_string())]); + } + + #[test] + fn coalesce_merges_adjacent_runs_only() { + let rows = vec![ + ("a".to_string(), Op::Insert), + ("a".to_string(), Op::Insert), + ("a".to_string(), Op::Update), + ("b".to_string(), Op::Update), + ("a".to_string(), Op::Insert), + ]; + assert_eq!( + coalesce(rows), + vec![ + ("a".to_string(), Op::Insert, 2), + ("a".to_string(), Op::Update, 1), + ("b".to_string(), Op::Update, 1), + ("a".to_string(), Op::Insert, 1), + ] + ); + assert!(coalesce(Vec::new()).is_empty()); + } + + /// The cross-process claim, sqlite edition: a write on a completely + /// separate connection (stand-in for another process) reaches the + /// watcher through triggers + changelog + fs wake-up. + #[test] + fn watcher_hears_writes_from_another_connection() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("watched.db"); + + // "External" client: creates the table and installs capture DDL the + // way the handler does, then writes. + let external = Connection::open(&db_path).unwrap(); + external + .execute_batch("CREATE TABLE items (id INTEGER PRIMARY KEY, n INT)") + .unwrap(); + external + .execute_batch(&install_sql("items").unwrap()) + .unwrap(); + + let stop = Arc::new(AtomicBool::new(false)); + let (tx, rx) = mpsc::channel(); + let (wake_tx, wake_rx) = mpsc::channel(); + let stop_wake = wake_tx.clone(); + let thread = { + let stop = Arc::clone(&stop); + let path = db_path.clone(); + std::thread::spawn(move || { + run_watcher("primary", &path, &stop, wake_tx, &wake_rx, move |ev| { + tx.send(ev).is_ok() + }) + }) + }; + + // Wait for the watcher to establish its baseline (it logs first, + // then reads MAX(id)); a short settle keeps the test deterministic + // without exposing internals. + std::thread::sleep(Duration::from_millis(300)); + + external + .execute_batch( + "INSERT INTO items (n) VALUES (1), (2), (3); \ + UPDATE items SET n = n + 1; \ + DELETE FROM items WHERE n > 2;", + ) + .unwrap(); + + let mut events = Vec::new(); + while events.len() < 3 { + events.push( + rx.recv_timeout(Duration::from_secs(10)) + .expect("watcher event within 10s"), + ); + } + assert_eq!(events[0].op, Op::Insert); + assert_eq!(events[0].affected_rows, 3); + assert_eq!(events[0].table.as_deref(), Some("items")); + assert_eq!(events[0].db, "primary"); + assert_eq!(events[1].op, Op::Update); + assert_eq!(events[1].affected_rows, 3); + assert_eq!(events[2].op, Op::Delete); + assert_eq!(events[2].affected_rows, 2); + + // A rolled-back write is invisible: the changelog rows die with it. + external + .execute_batch("BEGIN; INSERT INTO items (n) VALUES (9); ROLLBACK;") + .unwrap(); + // And a zero-row statement appends nothing. + external + .execute_batch("UPDATE items SET n = 0 WHERE n = -777") + .unwrap(); + assert!( + rx.recv_timeout(Duration::from_millis(600)).is_err(), + "rolled-back / zero-row writes must not produce events" + ); + + stop.store(true, Ordering::Relaxed); + let _ = stop_wake.send(()); + thread.join().unwrap(); + } + + /// History from before the watcher started is skipped (at-most-once, + /// postgres parity) — and the GC keeps the changelog from growing. + #[test] + fn watcher_skips_history_and_gcs_the_changelog() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("watched.db"); + let external = Connection::open(&db_path).unwrap(); + external + .execute_batch("CREATE TABLE items (id INTEGER PRIMARY KEY, n INT)") + .unwrap(); + external + .execute_batch(&install_sql("items").unwrap()) + .unwrap(); + // Rows written before any watcher exists. + external + .execute_batch("INSERT INTO items (n) VALUES (1), (2)") + .unwrap(); + + let stop = Arc::new(AtomicBool::new(false)); + let (tx, rx) = mpsc::channel(); + let (wake_tx, wake_rx) = mpsc::channel(); + let stop_wake = wake_tx.clone(); + let thread = { + let stop = Arc::clone(&stop); + let path = db_path.clone(); + std::thread::spawn(move || { + run_watcher("primary", &path, &stop, wake_tx, &wake_rx, move |ev| { + tx.send(ev).is_ok() + }) + }) + }; + std::thread::sleep(Duration::from_millis(300)); + + // Nothing replayed… + assert!(rx.recv_timeout(Duration::from_millis(400)).is_err()); + // …but a new write arrives, and afterwards the changelog is drained. + external.execute_batch("DELETE FROM items").unwrap(); + let ev = rx.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!(ev.op, Op::Delete); + assert_eq!(ev.affected_rows, 2); + + // GC happened: nothing at or below the cursor survives. Retry + // briefly — the DELETE runs just after the event is sent. + let deadline = std::time::Instant::now() + Duration::from_secs(5); + loop { + let left: i64 = external + .query_row(&format!("SELECT count(*) FROM {CHANGELOG}"), [], |r| { + r.get(0) + }) + .unwrap(); + if left == 0 { + break; + } + assert!( + std::time::Instant::now() < deadline, + "changelog was not GC'd, {left} rows left" + ); + std::thread::sleep(Duration::from_millis(50)); + } + + stop.store(true, Ordering::Relaxed); + let _ = stop_wake.send(()); + thread.join().unwrap(); + } +} diff --git a/database/src/ui.rs b/database/src/ui.rs index 20bf1d089..69e7a205e 100644 --- a/database/src/ui.rs +++ b/database/src/ui.rs @@ -66,6 +66,16 @@ mod tests { assert!(PAGE_JS.contains("export"), "built page.js looks wrong"); } + #[test] + fn embedded_page_registers_the_config_form() { + // The configuration form ships inside page.js; a build that lost it + // silently reverts the Workers tab to the generic schema editor. + assert!( + PAGE_JS.contains("configForms"), + "built page.js no longer registers the configuration form" + ); + } + #[test] fn embedded_styles_are_scoped() { // esbuild prints the attribute selector unquoted ([data-iii-ui=database]). diff --git a/database/tests/e2e/README.md b/database/tests/e2e/README.md index c3e8534e0..ccf6fe6ef 100644 --- a/database/tests/e2e/README.md +++ b/database/tests/e2e/README.md @@ -115,6 +115,8 @@ accepted; outside-tx COUNT=1`). | `workers/harness/` | TypeScript smoke-test worker (runs as a host process) | | `workers/harness/src/cases-interactive-tx.ts` | Interactive-transaction lifecycle cases | | `workers/harness/src/cases-row-changed.ts` | Row-change trigger delivery cases | +| `workers/harness/src/cases-native-capture.ts` | `capture: native` cases for postgres (LISTEN/NOTIFY, `pg_native_db`), sqlite (changelog + fs watch, `sqlite_native_db`), and mysql (binlog stream, `mysql_native_db`) — cross-client delivery, no double-fire on own writes, table-less binding rejection, commit/rollback gating through interactive transactions, multi-subscriber fan-out with ops filters across trigger reinstall, bulk-statement coalescing (100 rows = 1 event) | +| `mysql-init/grant-replication.sql` | Replication grants for the `iii` mysql user (binlog capture streams as a replica). Applied on first volume init only — on an older volume run `docker compose down -v` once | | `workers/harness/src/cases-tx-control-bypass.ts` | Side-channel-finalization repros | | `reports/report.json` | Per-case results (latest run) | diff --git a/database/tests/e2e/docker-compose.yml b/database/tests/e2e/docker-compose.yml index c8aff852d..bd4fb604b 100644 --- a/database/tests/e2e/docker-compose.yml +++ b/database/tests/e2e/docker-compose.yml @@ -26,6 +26,9 @@ services: - "53306:3306" volumes: - mysql_data:/var/lib/mysql + # Native capture streams the binlog as a replica; the app user needs + # the global replication grants. Runs once on first volume init. + - ./mysql-init:/docker-entrypoint-initdb.d:ro healthcheck: test: ["CMD", "mysqladmin", "ping", "-h", "localhost", "-uiii", "-piii"] interval: 2s diff --git a/database/tests/e2e/mysql-init/grant-replication.sql b/database/tests/e2e/mysql-init/grant-replication.sql new file mode 100644 index 000000000..96686e90f --- /dev/null +++ b/database/tests/e2e/mysql-init/grant-replication.sql @@ -0,0 +1,5 @@ +-- Native change capture (`capture: native`) reads the binlog as a replica. +-- REPLICATION CLIENT: SHOW BINARY LOG STATUS (position snapshot + the +-- registration-time privilege probe). REPLICATION SLAVE: COM_BINLOG_DUMP. +GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'iii'@'%'; +FLUSH PRIVILEGES; diff --git a/database/tests/e2e/workers/harness/fixtures/database.schema.json b/database/tests/e2e/workers/harness/fixtures/database.schema.json index f223dbaf2..3ab99ff2c 100644 --- a/database/tests/e2e/workers/harness/fixtures/database.schema.json +++ b/database/tests/e2e/workers/harness/fixtures/database.schema.json @@ -1,8 +1,35 @@ { "definitions": { + "CaptureMode": { + "description": "How row-change events are captured for one database.", + "oneOf": [ + { + "description": "Classify the SQL this worker executes. Writes from other clients are invisible. Works on every driver. The default.", + "enum": [ + "statements" + ], + "type": "string" + }, + { + "description": "Capture writes from any client, including other processes. Table-scoped bindings only. Postgres: triggers + LISTEN/NOTIFY on a dedicated connection (needs DDL rights). File-backed sqlite: triggers + changelog table + filesystem watch. MySQL: the binlog replication stream (needs replication grants, nothing installed).", + "enum": [ + "native" + ], + "type": "string" + } + ] + }, "DatabaseConfig": { "description": "Per-database connection settings. The URL scheme selects the driver; `pool` and `tls` are optional and default when omitted.", "properties": { + "capture": { + "allOf": [ + { + "$ref": "#/definitions/CaptureMode" + } + ], + "description": "How `database::row-changed` events are captured for this database. `statements` (default) classifies the SQL this worker executes; `native` makes writes from ANY client — psql, other processes — fire too. Postgres: triggers + LISTEN/NOTIFY. File-backed sqlite: triggers + changelog drained on filesystem wake-up. MySQL: the binlog replication stream (needs REPLICATION SLAVE + REPLICATION CLIENT)." + }, "pool": { "allOf": [ { diff --git a/database/tests/e2e/workers/harness/src/cases-native-capture.ts b/database/tests/e2e/workers/harness/src/cases-native-capture.ts new file mode 100644 index 000000000..7aa593b39 --- /dev/null +++ b/database/tests/e2e/workers/harness/src/cases-native-capture.ts @@ -0,0 +1,956 @@ +/** + * Native change capture (`capture: native` — postgres, file-backed sqlite, + * and mysql). + * + * Each native handle points at the same physical database as its + * statements-path sibling (`pg_native_db` ↔ `pg_db`, `sqlite_native_db` ↔ + * `sqlite_db`, `mysql_native_db` ↔ `mysql_db`). A write that enters through + * the sibling's pool is — from the native handle's perspective — an + * external client: different pool, different connections, invisible to SQL + * classification. If the native subscriber still hears it, the cross-client + * claim holds end-to-end (postgres: DDL triggers → pg_notify → dedicated + * LISTEN connection; sqlite: DDL triggers → changelog table → fs-watch + * drain; mysql: the binlog replication stream, nothing installed). + * + * Cases carry `applies` for the sibling driver so they run once inside that + * driver's loop, and address the native handle explicitly in payloads. + */ + +import type { DriverKey } from './dialect.ts' +import type { TestCase } from './cases.ts' +import { expect, expectEqual } from './cases.ts' + +const EVENT_TIMEOUT_MS = 5_000 +const SILENCE_WINDOW_MS = 500 + +interface RowChangedEvent { + db: string + table: string | null + op: 'insert' | 'update' | 'delete' | 'other' + affected_rows: number + returning?: Record[] + at: number +} + +/** Everything that differs between the native targets. */ +interface NativeTarget { + /** Driver loop that hosts these cases (the statements-path sibling). */ + applies: DriverKey + /** The `capture: native` handle. */ + nativeDb: string + idColumnDDL: string + ph: (i: number) => string + /** How the database reports the table in events (pg schema-qualifies). */ + eventTable: (table: string) => string + /** + * Catalog probe returning the number of installed capture triggers. + * Absent for binlog capture (mysql), which installs nothing — readiness + * is proven by the warmup write loop instead. + */ + triggerCountSql?: (table: string) => { sql: string; params: unknown[] } + /** + * Whether a native binding spelled in the WRONG case still captures. + * pg: no — the table name is quoted verbatim into DDL and quoted + * postgres identifiers are case-sensitive, so install fails loudly. + * sqlite: yes — quoted identifiers match case-insensitively. + * mysql: yes — no DDL at all; the bus filter matches case-insensitively. + */ + uppercaseBindingCaptures: boolean + /** + * Whether registering against a table that does not exist is ACCEPTED. + * pg/sqlite reject at DDL install; mysql has nothing to install, so the + * binding is accepted and starts capturing when the table appears. + */ + missingTableBindingAccepted: boolean +} + +const TARGETS: NativeTarget[] = [ + { + applies: 'pg_db', + nativeDb: 'pg_native_db', + idColumnDDL: 'BIGSERIAL PRIMARY KEY', + ph: (i) => `$${i}`, + eventTable: (table) => `public.${table}`, + // `$1::text::regclass`, not `$1::regclass` — a bare regclass cast makes + // the driver bind the parameter AS regclass (22P03); text binds cleanly + // and the server does the regclass conversion. + triggerCountSql: (table) => ({ + sql: `SELECT count(*) AS n FROM pg_trigger WHERE tgrelid = $1::text::regclass AND tgname LIKE 'iii_row_changed_%'`, + params: [table], + }), + uppercaseBindingCaptures: false, + missingTableBindingAccepted: false, + }, + { + applies: 'sqlite_db', + nativeDb: 'sqlite_native_db', + idColumnDDL: 'INTEGER PRIMARY KEY AUTOINCREMENT', + ph: () => '?', + eventTable: (table) => table, + // lower() on both sides: a reinstall from a differently-cased binding + // stores tbl_name with THAT spelling, and sqlite's `=` is case-sensitive. + triggerCountSql: (table) => ({ + sql: `SELECT count(*) AS n FROM sqlite_master WHERE type = 'trigger' AND lower(tbl_name) = lower(?1) AND name LIKE 'iii_row_changed_%'`, + params: [table], + }), + uppercaseBindingCaptures: true, + missingTableBindingAccepted: false, + }, + { + applies: 'mysql_db', + nativeDb: 'mysql_native_db', + idColumnDDL: 'BIGINT AUTO_INCREMENT PRIMARY KEY', + ph: () => '?', + eventTable: (table) => table, + // Binlog capture installs nothing to probe for. + uppercaseBindingCaptures: true, + missingTableBindingAccepted: true, + }, +] + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) + +function sink(events: RowChangedEvent[], label: string) { + let cursor = 0 + return { + async next(): Promise { + const deadline = Date.now() + EVENT_TIMEOUT_MS + while (events.length <= cursor && Date.now() < deadline) await sleep(20) + if (events.length <= cursor) throw new Error(`${label}: event ${cursor + 1} was not delivered`) + return events[cursor++] + }, + expectDrained(): void { + expectEqual(events.length, cursor, `${label}: unexpected extra event`) + }, + } +} + +/** + * Wait until native capture is actually delivering for this target. + * Registration acks race the first write otherwise — and delivery is + * at-most-once, so a racing write is silently unheard. + * + * Trigger-based targets (pg, sqlite): poll the catalog through the worker + * until the three capture triggers exist. Binlog capture (mysql) installs + * nothing to probe, so prove the stream is attached empirically: a warmup + * table with its own binding is poked until an event comes back. + */ +async function waitForCaptureReady( + call: (functionId: string, payload: unknown) => Promise, + iii: any, + target: NativeTarget, + table: string, +): Promise { + if (target.triggerCountSql) { + const probe = target.triggerCountSql(table) + const deadline = Date.now() + EVENT_TIMEOUT_MS + for (;;) { + const r = await call('database::query', { db: target.applies, ...probe }) + if (Number(r.rows?.[0]?.n) === 3) return + if (Date.now() > deadline) { + throw new Error(`capture triggers for ${table} were not installed within ${EVENT_TIMEOUT_MS}ms`) + } + await sleep(50) + } + } + + const warmupTable = `e2e_native_warmup_${target.applies}` + const fnId = `harness::native_warmup_${target.applies}` + const events: RowChangedEvent[] = [] + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${warmupTable}` }) + await call('database::execute', { + db: target.nativeDb, + sql: `CREATE TABLE ${warmupTable} (id ${target.idColumnDDL}, n INT NOT NULL)`, + }) + const fnRef = iii.registerFunction( + fnId, + async (payload: RowChangedEvent) => { + events.push(payload) + return null + }, + { description: 'Warmup sink proving the capture stream is attached.' }, + ) + const triggerRef = iii.registerTrigger({ + type: 'database::row-changed', + function_id: fnId, + config: { db: target.nativeDb, table: warmupTable }, + }) + try { + const deadline = Date.now() + 15_000 + while (events.length === 0) { + if (Date.now() > deadline) { + throw new Error(`capture stream for ${target.nativeDb} did not deliver within 15s`) + } + await call('database::execute', { + db: target.nativeDb, + sql: `INSERT INTO ${warmupTable} (n) VALUES (${target.ph(1)})`, + params: [1], + }) + const poked = Date.now() + while (events.length === 0 && Date.now() - poked < 700) await sleep(20) + } + } finally { + triggerRef.unregister() + fnRef.unregister() + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${warmupTable}` }) + } +} + +function crossClientCase(target: NativeTarget): TestCase { + return { + name: 'native capture hears writes from another client, own writes fire once', + applies: [target.applies], + async run({ call, iii }) { + const table = `e2e_native_capture_${target.applies}` + const nativeFnId = `harness::native_capture_events_${target.applies}` + const classifiedFnId = `harness::native_capture_classified_${target.applies}` + const nativeEvents: RowChangedEvent[] = [] + const classifiedEvents: RowChangedEvent[] = [] + const native = sink(nativeEvents, 'native subscriber') + const classified = sink(classifiedEvents, 'classified subscriber') + const ph = target.ph + + // The watched table must exist before the binding registers — the + // worker installs the capture triggers at registration time. + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + await call('database::execute', { + db: target.nativeDb, + sql: `CREATE TABLE ${table} (id ${target.idColumnDDL}, n INT NOT NULL)`, + }) + + const nativeFn = iii.registerFunction( + nativeFnId, + async (payload: RowChangedEvent) => { + nativeEvents.push(payload) + return null + }, + { description: 'Native-capture E2E event sink.' }, + ) + const nativeTrigger = iii.registerTrigger({ + type: 'database::row-changed', + function_id: nativeFnId, + config: { db: target.nativeDb, table }, + }) + // Statements-path subscriber on the sibling handle, watching the SAME + // physical table. Proves the two capture modes coexist and attribute + // correctly. + const classifiedFn = iii.registerFunction( + classifiedFnId, + async (payload: RowChangedEvent) => { + classifiedEvents.push(payload) + return null + }, + { description: 'Statements-path E2E event sink for the native-capture table.' }, + ) + const classifiedTrigger = iii.registerTrigger({ + type: 'database::row-changed', + function_id: classifiedFnId, + config: { db: target.applies, table }, + }) + + const expectNative = (event: RowChangedEvent, op: RowChangedEvent['op'], affectedRows: number): void => { + expectEqual(event.db, target.nativeDb, 'native event db') + expectEqual(event.table, target.eventTable(table), 'native event table') + expectEqual(event.op, op, 'native event op') + expectEqual(event.affected_rows, affectedRows, 'native event affected_rows') + expect(event.returning === undefined, 'native events carry no RETURNING rows') + expect(Number.isFinite(event.at) && event.at > 0, 'native event at is an epoch timestamp') + } + + try { + // Wait until BOTH bindings are visible to the engine before writing. + const registered = await call('engine::registered-triggers::list', {}) + for (const fn of [nativeFnId, classifiedFnId]) { + expect( + registered.registered_triggers.some( + (t: { trigger_type: string; function_id: string }) => + t.trigger_type === 'database::row-changed' && t.function_id === fn, + ), + `trigger registration for ${fn} is visible to the engine`, + ) + } + // …and until the worker's DDL install has landed in the database. + await waitForCaptureReady(call, iii, target, table) + + // 1. External write: enters through the sibling pool. The native + // subscriber must hear it via the database; the classified + // subscriber hears the same write attributed to the sibling. + await call('database::execute', { + db: target.applies, + sql: `INSERT INTO ${table} (n) VALUES (${ph(1)}), (${ph(2)})`, + params: [10, 20], + }) + expectNative(await native.next(), 'insert', 2) + const viaSibling = await classified.next() + expectEqual(viaSibling.db, target.applies, 'classified event db') + expectEqual(viaSibling.op, 'insert', 'classified event op') + + // 2. Own write through the native handle: must fire exactly ONCE, + // never twice — self-writes leave the classification path on a + // native database. + await call('database::execute', { + db: target.nativeDb, + sql: `UPDATE ${table} SET n = n + 1`, + }) + expectNative(await native.next(), 'update', 2) + + // 3. A write that changes no rows fires nothing on either path. + await call('database::execute', { + db: target.applies, + sql: `UPDATE ${table} SET n = ${ph(1)} WHERE n = ${ph(2)}`, + params: [0, -999], + }) + + // 4. Delete via the external client. + await call('database::execute', { + db: target.applies, + sql: `DELETE FROM ${table} WHERE n > ${ph(1)}`, + params: [0], + }) + expectNative(await native.next(), 'delete', 2) + const deleted = await classified.next() + expectEqual(deleted.op, 'delete', 'classified delete op') + + // The zero-row update (step 3) must not have queued anything. + await sleep(SILENCE_WINDOW_MS) + native.expectDrained() + classified.expectDrained() + } finally { + nativeTrigger.unregister() + nativeFn.unregister() + classifiedTrigger.unregister() + classifiedFn.unregister() + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + } + }, + } +} + +function tablelessRejectionCase(target: NativeTarget): TestCase { + return { + name: 'native capture rejects table-less bindings', + applies: [target.applies], + async run({ call, iii }) { + const table = `e2e_native_tableless_${target.applies}` + const sentinelFnId = `harness::native_capture_sentinel_${target.applies}` + const tablelessFnId = `harness::native_capture_tableless_${target.applies}` + const sentinelEvents: RowChangedEvent[] = [] + const tablelessEvents: RowChangedEvent[] = [] + const sentinel = sink(sentinelEvents, 'sentinel subscriber') + + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + await call('database::execute', { + db: target.nativeDb, + sql: `CREATE TABLE ${table} (id ${target.idColumnDDL}, n INT NOT NULL)`, + }) + + // Valid table-scoped binding: installs the triggers and proves events + // DO flow for this table — without it, the table-less binding's + // silence below would be vacuous (no triggers, nothing to hear). + const sentinelFn = iii.registerFunction( + sentinelFnId, + async (payload: RowChangedEvent) => { + sentinelEvents.push(payload) + return null + }, + { description: 'Valid table-scoped sink proving events flow.' }, + ) + const sentinelTrigger = iii.registerTrigger({ + type: 'database::row-changed', + function_id: sentinelFnId, + config: { db: target.nativeDb, table }, + }) + // A db-wide binding is invalid on a native database: per-table + // triggers are what make external writes visible, so the worker must + // refuse it. If it were wrongly accepted, its filter (db, no table) + // would match the sentinel table's events. + const tablelessFn = iii.registerFunction( + tablelessFnId, + async (payload: RowChangedEvent) => { + tablelessEvents.push(payload) + return null + }, + { description: 'Sink that must never receive events (rejected binding).' }, + ) + const tablelessTrigger = iii.registerTrigger({ + type: 'database::row-changed', + function_id: tablelessFnId, + config: { db: target.nativeDb }, + }) + + try { + await waitForCaptureReady(call, iii, target, table) + await sleep(SILENCE_WINDOW_MS) // let the table-less registration settle too + await call('database::execute', { + db: target.nativeDb, + sql: `INSERT INTO ${table} (n) VALUES (${target.ph(1)})`, + params: [1], + }) + const heard = await sentinel.next() + expectEqual(heard.op, 'insert', 'sentinel hears the insert') + await sleep(SILENCE_WINDOW_MS) + expectEqual(tablelessEvents.length, 0, 'rejected table-less binding received an event') + } finally { + tablelessTrigger.unregister() + tablelessFn.unregister() + sentinelTrigger.unregister() + sentinelFn.unregister() + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + } + }, + } +} + +/** + * Commit gating through the real database, not the worker's staging: on a + * native handle the worker's own classification path is off, so an + * interactive transaction's visibility is decided entirely by the capture + * mechanism (pg: NOTIFY is transactional; sqlite: changelog rows ride the + * writer's transaction; mysql: only committed transactions reach the + * binlog). Rollback must be absolute silence; commit must deliver. + */ +function txGatingCase(target: NativeTarget): TestCase { + return { + name: 'native capture is commit-gated through interactive transactions', + applies: [target.applies], + async run({ call, iii }) { + const table = `e2e_native_tx_${target.applies}` + const fnId = `harness::native_tx_${target.applies}` + const events: RowChangedEvent[] = [] + const native = sink(events, 'native tx subscriber') + const ph = target.ph + + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + await call('database::execute', { + db: target.nativeDb, + sql: `CREATE TABLE ${table} (id ${target.idColumnDDL}, n INT NOT NULL)`, + }) + const fnRef = iii.registerFunction( + fnId, + async (payload: RowChangedEvent) => { + events.push(payload) + return null + }, + { description: 'Native-capture transaction-gating E2E sink.' }, + ) + const triggerRef = iii.registerTrigger({ + type: 'database::row-changed', + function_id: fnId, + config: { db: target.nativeDb, table }, + }) + + let activeTransaction: string | undefined + try { + await waitForCaptureReady(call, iii, target, table) + + // 1. Uncommitted writes are invisible… + activeTransaction = ( + await call('database::beginTransaction', { db: target.nativeDb }) + ).transaction.id + await call('database::transactionExecute', { + transaction_id: activeTransaction, + sql: `INSERT INTO ${table} (n) VALUES (${ph(1)})`, + params: [1], + }) + await sleep(SILENCE_WINDOW_MS) + native.expectDrained() + + // …and a rollback erases them for good. + await call('database::rollbackTransaction', { transaction_id: activeTransaction }) + activeTransaction = undefined + await sleep(SILENCE_WINDOW_MS) + native.expectDrained() + + // 2. A committed transaction delivers — once, after the commit. + activeTransaction = ( + await call('database::beginTransaction', { db: target.nativeDb }) + ).transaction.id + await call('database::transactionExecute', { + transaction_id: activeTransaction, + sql: `INSERT INTO ${table} (n) VALUES (${ph(1)})`, + params: [2], + }) + await call('database::transactionExecute', { + transaction_id: activeTransaction, + sql: `UPDATE ${table} SET n = n + 1`, + }) + await sleep(SILENCE_WINDOW_MS) + native.expectDrained() + await call('database::commitTransaction', { transaction_id: activeTransaction }) + activeTransaction = undefined + + // Exactly two events, one per statement — but back-to-back Void + // dispatches carry no cross-event ordering guarantee, so assert the + // set, not the sequence. + const committed = [await native.next(), await native.next()] + const ops = committed.map((e) => e.op).sort() + expectEqual(ops, ['insert', 'update'], 'committed transaction delivers both events') + for (const event of committed) { + expectEqual(event.affected_rows, 1, `committed ${event.op} affected_rows`) + } + await sleep(SILENCE_WINDOW_MS) + native.expectDrained() + } finally { + if (activeTransaction) { + try { + await call('database::rollbackTransaction', { transaction_id: activeTransaction }) + } catch { + /* transaction may already be finalized */ + } + } + triggerRef.unregister() + fnRef.unregister() + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + } + }, + } +} + +/** + * Fan-out and filtering on the native path: an all-ops subscriber and a + * delete-only subscriber share one table. Registering the second binding + * REINSTALLS the capture DDL (pg/sqlite) — the first subscriber must keep + * hearing through it. After both unregister, external writes still succeed + * (orphaned triggers are inert, not broken) and deliver to no one. + */ +function fanOutOpsCase(target: NativeTarget): TestCase { + return { + name: 'native capture fans out, filters ops, and survives trigger reinstall', + applies: [target.applies], + async run({ call, iii }) { + const table = `e2e_native_fanout_${target.applies}` + const allFnId = `harness::native_fanout_all_${target.applies}` + const deletesFnId = `harness::native_fanout_deletes_${target.applies}` + const allEvents: RowChangedEvent[] = [] + const deleteEvents: RowChangedEvent[] = [] + const all = sink(allEvents, 'all-ops subscriber') + const deletes = sink(deleteEvents, 'delete-only subscriber') + const ph = target.ph + + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + await call('database::execute', { + db: target.nativeDb, + sql: `CREATE TABLE ${table} (id ${target.idColumnDDL}, n INT NOT NULL)`, + }) + const allFn = iii.registerFunction( + allFnId, + async (payload: RowChangedEvent) => { + allEvents.push(payload) + return null + }, + { description: 'Native-capture fan-out E2E sink (all ops).' }, + ) + const allTrigger = iii.registerTrigger({ + type: 'database::row-changed', + function_id: allFnId, + config: { db: target.nativeDb, table }, + }) + // Second binding on the SAME table: worker-side this re-runs the DDL + // install (DROP + CREATE trigger) while the first binding is live. + const deletesFn = iii.registerFunction( + deletesFnId, + async (payload: RowChangedEvent) => { + deleteEvents.push(payload) + return null + }, + { description: 'Native-capture fan-out E2E sink (deletes only).' }, + ) + const deletesTrigger = iii.registerTrigger({ + type: 'database::row-changed', + function_id: deletesFnId, + config: { db: target.nativeDb, table, ops: ['delete'] }, + }) + + let bindingsLive = true + const unregisterBindings = () => { + if (!bindingsLive) return + bindingsLive = false + allTrigger.unregister() + deletesTrigger.unregister() + } + + try { + const registered = await call('engine::registered-triggers::list', {}) + for (const fn of [allFnId, deletesFnId]) { + expect( + registered.registered_triggers.some( + (t: { trigger_type: string; function_id: string }) => + t.trigger_type === 'database::row-changed' && t.function_id === fn, + ), + `trigger registration for ${fn} is visible to the engine`, + ) + } + await waitForCaptureReady(call, iii, target, table) + // The second registration's reinstall races the engine ack; give the + // worker a beat so no write lands mid DROP/CREATE. + await sleep(300) + + // Insert (external client): all-ops hears, delete-only does not. + await call('database::execute', { + db: target.applies, + sql: `INSERT INTO ${table} (n) VALUES (${ph(1)})`, + params: [10], + }) + expectEqual((await all.next()).op, 'insert', 'all-ops subscriber hears insert') + await sleep(SILENCE_WINDOW_MS) + deletes.expectDrained() + + // Delete: both hear exactly one event. + await call('database::execute', { + db: target.applies, + sql: `DELETE FROM ${table} WHERE n = ${ph(1)}`, + params: [10], + }) + expectEqual((await all.next()).op, 'delete', 'all-ops subscriber hears delete') + expectEqual((await deletes.next()).op, 'delete', 'delete-only subscriber hears delete') + + // Unregister both; external writes still succeed and nobody hears. + unregisterBindings() + await sleep(SILENCE_WINDOW_MS) + const r = await call('database::execute', { + db: target.applies, + sql: `INSERT INTO ${table} (n) VALUES (${ph(1)})`, + params: [20], + }) + expectEqual(r.affected_rows, 1, 'write succeeds after unregister (orphan capture is inert)') + await sleep(SILENCE_WINDOW_MS) + all.expectDrained() + deletes.expectDrained() + } finally { + unregisterBindings() + allFn.unregister() + deletesFn.unregister() + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + } + }, + } +} + +/** + * Bulk statements stay single events with true counts: 100 rows inserted, + * updated, deleted must arrive as exactly three events with + * affected_rows=100 — never one event per row. Each driver earns this a + * different way (pg statement-level triggers with transition tables, + * sqlite run-length coalescing of changelog rows, mysql merging of chunked + * binlog row events), so proving it end-to-end covers all three coalescers. + */ +function bulkCoalescingCase(target: NativeTarget): TestCase { + return { + name: 'native capture coalesces bulk statements into single events', + applies: [target.applies], + async run({ call, iii }) { + const table = `e2e_native_bulk_${target.applies}` + const fnId = `harness::native_bulk_${target.applies}` + const events: RowChangedEvent[] = [] + const native = sink(events, 'bulk subscriber') + const ROWS = 100 + + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + await call('database::execute', { + db: target.nativeDb, + sql: `CREATE TABLE ${table} (id ${target.idColumnDDL}, n INT NOT NULL)`, + }) + const fnRef = iii.registerFunction( + fnId, + async (payload: RowChangedEvent) => { + events.push(payload) + return null + }, + { description: 'Native-capture bulk-coalescing E2E sink.' }, + ) + const triggerRef = iii.registerTrigger({ + type: 'database::row-changed', + function_id: fnId, + config: { db: target.nativeDb, table }, + }) + + try { + await waitForCaptureReady(call, iii, target, table) + + const values = Array.from({ length: ROWS }, (_, i) => `(${i})`).join(', ') + await call('database::execute', { + db: target.applies, + sql: `INSERT INTO ${table} (n) VALUES ${values}`, + }) + const inserted = await native.next() + expectEqual(inserted.op, 'insert', 'bulk insert op') + expectEqual(inserted.affected_rows, ROWS, 'bulk insert arrives as ONE event') + + await call('database::execute', { + db: target.applies, + sql: `UPDATE ${table} SET n = n + 1`, + }) + const updated = await native.next() + expectEqual(updated.op, 'update', 'bulk update op') + expectEqual(updated.affected_rows, ROWS, 'bulk update arrives as ONE event') + + await call('database::execute', { db: target.applies, sql: `DELETE FROM ${table}` }) + const deleted = await native.next() + expectEqual(deleted.op, 'delete', 'bulk delete op') + expectEqual(deleted.affected_rows, ROWS, 'bulk delete arrives as ONE event') + + // Exactly three events total — a per-row implementation would have + // flooded 300. + await sleep(SILENCE_WINDOW_MS) + native.expectDrained() + } finally { + triggerRef.unregister() + fnRef.unregister() + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + } + }, + } +} + +/** + * Table-name casing on the NATIVE path is driver-specific, and the + * difference is deliberate, documented behavior — pin it: an + * uppercase-spelled binding on a lowercase table captures on sqlite + * (case-insensitive quoted identifiers) and mysql (no DDL; the bus filter + * matches case-insensitively), but is rejected at DDL install on postgres + * (quoted identifiers are case-sensitive). An exact-spelling sentinel + * binding proves events flow either way — silence is never vacuous. + */ +function caseSensitivityCase(target: NativeTarget): TestCase { + return { + name: 'native capture table casing behaves per driver contract', + applies: [target.applies], + async run({ call, iii }) { + const table = `e2e_native_case_${target.applies}` + const exactFnId = `harness::native_case_exact_${target.applies}` + const upperFnId = `harness::native_case_upper_${target.applies}` + const exactEvents: RowChangedEvent[] = [] + const upperEvents: RowChangedEvent[] = [] + const exact = sink(exactEvents, 'exact-spelling subscriber') + const upper = sink(upperEvents, 'uppercase subscriber') + const ph = target.ph + + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + await call('database::execute', { + db: target.nativeDb, + sql: `CREATE TABLE ${table} (id ${target.idColumnDDL}, n INT NOT NULL)`, + }) + const exactFn = iii.registerFunction( + exactFnId, + async (payload: RowChangedEvent) => { + exactEvents.push(payload) + return null + }, + { description: 'Exact-spelling native binding (sentinel).' }, + ) + const exactTrigger = iii.registerTrigger({ + type: 'database::row-changed', + function_id: exactFnId, + config: { db: target.nativeDb, table }, + }) + const upperFn = iii.registerFunction( + upperFnId, + async (payload: RowChangedEvent) => { + upperEvents.push(payload) + return null + }, + { description: 'Uppercase-spelled native binding.' }, + ) + const upperTrigger = iii.registerTrigger({ + type: 'database::row-changed', + function_id: upperFnId, + config: { db: target.nativeDb, table: table.toUpperCase() }, + }) + + try { + await waitForCaptureReady(call, iii, target, table) + // Let the uppercase registration finish its install attempt (which + // on pg fails, on sqlite reinstalls the same triggers). + await sleep(500) + + await call('database::execute', { + db: target.applies, + sql: `INSERT INTO ${table} (n) VALUES (${ph(1)})`, + params: [1], + }) + const heard = await exact.next() + expectEqual(heard.op, 'insert', 'exact-spelling binding hears the write') + + if (target.uppercaseBindingCaptures) { + const viaUpper = await upper.next() + expectEqual(viaUpper.op, 'insert', 'uppercase binding hears the write') + } + // Exactly one event per subscriber, ever: a differently-cased + // reinstall must converge on ONE trigger set — a second set would + // double-log every write and fail here. + await sleep(SILENCE_WINDOW_MS) + exact.expectDrained() + upper.expectDrained() + } finally { + upperTrigger.unregister() + upperFn.unregister() + exactTrigger.unregister() + exactFn.unregister() + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + } + }, + } +} + +/** + * Two tables, two bindings, no cross-talk: per-table capture installation + * and per-binding filtering keep each subscriber scoped to its own table. + */ +function twoTableIsolationCase(target: NativeTarget): TestCase { + return { + name: 'native capture isolates bindings per table', + applies: [target.applies], + async run({ call, iii }) { + const tableA = `e2e_native_iso_a_${target.applies}` + const tableB = `e2e_native_iso_b_${target.applies}` + const fnA = `harness::native_iso_a_${target.applies}` + const fnB = `harness::native_iso_b_${target.applies}` + const eventsA: RowChangedEvent[] = [] + const eventsB: RowChangedEvent[] = [] + const sinkA = sink(eventsA, 'table-A subscriber') + const sinkB = sink(eventsB, 'table-B subscriber') + const ph = target.ph + + for (const table of [tableA, tableB]) { + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + await call('database::execute', { + db: target.nativeDb, + sql: `CREATE TABLE ${table} (id ${target.idColumnDDL}, n INT NOT NULL)`, + }) + } + const fnARef = iii.registerFunction( + fnA, + async (payload: RowChangedEvent) => { + eventsA.push(payload) + return null + }, + { description: 'Table-A isolation E2E sink.' }, + ) + const triggerA = iii.registerTrigger({ + type: 'database::row-changed', + function_id: fnA, + config: { db: target.nativeDb, table: tableA }, + }) + const fnBRef = iii.registerFunction( + fnB, + async (payload: RowChangedEvent) => { + eventsB.push(payload) + return null + }, + { description: 'Table-B isolation E2E sink.' }, + ) + const triggerB = iii.registerTrigger({ + type: 'database::row-changed', + function_id: fnB, + config: { db: target.nativeDb, table: tableB }, + }) + + try { + await waitForCaptureReady(call, iii, target, tableA) + await waitForCaptureReady(call, iii, target, tableB) + + await call('database::execute', { + db: target.applies, + sql: `INSERT INTO ${tableA} (n) VALUES (${ph(1)})`, + params: [1], + }) + expectEqual((await sinkA.next()).op, 'insert', 'A subscriber hears table A') + await sleep(SILENCE_WINDOW_MS) + sinkB.expectDrained() + + await call('database::execute', { + db: target.applies, + sql: `INSERT INTO ${tableB} (n) VALUES (${ph(1)})`, + params: [2], + }) + expectEqual((await sinkB.next()).op, 'insert', 'B subscriber hears table B') + await sleep(SILENCE_WINDOW_MS) + sinkA.expectDrained() + } finally { + triggerA.unregister() + fnARef.unregister() + triggerB.unregister() + fnBRef.unregister() + for (const table of [tableA, tableB]) { + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + } + } + }, + } +} + +/** + * Registering against a table that does not exist: pg/sqlite must reject + * at DDL install — and stay silent even after the table is created, + * proving the rejection was final, not deferred. mysql has nothing to + * install, so the binding is accepted and starts capturing the moment the + * table appears in the binlog. The asymmetry is contract; pin both sides. + */ +function missingTableRegistrationCase(target: NativeTarget): TestCase { + return { + name: 'native capture registration against a missing table behaves per driver contract', + applies: [target.applies], + async run({ call, iii }) { + const table = `e2e_native_missing_${target.applies}` + const fnId = `harness::native_missing_${target.applies}` + const events: RowChangedEvent[] = [] + const native = sink(events, 'missing-table subscriber') + const ph = target.ph + + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + const fnRef = iii.registerFunction( + fnId, + async (payload: RowChangedEvent) => { + events.push(payload) + return null + }, + { description: 'Missing-table registration E2E sink.' }, + ) + const triggerRef = iii.registerTrigger({ + type: 'database::row-changed', + function_id: fnId, + config: { db: target.nativeDb, table }, + }) + + try { + // Let the registration (and, on pg/sqlite, its failing DDL install) + // fully settle BEFORE the table exists — creating it too early + // would let the install succeed and test nothing. + await sleep(1_000) + + await call('database::execute', { + db: target.nativeDb, + sql: `CREATE TABLE ${table} (id ${target.idColumnDDL}, n INT NOT NULL)`, + }) + await call('database::execute', { + db: target.applies, + sql: `INSERT INTO ${table} (n) VALUES (${ph(1)})`, + params: [1], + }) + + if (target.missingTableBindingAccepted) { + const heard = await native.next() + expectEqual(heard.op, 'insert', 'accepted binding captures once the table exists') + } else { + await sleep(SILENCE_WINDOW_MS) + native.expectDrained() + } + } finally { + triggerRef.unregister() + fnRef.unregister() + await call('database::execute', { db: target.nativeDb, sql: `DROP TABLE IF EXISTS ${table}` }) + } + }, + } +} + +export const NATIVE_CAPTURE_CASES: TestCase[] = TARGETS.flatMap((target) => [ + crossClientCase(target), + tablelessRejectionCase(target), + txGatingCase(target), + fanOutOpsCase(target), + bulkCoalescingCase(target), + caseSensitivityCase(target), + twoTableIsolationCase(target), + missingTableRegistrationCase(target), +]) diff --git a/database/tests/e2e/workers/harness/src/cases-row-changed.ts b/database/tests/e2e/workers/harness/src/cases-row-changed.ts index 736e870c0..1c6775d8a 100644 --- a/database/tests/e2e/workers/harness/src/cases-row-changed.ts +++ b/database/tests/e2e/workers/harness/src/cases-row-changed.ts @@ -16,6 +16,66 @@ interface RowChangedEvent { const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) export const ROW_CHANGED_CASES: TestCase[] = [ + { + // A subscriber should not have to guess how the writer spells the + // table: bindings match case-insensitively (and ignoring any schema + // qualifier) on the statements path. + name: 'row-changed table filter matches case-insensitively', + async run({ driver, dialect, call, iii }) { + const table = 'e2e_row_changed_case' + const functionId = `harness::row_changed_case_${driver}` + const events: RowChangedEvent[] = [] + + await call('database::execute', { db: driver, sql: `DROP TABLE IF EXISTS ${table}` }) + await call('database::execute', { + db: driver, + sql: `CREATE TABLE ${table} (id ${dialect.idColumnDDL()}, n INT NOT NULL)`, + }) + const fnRef = iii.registerFunction( + functionId, + async (payload: RowChangedEvent) => { + events.push(payload) + return null + }, + { description: 'Case-insensitive table filter E2E sink.' }, + ) + const triggerRef = iii.registerTrigger({ + type: 'database::row-changed', + function_id: functionId, + config: { db: driver, table: table.toUpperCase() }, + }) + + try { + // Registration propagates asynchronously; write only once the engine + // can see the binding, or a slow ack fails the case on timeout + // rather than on the behavior under test. + const registered = await iii.trigger< + Record, + { registered_triggers: Array<{ trigger_type: string; function_id: string }> } + >({ function_id: 'engine::registered-triggers::list', payload: {} }) + expect( + registered.registered_triggers.some( + (t) => t.trigger_type === 'database::row-changed' && t.function_id === functionId, + ), + 'case-insensitive binding is visible to the engine', + ) + + await call('database::execute', { + db: driver, + sql: `INSERT INTO ${table} (n) VALUES (${dialect.placeholder(1)})`, + params: [1], + }) + const deadline = Date.now() + EVENT_TIMEOUT_MS + while (events.length === 0 && Date.now() < deadline) await sleep(20) + expectEqual(events.length, 1, 'uppercase binding hears the lowercase table') + expectEqual(events[0].op, 'insert', 'case-insensitive match op') + } finally { + triggerRef.unregister() + fnRef.unregister() + await call('database::execute', { db: driver, sql: `DROP TABLE IF EXISTS ${table}` }) + } + }, + }, { name: 'row-changed filters ops and emits committed mutations only', async run({ driver, dialect, call, iii }) { diff --git a/database/tests/e2e/workers/harness/src/database-config.ts b/database/tests/e2e/workers/harness/src/database-config.ts index e889a56c9..e470ef3a6 100644 --- a/database/tests/e2e/workers/harness/src/database-config.ts +++ b/database/tests/e2e/workers/harness/src/database-config.ts @@ -24,6 +24,31 @@ export const DATABASE_CONFIG_VALUE = { // any system CA. The worker's tls.mode defaults to require. tls: { mode: 'disable' as const }, }, + // Same postgres instance as pg_db, but with native change capture: + // database triggers + LISTEN/NOTIFY instead of SQL classification. Writes + // arriving through OTHER handles (pg_db) are, from this handle's point of + // view, external clients — which is exactly what the native cases prove. + pg_native_db: { + url: process.env.TEST_POSTGRES_URL ?? 'postgres://iii:iii@127.0.0.1:55432/iii_test', + pool: { ...DEFAULT_POOL }, + tls: { mode: 'disable' as const }, + capture: 'native' as const, + }, + // Same file as sqlite_db, native capture via changelog + fs watch. + sqlite_native_db: { + url: 'sqlite:./data/iii.db', + pool: { ...DEFAULT_POOL }, + capture: 'native' as const, + }, + // Same mysql instance as mysql_db, native capture via the binlog + // replication stream. Requires the replication grants installed by + // mysql-init/grant-replication.sql on first compose volume init. + mysql_native_db: { + url: process.env.TEST_MYSQL_URL ?? 'mysql://iii:iii@127.0.0.1:53306/iii_test', + pool: { ...DEFAULT_POOL }, + tls: { mode: 'disable' as const }, + capture: 'native' as const, + }, mysql_db: { url: process.env.TEST_MYSQL_URL ?? 'mysql://iii:iii@127.0.0.1:53306/iii_test', pool: { ...DEFAULT_POOL }, diff --git a/database/tests/e2e/workers/harness/src/runner.ts b/database/tests/e2e/workers/harness/src/runner.ts index 937ebaa0f..8d6776573 100644 --- a/database/tests/e2e/workers/harness/src/runner.ts +++ b/database/tests/e2e/workers/harness/src/runner.ts @@ -10,6 +10,7 @@ import { INTERACTIVE_TX_CASES } from './cases-interactive-tx.ts' import { CONCURRENCY_CASES } from './cases-concurrency.ts' import { TX_CONTROL_BYPASS_CASES } from './cases-tx-control-bypass.ts' import { ROW_CHANGED_CASES } from './cases-row-changed.ts' +import { NATIVE_CAPTURE_CASES } from './cases-native-capture.ts' interface CaseResult { driver: DriverKey @@ -154,6 +155,7 @@ export class Runner { ...INTERACTIVE_TX_CASES, ...CONCURRENCY_CASES, ...ROW_CHANGED_CASES, + ...NATIVE_CAPTURE_CASES, ]) { if (!matchesDriver(driver, c)) continue record(await this.runCase(driver, c)) diff --git a/database/tests/integration.rs b/database/tests/integration.rs index 40438b5c6..8b743c191 100644 --- a/database/tests/integration.rs +++ b/database/tests/integration.rs @@ -172,7 +172,9 @@ async fn apply_config_updates_list_snapshot() { let new_cfg = WorkerConfig::from_yaml(new_yaml).unwrap(); // Act - configuration::apply_config(&st, new_cfg).await.unwrap(); + configuration::apply_config(&st, new_cfg, None) + .await + .unwrap(); let resp = list_databases::handle(&st, ListDatabasesReq::default()) .await .unwrap(); diff --git a/database/ui/page.tsx b/database/ui/page.tsx index a1c00905d..82e432160 100644 --- a/database/ui/page.tsx +++ b/database/ui/page.tsx @@ -5,18 +5,21 @@ * asset: ../styles.css ships over `console:style` as database/styles.css — * the console mounts and link-swaps it, styles-before-scripts on boot. * - * `setup(host)` registers two contributions: + * `setup(host)` registers three contributions: * - src/function-trigger-message/ — how every database::* call renders in * chat and traces (SQL, request chips, result tables). * - src/page/ — the `#/ext/database` browser: schema tree, sortable row * grid, row inspector, and a read-only SQL editor (shared Monaco). Reads * the live database over `database::query`/`database::listDatabases`. + * - src/configuration/ — the configuration form for the `database` entry + * on the Workers tab, replacing the generic schema-driven editor. * * Registrations go through `host` so the loader disposes them on hot * reload / worker disconnect. */ import type { Host } from '@iii-dev/console-ui' +import { DatabaseConfigForm } from './src/configuration' import { createDatabaseTriggerRenderer } from './src/function-trigger-message' import { DatabasePage } from './src/page' @@ -28,4 +31,8 @@ export default function setup(host: Host) { title: 'database', render: () => , }) + + host.configForms.register('database', (props) => ( + + )) } diff --git a/database/ui/src/configuration/index.tsx b/database/ui/src/configuration/index.tsx new file mode 100644 index 000000000..bd361784c --- /dev/null +++ b/database/ui/src/configuration/index.tsx @@ -0,0 +1,383 @@ +/** + * Custom configuration form for the `database` configuration entry — + * registered through `host.configForms`, replacing the console's generic + * schema-driven form for this worker only. + * + * One card per configured database: connection URL with a live driver + * badge, capture mode with driver-aware guidance, TLS (hidden for sqlite, + * which ignores it), and the pool knobs. The form edits the working draft + * via `onChange`; dirty tracking, save/reset, validation and the SaveBar + * stay host-owned. Mirrors DatabaseConfig (database/src/config.rs). + */ + +import { useEffect, useRef, useState } from 'react' +import type { ConfigFormProps, Host, JsonValue } from '@iii-dev/console-ui' + +type JsonObject = { [key: string]: JsonValue } + +function asObject(v: JsonValue | undefined): JsonObject { + return v && typeof v === 'object' && !Array.isArray(v) ? { ...v } : {} +} + +function asString(v: JsonValue | undefined): string { + return typeof v === 'string' ? v : '' +} + +type Driver = 'postgres' | 'mysql' | 'sqlite' | 'unknown' + +function driverOf(url: string): Driver { + if (url.startsWith('postgres://') || url.startsWith('postgresql://')) return 'postgres' + if (url.startsWith('mysql://')) return 'mysql' + if (url.startsWith('sqlite:')) return 'sqlite' + return 'unknown' +} + +const CAPTURE_HINTS: Record = { + postgres: + 'native: any client’s committed writes fire database::row-changed via triggers + LISTEN/NOTIFY. The role needs DDL rights on watched tables; bindings must name a table.', + sqlite: + 'native: triggers + changelog table + filesystem watch hear every process writing the file. Bindings must name a table.', + mysql: + 'native: streams the binlog as a replica — nothing installed in the schema, but the user needs GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.*', + unknown: 'set a url first — capture support depends on the driver.', +} + +const POOL_FIELDS = [ + { key: 'max', label: 'max connections', placeholder: '10' }, + { key: 'idle_timeout_ms', label: 'idle timeout (ms)', placeholder: '30000' }, + { key: 'acquire_timeout_ms', label: 'acquire timeout (ms)', placeholder: '5000' }, +] as const + +/** Wire shape of `database::testConnection`. */ +interface TestConnectionResp { + ok: boolean + driver: string + latency_ms: number + server_version?: string + message?: string +} + +type TestResult = { status: 'testing' } | { status: 'done'; ok: boolean; text: string } + +export function DatabaseConfigForm(props: ConfigFormProps & { host: Host }) { + const value = asObject(props.value) + const databases = asObject(value.databases) + const names = Object.keys(databases) + + // Renames commit on blur: committing per keystroke would collide with a + // sibling entry mid-typing and silently swallow it. + const [pendingNames, setPendingNames] = useState>({}) + // Probe outcomes are keyed by handle and dropped on any edit of that + // handle — a stale "connected" next to a changed url would be a lie. The + // token guards the async completion: an edit/rename/remove while a probe + // is in flight bumps it, and the completion for the superseded probe is + // discarded instead of resurrecting a result for a url it never tested. + const [testResults, setTestResults] = useState>({}) + const testTokens = useRef>({}) + + const commit = (nextDatabases: JsonObject) => + props.onChange({ ...value, databases: nextDatabases }) + + const clearTest = (name: string) => { + testTokens.current[name] = (testTokens.current[name] ?? 0) + 1 + setTestResults((r) => { + const next = { ...r } + delete next[name] + return next + }) + } + + const setDb = (name: string, next: JsonObject) => { + clearTest(name) + commit({ ...databases, [name]: next }) + } + + const runTest = async (name: string) => { + const db = asObject(databases[name]) + const token = (testTokens.current[name] = (testTokens.current[name] ?? 0) + 1) + setTestResults((r) => ({ ...r, [name]: { status: 'testing' } })) + let result: TestResult + try { + const resp = await props.host.iii.trigger( + 'database::testConnection', + { url: asString(db.url), tls: db.tls ?? undefined, timeout_ms: 8000 }, + { timeoutMs: 10_000 }, + ) + result = { + status: 'done', + ok: resp.ok, + text: resp.ok + ? `connected · ${resp.server_version ?? resp.driver} · ${resp.latency_ms}ms` + : resp.message ?? 'connection failed', + } + } catch (e) { + result = { status: 'done', ok: false, text: e instanceof Error ? e.message : String(e) } + } + if (testTokens.current[name] !== token) return // superseded by an edit + setTestResults((r) => ({ ...r, [name]: result })) + } + + const removeDb = (name: string) => { + clearTest(name) + const next = { ...databases } + delete next[name] + commit(next) + } + + const addDb = () => { + let i = names.length + 1 + let name = names.length === 0 ? 'primary' : `db${i}` + while (databases[name] !== undefined) name = `db${++i}` + commit({ ...databases, [name]: { url: `sqlite:./data/${name}.db` } }) + } + + const renameDb = (from: string, to: string) => { + const trimmed = to.trim() + setPendingNames((p) => { + const next = { ...p } + delete next[from] + return next + }) + if (trimmed === '' || trimmed === from || databases[trimmed] !== undefined) return + clearTest(from) + // Rebuild in place so the card doesn't jump to the end of the list. + const next: JsonObject = {} + for (const key of names) { + next[key === from ? trimmed : key] = databases[key] + } + commit(next) + } + + // Deep-link focus (`#/workers/configuration/database/`): a custom + // form honors `focusField` itself. First segment `databases` + a name + // scrolls that card; anything else matches a `data-field` directly. + const rootRef = useRef(null) + useEffect(() => { + const path = props.focusField + if (!path || path.length === 0 || !rootRef.current) return + const selector = + path[0] === 'databases' && path[1] + ? `[data-field="db-${path[1]}"]` + : `[data-field="${path[0]}"]` + const target = rootRef.current.querySelector(selector) + target?.focus() + target?.scrollIntoView({ block: 'center' }) + }, [props.focusField]) + + return ( +
+ custom form · shipped by the database worker + + {names.length === 0 ? ( +
No databases configured — the worker refuses to start without at least one.
+ ) : null} + + {names.map((name) => ( + setPendingNames((p) => ({ ...p, [name]: v }))} + onRename={(to) => renameDb(name, to)} + onChange={(next) => setDb(name, next)} + onRemove={() => removeDb(name)} + removable={names.length > 1} + test={testResults[name]} + onTest={() => runTest(name)} + /> + ))} + + + + {props.errors && props.errors.size > 0 ? ( +
+ {[...props.errors.entries()].map(([pointer, message]) => ( +
+ {pointer ? `${pointer}: ` : ''} + {message} +
+ ))} +
+ ) : null} +
+ ) +} + +function DatabaseCard(card: { + name: string + db: JsonObject + pendingName: string | undefined + onPendingName: (v: string) => void + onRename: (to: string) => void + onChange: (next: JsonObject) => void + onRemove: () => void + removable: boolean + test: TestResult | undefined + onTest: () => void +}) { + const { name, db } = card + const url = asString(db.url) + const driver = driverOf(url) + const capture = asString(db.capture) || 'statements' + const tls = asObject(db.tls) + const pool = asObject(db.pool) + const isMemorySqlite = driver === 'sqlite' && url.includes(':memory:') + + const set = (mutate: (next: JsonObject) => void) => { + const next = { ...db } + mutate(next) + card.onChange(next) + } + + const setBlock = (key: 'tls' | 'pool', mutate: (block: JsonObject) => void) => + set((next) => { + const block = asObject(next[key]) + mutate(block) + if (Object.keys(block).length > 0) next[key] = block + else delete next[key] + }) + + return ( +
+
+ card.onPendingName(e.target.value)} + onBlur={(e) => card.onRename(e.target.value)} + /> + {driver} + {capture === 'native' ? native capture : null} + + {card.removable ? ( + + ) : null} +
+ +
+ +
+ set((next) => (next.url = e.target.value))} + /> + +
+ {card.test?.status === 'done' ? ( + + {card.test.text} + + ) : null} +
+ +
+ + + {capture === 'native' && isMemorySqlite ? ( + + a `:memory:` database is per-connection and cannot be captured — the worker rejects + this configuration + + ) : ( + {CAPTURE_HINTS[driver]} + )} +
+ + {driver === 'postgres' || driver === 'mysql' ? ( +
+
+ + +
+
+ + + setBlock('tls', (block) => { + if (e.target.value === '') delete block.ca_cert + else block.ca_cert = e.target.value + }) + } + /> +
+
+ ) : null} + +
+ connection pool +
+ {POOL_FIELDS.map((f) => ( +
+ + + setBlock('pool', (block) => { + if (e.target.value.trim() === '') delete block[f.key] + else if (!Number.isNaN(Number(e.target.value))) block[f.key] = Number(e.target.value) + }) + } + /> +
+ ))} +
+
+
+ ) +} diff --git a/database/ui/styles.css b/database/ui/styles.css index 61e177c8a..2afaa3fce 100644 --- a/database/ui/styles.css +++ b/database/ui/styles.css @@ -449,3 +449,180 @@ [data-iii-ui="database"] .db-data { flex-direction: column; } [data-iii-ui="database"] .db-rowdetail { width: auto; border-left: 0; border-top: 1px solid var(--color-rule); } } + +/* --- configuration form (Workers tab, `database` entry) --------------- */ +[data-iii-ui="database"] .db-cfg { + display: flex; + flex-direction: column; + gap: 12px; +} +[data-iii-ui="database"] .db-cfg-caption { + font-size: 11px; + color: var(--color-ink-ghost); +} +[data-iii-ui="database"] .db-cfg-empty { + padding: 16px; + border: 1px dashed var(--color-rule); + border-radius: 8px; + color: var(--color-ink-faint); + font-size: 13px; +} +[data-iii-ui="database"] .db-cfg-card { + border: 1px solid var(--color-rule); + border-radius: 8px; + background: var(--color-panel); + padding: 12px; + display: flex; + flex-direction: column; + gap: 10px; +} +[data-iii-ui="database"] .db-cfg-card:focus { + outline: 2px solid var(--color-ring); + outline-offset: 1px; +} +[data-iii-ui="database"] .db-cfg-card-head { + display: flex; + align-items: center; + gap: 8px; +} +[data-iii-ui="database"] .db-cfg-name { + font-family: var(--font-mono, ui-monospace, monospace); + font-size: 14px; + font-weight: 600; + color: var(--color-ink); + background: transparent; + border: 1px solid transparent; + border-radius: 6px; + padding: 2px 6px; + width: 12ch; +} +[data-iii-ui="database"] .db-cfg-name:hover, +[data-iii-ui="database"] .db-cfg-name:focus { + border-color: var(--color-rule); + background: var(--color-bg); +} +[data-iii-ui="database"] .db-cfg-driver { + font-size: 11px; + text-transform: uppercase; + letter-spacing: 0.04em; + padding: 1px 8px; + border-radius: 999px; + border: 1px solid var(--color-rule-2); + color: var(--color-ink-faint); +} +[data-iii-ui="database"] .db-cfg-driver-postgres, +[data-iii-ui="database"] .db-cfg-driver-mysql, +[data-iii-ui="database"] .db-cfg-driver-sqlite { + border-color: var(--color-accent); + color: var(--color-accent); +} +[data-iii-ui="database"] .db-cfg-capture-pill { + font-size: 11px; + padding: 1px 8px; + border-radius: 999px; + border: 1px solid var(--color-ok); + color: var(--color-ok); +} +[data-iii-ui="database"] .db-cfg-spacer { flex: 1; } +[data-iii-ui="database"] .db-cfg-remove, +[data-iii-ui="database"] .db-cfg-add { + font-size: 12px; + color: var(--color-ink-faint); + background: transparent; + border: 1px solid var(--color-rule); + border-radius: 6px; + padding: 3px 10px; + cursor: pointer; +} +[data-iii-ui="database"] .db-cfg-remove:hover { color: var(--color-alert); border-color: var(--color-alert); } +[data-iii-ui="database"] .db-cfg-add { align-self: flex-start; } +[data-iii-ui="database"] .db-cfg-add:hover { color: var(--color-accent); border-color: var(--color-accent); } +[data-iii-ui="database"] .db-cfg-field { + display: flex; + flex-direction: column; + gap: 4px; + min-width: 0; +} +[data-iii-ui="database"] .db-cfg-field > label { + font-size: 11px; + text-transform: uppercase; + letter-spacing: 0.04em; + color: var(--color-ink-faint); +} +[data-iii-ui="database"] .db-cfg-input, +[data-iii-ui="database"] .db-cfg-select { + font-size: 13px; + font-family: var(--font-mono, ui-monospace, monospace); + color: var(--color-ink); + background: var(--color-bg); + border: 1px solid var(--color-rule); + border-radius: 6px; + padding: 6px 8px; + min-width: 0; +} +[data-iii-ui="database"] .db-cfg-input:focus, +[data-iii-ui="database"] .db-cfg-select:focus, +[data-iii-ui="database"] .db-cfg-name:focus-visible { + outline: 2px solid var(--color-ring); + outline-offset: -1px; +} +[data-iii-ui="database"] .db-cfg-row { + display: flex; + gap: 10px; + flex-wrap: wrap; +} +[data-iii-ui="database"] .db-cfg-row > .db-cfg-field { flex: 1 1 160px; } +[data-iii-ui="database"] .db-cfg-grow { flex: 2 1 240px; } +[data-iii-ui="database"] .db-cfg-hint, +[data-iii-ui="database"] .db-cfg-warn { + font-size: 12px; + color: var(--color-ink-faint); + line-height: 1.4; +} +[data-iii-ui="database"] .db-cfg-warn { color: var(--color-alert); } +[data-iii-ui="database"] .db-cfg-pool > summary { + font-size: 11px; + text-transform: uppercase; + letter-spacing: 0.04em; + color: var(--color-ink-faint); + cursor: pointer; +} +[data-iii-ui="database"] .db-cfg-pool[open] > summary { margin-bottom: 8px; } +[data-iii-ui="database"] .db-cfg-errors { + border: 1px solid var(--color-alert); + border-radius: 8px; + padding: 8px 12px; + font-size: 12px; + color: var(--color-alert); + display: flex; + flex-direction: column; + gap: 4px; +} +[data-iii-ui="database"] .db-cfg-url-row { + display: flex; + gap: 8px; + align-items: stretch; +} +[data-iii-ui="database"] .db-cfg-test { + font-size: 12px; + white-space: nowrap; + color: var(--color-ink-faint); + background: transparent; + border: 1px solid var(--color-rule); + border-radius: 6px; + padding: 3px 10px; + cursor: pointer; +} +[data-iii-ui="database"] .db-cfg-test:hover:not(:disabled) { + color: var(--color-accent); + border-color: var(--color-accent); +} +[data-iii-ui="database"] .db-cfg-test:disabled { opacity: 0.6; cursor: default; } +[data-iii-ui="database"] .db-cfg-test-ok, +[data-iii-ui="database"] .db-cfg-test-fail { + font-size: 12px; + font-family: var(--font-mono, ui-monospace, monospace); + line-height: 1.4; +} +[data-iii-ui="database"] .db-cfg-test-ok { color: var(--color-ok); } +[data-iii-ui="database"] .db-cfg-test-fail { color: var(--color-alert); }