#[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
extern crate libsqlite3_sys as ffi;
#[cfg(all(target_family = "wasm", target_os = "unknown"))]
use sqlite_wasm_rs as ffi;
use super::SqliteConnection;
use crate::query_source::NamedTable;
use alloc::boxed::Box;
use alloc::vec::Vec;
bitflags::bitflags! {
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct SqliteChangeOps: u8 {
const INSERT = 1;
const UPDATE = 2;
const DELETE = 4;
const UNKNOWN = 8;
const ALL =
Self::INSERT.bits() | Self::UPDATE.bits() | Self::DELETE.bits() | Self::UNKNOWN.bits();
}
}
impl SqliteChangeOps {
pub(crate) fn matches_op(self, op: SqliteChangeOp) -> bool {
self.contains(op.to_ops())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum SqliteChangeOp {
Insert,
Update,
Delete,
Unknown(i32),
}
impl SqliteChangeOp {
pub(crate) fn from_ffi(code: i32) -> Self {
#[allow(non_upper_case_globals)]
match code {
ffi::SQLITE_INSERT => SqliteChangeOp::Insert,
ffi::SQLITE_UPDATE => SqliteChangeOp::Update,
ffi::SQLITE_DELETE => SqliteChangeOp::Delete,
other => SqliteChangeOp::Unknown(other),
}
}
pub(crate) fn to_ops(self) -> SqliteChangeOps {
match self {
SqliteChangeOp::Insert => SqliteChangeOps::INSERT,
SqliteChangeOp::Update => SqliteChangeOps::UPDATE,
SqliteChangeOp::Delete => SqliteChangeOps::DELETE,
SqliteChangeOp::Unknown(_) => SqliteChangeOps::UNKNOWN,
}
}
}
#[derive(Debug, Clone, Copy)]
#[non_exhaustive]
pub struct SqliteChangeEvent<'a> {
pub op: SqliteChangeOp,
pub db_name: &'a str,
pub table_name: &'a str,
pub rowid: i64,
}
impl SqliteChangeEvent<'_> {
pub fn is_from(&self, table: impl NamedTable) -> bool {
self.table_name == table.table() && table.schema().is_none_or(|db| self.db_name == db)
}
pub fn rowid_in(&self, table: impl NamedTable) -> Option<i64> {
if self.is_from(table) {
Some(self.rowid)
} else {
None
}
}
}
trait DynNamedTable {
fn schema(&self) -> Option<&str>;
fn table(&self) -> &str;
}
impl<T> DynNamedTable for T
where
T: NamedTable,
{
fn schema(&self) -> Option<&str> {
NamedTable::schema(self)
}
fn table(&self) -> &str {
NamedTable::table(self)
}
}
struct Route {
table: Option<Box<dyn DynNamedTable + Send>>,
ops: SqliteChangeOps,
callback: Box<dyn FnMut(SqliteChangeEvent<'_>) + Send>,
}
impl Route {
fn matches(&self, event: &SqliteChangeEvent<'_>) -> bool {
self.ops.matches_op(event.op)
&& self.table.as_deref().is_none_or(|name| {
name.table() == event.table_name
&& name.schema().is_none_or(|db| db == event.db_name)
})
}
}
#[allow(missing_debug_implementations)]
pub struct SqliteUpdateRouter {
routes: Vec<Route>,
}
impl SqliteUpdateRouter {
pub fn new() -> Self {
SqliteUpdateRouter { routes: Vec::new() }
}
pub fn on<T, F>(mut self, table: T, ops: SqliteChangeOps, callback: F) -> Self
where
T: NamedTable + Send + 'static,
F: FnMut(SqliteChangeEvent<'_>) + Send + 'static,
{
let _ = table;
self.routes.push(Route {
table: Some(Box::new(table)),
ops,
callback: Box::new(callback),
});
self
}
pub fn on_any<F>(mut self, ops: SqliteChangeOps, callback: F) -> Self
where
F: FnMut(SqliteChangeEvent<'_>) + Send + 'static,
{
self.routes.push(Route {
table: None,
ops,
callback: Box::new(callback),
});
self
}
fn dispatch(&mut self, event: SqliteChangeEvent<'_>) {
for route in &mut self.routes {
if route.matches(&event) {
(route.callback)(event);
}
}
}
pub(crate) fn into_hook(mut self) -> impl FnMut(SqliteChangeEvent<'_>) + Send {
move |event| self.dispatch(event)
}
}
impl Default for SqliteUpdateRouter {
fn default() -> Self {
SqliteUpdateRouter::new()
}
}
impl SqliteConnection {
pub fn on_update(&mut self, router: SqliteUpdateRouter) {
self.raw_connection.set_update_hook(router.into_hook());
}
pub fn remove_update_hook(&mut self) {
self.raw_connection.remove_update_hook();
}
}
#[cfg(all(test, not(miri)))]
mod tests {
use super::*;
use crate::connection::Connection;
use crate::prelude::*;
use crate::query_dsl::RunQueryDsl;
impl SqliteChangeOps {
fn matches(self, op_code: i32) -> bool {
self.matches_op(SqliteChangeOp::from_ffi(op_code))
}
}
#[test]
fn insert_or_delete_matches_both_but_not_update() {
let mask = SqliteChangeOps::INSERT | SqliteChangeOps::DELETE;
assert!(mask.matches(ffi::SQLITE_INSERT));
assert!(mask.matches(ffi::SQLITE_DELETE));
assert!(!mask.matches(ffi::SQLITE_UPDATE));
}
#[test]
fn all_matches_all_three() {
assert!(SqliteChangeOps::ALL.matches(ffi::SQLITE_INSERT));
assert!(SqliteChangeOps::ALL.matches(ffi::SQLITE_UPDATE));
assert!(SqliteChangeOps::ALL.matches(ffi::SQLITE_DELETE));
}
#[test]
fn combining_identical_masks_is_idempotent() {
assert_eq!(
SqliteChangeOps::INSERT | SqliteChangeOps::INSERT,
SqliteChangeOps::INSERT,
);
}
#[test]
fn contains_single() {
assert!(SqliteChangeOps::INSERT.contains(SqliteChangeOps::INSERT));
}
#[test]
fn all_contains_insert_or_delete() {
assert!(SqliteChangeOps::ALL.contains(SqliteChangeOps::INSERT | SqliteChangeOps::DELETE));
}
#[test]
fn insert_does_not_contain_all() {
assert!(!SqliteChangeOps::INSERT.contains(SqliteChangeOps::ALL));
}
#[test]
fn from_ffi_insert() {
assert_eq!(
SqliteChangeOp::from_ffi(ffi::SQLITE_INSERT),
SqliteChangeOp::Insert
);
}
#[test]
fn from_ffi_update() {
assert_eq!(
SqliteChangeOp::from_ffi(ffi::SQLITE_UPDATE),
SqliteChangeOp::Update
);
}
#[test]
fn from_ffi_delete() {
assert_eq!(
SqliteChangeOp::from_ffi(ffi::SQLITE_DELETE),
SqliteChangeOp::Delete
);
}
#[test]
fn to_ops_roundtrip() {
assert_eq!(SqliteChangeOp::Insert.to_ops(), SqliteChangeOps::INSERT);
assert_eq!(SqliteChangeOp::Update.to_ops(), SqliteChangeOps::UPDATE);
assert_eq!(SqliteChangeOp::Delete.to_ops(), SqliteChangeOps::DELETE);
assert_eq!(
SqliteChangeOp::Unknown(999).to_ops(),
SqliteChangeOps::UNKNOWN
);
}
#[test]
fn from_ffi_unknown_code() {
assert_eq!(SqliteChangeOp::from_ffi(999), SqliteChangeOp::Unknown(999));
}
#[test]
fn sqlite_change_event_is_copy() {
let event = SqliteChangeEvent {
op: SqliteChangeOp::Delete,
db_name: "main",
table_name: "posts",
rowid: 7,
};
let a = event;
let b = event;
assert_eq!(a.rowid, b.rowid);
}
#[test]
fn debug_formatting() {
assert!(format!("{:?}", SqliteChangeOps::INSERT).contains("INSERT"));
assert!(format!("{:?}", SqliteChangeOps::DELETE).contains("DELETE"));
let _ = format!("{:?}", SqliteChangeOps::empty());
let _ = format!("{:?}", SqliteChangeOps::ALL);
}
#[test]
fn bitand_works() {
let mask = SqliteChangeOps::ALL & SqliteChangeOps::INSERT;
assert_eq!(mask, SqliteChangeOps::INSERT);
}
fn make_event(
op: SqliteChangeOp,
table: &'static str,
rowid: i64,
) -> SqliteChangeEvent<'static> {
SqliteChangeEvent {
op,
db_name: "main",
table_name: table,
rowid,
}
}
#[test]
fn empty_router_dispatches_nothing() {
let mut router = SqliteUpdateRouter::new();
router.dispatch(make_event(SqliteChangeOp::Insert, "users", 1));
}
#[test]
fn on_any_dispatches_for_every_table() {
let fired = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let f2 = fired.clone();
let mut router = SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |e| {
f2.lock().unwrap().push((e.op, e.rowid));
});
router.dispatch(make_event(SqliteChangeOp::Insert, "users", 1));
router.dispatch(make_event(SqliteChangeOp::Delete, "posts", 2));
assert_eq!(
*fired.lock().unwrap(),
vec![(SqliteChangeOp::Insert, 1), (SqliteChangeOp::Delete, 2)],
);
}
#[test]
fn router_filters_by_op_mask() {
let fired = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let f2 = fired.clone();
let mut router = SqliteUpdateRouter::new().on_any(SqliteChangeOps::INSERT, move |e| {
f2.lock().unwrap().push(e.rowid);
});
router.dispatch(make_event(SqliteChangeOp::Insert, "users", 1));
router.dispatch(make_event(SqliteChangeOp::Update, "users", 2)); router.dispatch(make_event(SqliteChangeOp::Delete, "users", 3));
assert_eq!(*fired.lock().unwrap(), vec![1]);
}
#[test]
fn every_matching_route_fires() {
let count = std::sync::Arc::new(std::sync::Mutex::new(0u32));
let c1 = count.clone();
let c2 = count.clone();
let mut router = SqliteUpdateRouter::new()
.on_any(SqliteChangeOps::ALL, move |_| {
*c1.lock().unwrap() += 1;
})
.on_any(SqliteChangeOps::INSERT, move |_| {
*c2.lock().unwrap() += 1;
});
router.dispatch(make_event(SqliteChangeOp::Insert, "users", 1));
router.dispatch(make_event(SqliteChangeOp::Delete, "users", 2));
assert_eq!(*count.lock().unwrap(), 3);
}
fn connection() -> SqliteConnection {
SqliteConnection::establish(":memory:").unwrap()
}
table! {
hook_users {
id -> Integer,
name -> Text,
}
}
table! {
hook_posts {
id -> Integer,
title -> Text,
}
}
fn setup_hook_tables(conn: &mut SqliteConnection) {
crate::sql_query(
"CREATE TABLE hook_users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL)",
)
.execute(conn)
.unwrap();
crate::sql_query(
"CREATE TABLE hook_posts (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL)",
)
.execute(conn)
.unwrap();
}
#[diesel_test_helper::test]
fn router_on_matches_schema_qualified_table() {
use std::sync::{Arc, Mutex};
table! {
attached.shared_items (id) {
id -> Integer,
}
}
let conn = &mut connection();
crate::sql_query("ATTACH DATABASE ':memory:' AS attached")
.execute(conn)
.unwrap();
crate::sql_query("CREATE TABLE shared_items (id INTEGER PRIMARY KEY)")
.execute(conn)
.unwrap();
crate::sql_query("CREATE TABLE attached.shared_items (id INTEGER PRIMARY KEY)")
.execute(conn)
.unwrap();
let fired = Arc::new(Mutex::new(Vec::new()));
let f2 = fired.clone();
conn.on_update(SqliteUpdateRouter::new().on(
shared_items::table,
SqliteChangeOps::ALL,
move |ev| {
f2.lock().unwrap().push((ev.db_name.to_owned(), ev.rowid));
},
));
crate::sql_query("INSERT INTO main.shared_items (id) VALUES (1)")
.execute(conn)
.unwrap();
crate::sql_query("INSERT INTO attached.shared_items (id) VALUES (2)")
.execute(conn)
.unwrap();
assert_eq!(
*fired.lock().unwrap(),
vec![("attached".to_owned(), 2)],
"a schema-qualified route matches only its attached database"
);
}
#[diesel_test_helper::test]
fn router_on_dispatches_to_typed_table() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let fired = Arc::new(Mutex::new(Vec::new()));
let fired2 = fired.clone();
conn.on_update(SqliteUpdateRouter::new().on(
hook_users::table,
SqliteChangeOps::INSERT,
move |change| {
fired2.lock().unwrap().push((change.op, change.rowid));
},
));
crate::sql_query("INSERT INTO hook_users (name) VALUES ('Alice')")
.execute(conn)
.unwrap();
let events = fired.lock().unwrap().clone();
assert_eq!(events.len(), 1);
assert_eq!(events[0].0, SqliteChangeOp::Insert);
assert_eq!(events[0].1, 1); }
#[diesel_test_helper::test]
fn on_delete_fires_only_for_delete() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let fired = Arc::new(Mutex::new(Vec::new()));
let fired2 = fired.clone();
conn.on_update(SqliteUpdateRouter::new().on(
hook_users::table,
SqliteChangeOps::DELETE,
move |change| {
fired2.lock().unwrap().push(change.op);
},
));
crate::sql_query("INSERT INTO hook_users (name) VALUES ('Alice')")
.execute(conn)
.unwrap();
crate::sql_query("UPDATE hook_users SET name = 'Bob' WHERE id = 1")
.execute(conn)
.unwrap();
crate::sql_query("DELETE FROM hook_users WHERE id = 1")
.execute(conn)
.unwrap();
let events = fired.lock().unwrap().clone();
assert_eq!(events.len(), 1);
assert_eq!(events[0], SqliteChangeOp::Delete);
}
#[diesel_test_helper::test]
fn every_matching_route_fires_in_order() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let order = Arc::new(Mutex::new(Vec::new()));
let o1 = order.clone();
let o2 = order.clone();
conn.on_update(
SqliteUpdateRouter::new()
.on(hook_users::table, SqliteChangeOps::INSERT, move |_| {
o1.lock().unwrap().push(1);
})
.on(hook_users::table, SqliteChangeOps::INSERT, move |_| {
o2.lock().unwrap().push(2);
}),
);
crate::sql_query("INSERT INTO hook_users (name) VALUES ('X')")
.execute(conn)
.unwrap();
assert_eq!(*order.lock().unwrap(), vec![1, 2]);
}
#[diesel_test_helper::test]
fn remove_update_stops_dispatch() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let fired = Arc::new(Mutex::new(0u32));
let f2 = fired.clone();
conn.on_update(
SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |_| {
*f2.lock().unwrap() += 1;
}),
);
crate::sql_query("INSERT INTO hook_users (name) VALUES ('A')")
.execute(conn)
.unwrap();
assert_eq!(*fired.lock().unwrap(), 1);
conn.remove_update_hook();
crate::sql_query("INSERT INTO hook_users (name) VALUES ('B')")
.execute(conn)
.unwrap();
assert_eq!(*fired.lock().unwrap(), 1);
}
#[diesel_test_helper::test]
fn events_fire_immediately_during_statement() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
crate::sql_query("INSERT INTO hook_users (name) VALUES ('Z')")
.execute(conn)
.unwrap();
let fired = Arc::new(Mutex::new(Vec::new()));
let f2 = fired.clone();
conn.on_update(SqliteUpdateRouter::new().on(
hook_users::table,
SqliteChangeOps::UPDATE,
move |event| {
f2.lock().unwrap().push(event.rowid);
},
));
crate::sql_query("UPDATE hook_users SET name = 'W' WHERE id = 1")
.execute(conn)
.unwrap();
assert_eq!(*fired.lock().unwrap(), vec![1i64]);
}
#[diesel_test_helper::test]
fn on_update_fires_for_update_only() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let count = Arc::new(Mutex::new(0u32));
let c2 = count.clone();
conn.on_update(SqliteUpdateRouter::new().on(
hook_users::table,
SqliteChangeOps::UPDATE,
move |event| {
assert_eq!(event.op, SqliteChangeOp::Update);
*c2.lock().unwrap() += 1;
},
));
crate::sql_query("INSERT INTO hook_users (name) VALUES ('A')")
.execute(conn)
.unwrap();
crate::sql_query("UPDATE hook_users SET name = 'B' WHERE id = 1")
.execute(conn)
.unwrap();
crate::sql_query("DELETE FROM hook_users WHERE id = 1")
.execute(conn)
.unwrap();
assert_eq!(*count.lock().unwrap(), 1);
}
#[diesel_test_helper::test]
fn on_update_receives_every_change() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let events = Arc::new(Mutex::new(Vec::new()));
let e2 = events.clone();
conn.on_update(
SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |ev| {
e2.lock().unwrap().push((ev.op, ev.table_name.to_owned()));
}),
);
crate::sql_query("INSERT INTO hook_users (name) VALUES ('A')")
.execute(conn)
.unwrap();
crate::sql_query("INSERT INTO hook_posts (title) VALUES ('P')")
.execute(conn)
.unwrap();
crate::sql_query("UPDATE hook_users SET name = 'B' WHERE id = 1")
.execute(conn)
.unwrap();
crate::sql_query("DELETE FROM hook_posts WHERE id = 1")
.execute(conn)
.unwrap();
let evts = events.lock().unwrap().clone();
assert_eq!(evts.len(), 4);
assert_eq!(evts[0], (SqliteChangeOp::Insert, "hook_users".to_owned()));
assert_eq!(evts[1], (SqliteChangeOp::Insert, "hook_posts".to_owned()));
assert_eq!(evts[2], (SqliteChangeOp::Update, "hook_users".to_owned()));
assert_eq!(evts[3], (SqliteChangeOp::Delete, "hook_posts".to_owned()));
}
#[diesel_test_helper::test]
fn on_update_filters_by_op_mask() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let count = Arc::new(Mutex::new(0u32));
let c2 = count.clone();
conn.on_update(SqliteUpdateRouter::new().on_any(
SqliteChangeOps::INSERT | SqliteChangeOps::DELETE,
move |_| {
*c2.lock().unwrap() += 1;
},
));
crate::sql_query("INSERT INTO hook_users (name) VALUES ('A')")
.execute(conn)
.unwrap();
crate::sql_query("UPDATE hook_users SET name = 'B' WHERE id = 1")
.execute(conn)
.unwrap();
crate::sql_query("DELETE FROM hook_users WHERE id = 1")
.execute(conn)
.unwrap();
assert_eq!(*count.lock().unwrap(), 2);
}
#[diesel_test_helper::test]
fn router_dispatches_to_multiple_tables() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let user_count = Arc::new(Mutex::new(0u32));
let post_count = Arc::new(Mutex::new(0u32));
let uc = user_count.clone();
let pc = post_count.clone();
conn.on_update(
SqliteUpdateRouter::new()
.on(hook_users::table, SqliteChangeOps::ALL, move |_| {
*uc.lock().unwrap() += 1;
})
.on(hook_posts::table, SqliteChangeOps::ALL, move |_| {
*pc.lock().unwrap() += 1;
}),
);
crate::sql_query("INSERT INTO hook_users (name) VALUES ('X')")
.execute(conn)
.unwrap();
crate::sql_query("INSERT INTO hook_posts (title) VALUES ('Y')")
.execute(conn)
.unwrap();
assert_eq!(*user_count.lock().unwrap(), 1);
assert_eq!(*post_count.lock().unwrap(), 1);
}
#[diesel_test_helper::test]
fn on_any_audit_plus_specific_route() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let audit_count = Arc::new(Mutex::new(0u32));
let user_insert_count = Arc::new(Mutex::new(0u32));
let ac = audit_count.clone();
let uic = user_insert_count.clone();
conn.on_update(
SqliteUpdateRouter::new()
.on_any(SqliteChangeOps::ALL, move |_| {
*ac.lock().unwrap() += 1;
})
.on(hook_users::table, SqliteChangeOps::INSERT, move |_| {
*uic.lock().unwrap() += 1;
}),
);
crate::sql_query("INSERT INTO hook_users (name) VALUES ('X')")
.execute(conn)
.unwrap();
crate::sql_query("INSERT INTO hook_posts (title) VALUES ('Y')")
.execute(conn)
.unwrap();
assert_eq!(*audit_count.lock().unwrap(), 2);
assert_eq!(*user_insert_count.lock().unwrap(), 1);
}
#[diesel_test_helper::test]
fn rowid_in_filters_by_table() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let captured = Arc::new(Mutex::new(Vec::new()));
let c2 = captured.clone();
conn.on_update(
SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |change| {
if let Some(rowid) = change.rowid_in(hook_users::table) {
c2.lock().unwrap().push(rowid);
}
}),
);
crate::sql_query("INSERT INTO hook_users (name) VALUES ('A')")
.execute(conn)
.unwrap();
crate::sql_query("INSERT INTO hook_posts (title) VALUES ('P')")
.execute(conn)
.unwrap();
assert_eq!(*captured.lock().unwrap(), vec![1i64]);
}
#[diesel_test_helper::test]
fn is_from_matches_table_marker() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let captured = Arc::new(Mutex::new(Vec::new()));
let c2 = captured.clone();
conn.on_update(
SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |change| {
c2.lock().unwrap().push((
change.is_from(hook_users::table),
change.is_from(hook_posts::table),
));
}),
);
crate::sql_query("INSERT INTO hook_users (name) VALUES ('A')")
.execute(conn)
.unwrap();
assert_eq!(*captured.lock().unwrap(), vec![(true, false)]);
}
#[diesel_test_helper::test]
fn hooks_fire_across_transactions() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let fired = Arc::new(Mutex::new(Vec::new()));
let f2 = fired.clone();
conn.on_update(SqliteUpdateRouter::new().on(
hook_users::table,
SqliteChangeOps::INSERT,
move |event| {
f2.lock().unwrap().push(event.rowid);
},
));
conn.immediate_transaction(|conn| {
crate::sql_query("INSERT INTO hook_users (name) VALUES ('TxUser')")
.execute(conn)
.unwrap();
Ok::<_, crate::result::Error>(())
})
.unwrap();
assert_eq!(fired.lock().unwrap().len(), 1);
}
#[diesel_test_helper::test]
fn update_hook_silent_for_without_rowid_tables() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
crate::sql_query("CREATE TABLE kv (key TEXT PRIMARY KEY, val TEXT NOT NULL) WITHOUT ROWID")
.execute(conn)
.unwrap();
let events: Arc<Mutex<Vec<SqliteChangeOp>>> = Arc::new(Mutex::new(Vec::new()));
let e2 = events.clone();
conn.on_update(
SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |ev| {
if ev.table_name == "kv" {
e2.lock().unwrap().push(ev.op);
}
}),
);
crate::sql_query("INSERT INTO kv (key, val) VALUES ('a', '1')")
.execute(conn)
.unwrap();
crate::sql_query("UPDATE kv SET val = '2' WHERE key = 'a'")
.execute(conn)
.unwrap();
crate::sql_query("DELETE FROM kv WHERE key = 'a'")
.execute(conn)
.unwrap();
assert!(
events.lock().unwrap().is_empty(),
"update hook must not fire for WITHOUT ROWID tables"
);
}
#[diesel_test_helper::test]
fn update_hook_silent_for_on_conflict_replace_deletion() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
crate::sql_query("CREATE TABLE uq (id INTEGER PRIMARY KEY, val TEXT NOT NULL UNIQUE)")
.execute(conn)
.unwrap();
crate::sql_query("INSERT INTO uq (id, val) VALUES (1, 'original')")
.execute(conn)
.unwrap();
let events: Arc<Mutex<Vec<(SqliteChangeOp, i64)>>> = Arc::new(Mutex::new(Vec::new()));
let e2 = events.clone();
conn.on_update(
SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |ev| {
if ev.table_name == "uq" {
e2.lock().unwrap().push((ev.op, ev.rowid));
}
}),
);
crate::sql_query("INSERT OR REPLACE INTO uq (id, val) VALUES (2, 'original')")
.execute(conn)
.unwrap();
let recorded = events.lock().unwrap();
assert_eq!(
recorded.len(),
1,
"expected only 1 event (INSERT), got: {:?}",
*recorded
);
assert_eq!(recorded[0].0, SqliteChangeOp::Insert);
assert_eq!(recorded[0].1, 2, "new row should have rowid 2");
}
#[diesel_test_helper::test]
fn update_hook_silent_for_truncate_optimization() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
crate::sql_query("CREATE TABLE bulk (id INTEGER PRIMARY KEY, data TEXT NOT NULL)")
.execute(conn)
.unwrap();
crate::sql_query("INSERT INTO bulk (data) VALUES ('a'), ('b'), ('c')")
.execute(conn)
.unwrap();
let events: Arc<Mutex<Vec<SqliteChangeOp>>> = Arc::new(Mutex::new(Vec::new()));
let e2 = events.clone();
conn.on_update(
SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |ev| {
if ev.table_name == "bulk" {
e2.lock().unwrap().push(ev.op);
}
}),
);
crate::sql_query("DELETE FROM bulk").execute(conn).unwrap();
assert!(
events.lock().unwrap().is_empty(),
"truncate optimization should bypass the update hook"
);
}
#[diesel_test_helper::test]
fn update_hook_fires_for_delete_all_when_triggers_disable_truncate() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
crate::sql_query("CREATE TABLE triggered (id INTEGER PRIMARY KEY, data TEXT NOT NULL)")
.execute(conn)
.unwrap();
crate::sql_query(
"CREATE TRIGGER trg_triggered BEFORE DELETE ON triggered \
BEGIN SELECT 1; END",
)
.execute(conn)
.unwrap();
crate::sql_query("INSERT INTO triggered (data) VALUES ('x'), ('y'), ('z')")
.execute(conn)
.unwrap();
let deletes: Arc<Mutex<Vec<i64>>> = Arc::new(Mutex::new(Vec::new()));
let d2 = deletes.clone();
conn.on_update(
SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |ev| {
if ev.table_name == "triggered" && ev.op == SqliteChangeOp::Delete {
d2.lock().unwrap().push(ev.rowid);
}
}),
);
crate::sql_query("DELETE FROM triggered")
.execute(conn)
.unwrap();
assert_eq!(
deletes.lock().unwrap().len(),
3,
"with triggers present, DELETE without WHERE fires per-row hooks"
);
}
#[diesel_test_helper::test]
fn update_hook_silent_for_internal_sqlite_sequence() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
crate::sql_query(
"CREATE TABLE seq_test (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL)",
)
.execute(conn)
.unwrap();
let tables: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
let t2 = tables.clone();
conn.on_update(
SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |ev| {
t2.lock().unwrap().push(ev.table_name.to_owned());
}),
);
crate::sql_query("INSERT INTO seq_test (name) VALUES ('row1')")
.execute(conn)
.unwrap();
let recorded = tables.lock().unwrap();
assert!(
recorded.iter().all(|t| t == "seq_test"),
"expected only 'seq_test' events, got: {:?}",
*recorded
);
assert!(
!recorded.iter().any(|t| t == "sqlite_sequence"),
"sqlite_sequence modifications must not trigger the update hook"
);
}
#[diesel_test_helper::test]
fn update_hook_silent_for_replace_into_on_pk_conflict() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
crate::sql_query("CREATE TABLE rep (id INTEGER PRIMARY KEY, val TEXT NOT NULL)")
.execute(conn)
.unwrap();
crate::sql_query("INSERT INTO rep (id, val) VALUES (1, 'old')")
.execute(conn)
.unwrap();
let events: Arc<Mutex<Vec<(SqliteChangeOp, i64)>>> = Arc::new(Mutex::new(Vec::new()));
let e2 = events.clone();
conn.on_update(
SqliteUpdateRouter::new().on_any(SqliteChangeOps::ALL, move |ev| {
if ev.table_name == "rep" {
e2.lock().unwrap().push((ev.op, ev.rowid));
}
}),
);
crate::sql_query("REPLACE INTO rep (id, val) VALUES (1, 'new')")
.execute(conn)
.unwrap();
let recorded = events.lock().unwrap();
assert_eq!(
recorded.len(),
1,
"expected 1 event for REPLACE INTO, got: {:?}",
*recorded
);
assert_eq!(recorded[0].0, SqliteChangeOp::Insert);
assert_eq!(recorded[0].1, 1);
}
#[diesel_test_helper::test]
fn change_hook_fires_after_connection_move() {
use std::sync::{Arc, Mutex};
let count = Arc::new(Mutex::new(0u32));
let count2 = count.clone();
let mut conn = connection();
setup_hook_tables(&mut conn);
conn.on_update(SqliteUpdateRouter::new().on(
hook_users::table,
SqliteChangeOps::INSERT,
move |_| {
*count2.lock().unwrap() += 1;
},
));
let mut boxed = Box::new(conn);
crate::sql_query("INSERT INTO hook_users (name) VALUES ('Alice')")
.execute(&mut *boxed)
.unwrap();
assert_eq!(
*count.lock().unwrap(),
1,
"change hook did not fire after the connection was moved"
);
}
#[diesel_test_helper::test]
fn router_filters_table_and_op() {
use std::sync::{Arc, Mutex};
let conn = &mut connection();
setup_hook_tables(conn);
let fired = Arc::new(Mutex::new(Vec::new()));
let fired2 = fired.clone();
conn.on_update(SqliteUpdateRouter::new().on(
hook_users::table,
SqliteChangeOps::INSERT | SqliteChangeOps::UPDATE,
move |event| fired2.lock().unwrap().push(event.op),
));
crate::sql_query("INSERT INTO hook_users (name) VALUES ('Alice')")
.execute(conn)
.unwrap();
crate::sql_query("UPDATE hook_users SET name = 'Bob' WHERE id = 1")
.execute(conn)
.unwrap();
crate::sql_query("DELETE FROM hook_users WHERE id = 1")
.execute(conn)
.unwrap();
crate::sql_query("INSERT INTO hook_posts (title) VALUES ('Hello')")
.execute(conn)
.unwrap();
let events = fired.lock().unwrap().clone();
assert_eq!(events, vec![SqliteChangeOp::Insert, SqliteChangeOp::Update]);
}
}