drasi_source_sqlite/
thread.rs

1// Copyright 2025 The Drasi Authors.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use anyhow::{anyhow, Result};
16use base64::Engine;
17use rusqlite::hooks::{Action, PreUpdateCase};
18use rusqlite::types::ValueRef;
19use rusqlite::{Connection, OpenFlags};
20use serde_json::Value;
21use std::collections::{HashMap, HashSet};
22use std::sync::{Arc, Mutex};
23use tokio::sync::{mpsc, oneshot};
24
25pub type JsonRow = serde_json::Map<String, Value>;
26
27#[derive(Debug)]
28pub enum SqliteCommand {
29    Execute {
30        sql: String,
31        response_tx: oneshot::Sender<Result<usize>>,
32    },
33    ExecuteParameterized {
34        sql: String,
35        params: Vec<SqliteParam>,
36        response_tx: oneshot::Sender<Result<usize>>,
37    },
38    ExecuteBatch {
39        sql: String,
40        response_tx: oneshot::Sender<Result<()>>,
41    },
42    QueryRows {
43        sql: String,
44        response_tx: oneshot::Sender<Result<Vec<JsonRow>>>,
45    },
46    QueryRowsParameterized {
47        sql: String,
48        params: Vec<SqliteParam>,
49        response_tx: oneshot::Sender<Result<Vec<JsonRow>>>,
50    },
51    BeginTransaction {
52        response_tx: oneshot::Sender<Result<()>>,
53    },
54    CommitTransaction {
55        response_tx: oneshot::Sender<Result<()>>,
56    },
57    RollbackTransaction {
58        response_tx: oneshot::Sender<Result<()>>,
59    },
60    Shutdown,
61}
62
63/// A parameter value for parameterized SQL execution.
64#[derive(Debug, Clone)]
65pub enum SqliteParam {
66    Null,
67    Integer(i64),
68    Real(f64),
69    Text(String),
70    Bool(bool),
71}
72
73#[derive(Debug, Clone)]
74pub struct ChangeEvent {
75    pub action: Action,
76    pub table: String,
77    pub old_values: Option<Vec<(String, Value)>>,
78    pub new_values: Option<Vec<(String, Value)>>,
79    pub old_row_id: Option<i64>,
80    pub new_row_id: Option<i64>,
81    pub table_pk_columns: Vec<String>,
82    pub timestamp_ms: u64,
83}
84
85#[derive(Debug, Clone)]
86pub struct SqliteThreadConfig {
87    pub path: Option<String>,
88    pub tables: Option<Vec<String>>,
89}
90
91#[derive(Debug, Default, Clone)]
92struct TableSchema {
93    columns: Vec<String>,
94    pk_columns: Vec<String>,
95}
96
97#[derive(Debug, Default)]
98struct TransactionBuffer {
99    events: Vec<ChangeEvent>,
100    savepoint_markers: Vec<(String, usize)>,
101}
102
103impl TransactionBuffer {
104    fn push(&mut self, event: ChangeEvent) {
105        self.events.push(event);
106    }
107
108    fn begin_savepoint(&mut self, name: String) {
109        self.savepoint_markers.push((name, self.events.len()));
110    }
111
112    fn rollback_to_savepoint(&mut self, name: &str) {
113        if let Some(index) = self
114            .savepoint_markers
115            .iter()
116            .rposition(|(marker_name, _)| marker_name.eq_ignore_ascii_case(name))
117        {
118            let marker = self.savepoint_markers[index].1;
119            self.events.truncate(marker);
120            self.savepoint_markers.truncate(index + 1);
121        }
122    }
123
124    fn release_savepoint(&mut self, name: &str) {
125        if let Some(index) = self
126            .savepoint_markers
127            .iter()
128            .rposition(|(marker_name, _)| marker_name.eq_ignore_ascii_case(name))
129        {
130            self.savepoint_markers.remove(index);
131        }
132    }
133
134    fn take_all(&mut self) -> Vec<ChangeEvent> {
135        self.savepoint_markers.clear();
136        std::mem::take(&mut self.events)
137    }
138
139    fn clear(&mut self) {
140        self.events.clear();
141        self.savepoint_markers.clear();
142    }
143}
144
145#[derive(Debug, Clone)]
146enum SavepointCommand {
147    Savepoint(String),
148    RollbackTo(String),
149    Release(String),
150}
151
152pub fn run_sqlite_thread(
153    config: SqliteThreadConfig,
154    mut command_rx: mpsc::UnboundedReceiver<SqliteCommand>,
155    event_tx: mpsc::UnboundedSender<ChangeEvent>,
156) -> Result<()> {
157    let conn = if let Some(path) = config.path {
158        Connection::open_with_flags(
159            path,
160            OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_CREATE,
161        )?
162    } else {
163        Connection::open_in_memory()?
164    };
165
166    let allowed_tables = config
167        .tables
168        .map(|tables| tables.into_iter().collect::<HashSet<_>>());
169    let schema_cache: Arc<Mutex<HashMap<String, TableSchema>>> =
170        Arc::new(Mutex::new(HashMap::new()));
171    let tx_buffer = Arc::new(Mutex::new(TransactionBuffer::default()));
172
173    {
174        let mut cache = schema_cache
175            .lock()
176            .map_err(|_| anyhow!("schema cache poisoned"))?;
177        refresh_schema_cache(&conn, &mut cache, allowed_tables.as_ref())?;
178    }
179
180    let schema_cache_pre = Arc::clone(&schema_cache);
181    let tx_buffer_pre = Arc::clone(&tx_buffer);
182    let allowed_tables_pre = allowed_tables.clone();
183
184    conn.preupdate_hook(Some(
185        move |action: Action, db_name: &str, table_name: &str, case: &PreUpdateCase| {
186            if db_name != "main" {
187                return;
188            }
189
190            if let Some(allowed) = &allowed_tables_pre {
191                if !allowed.contains(table_name) {
192                    return;
193                }
194            }
195
196            let schema = schema_cache_pre
197                .lock()
198                .ok()
199                .and_then(|cache| cache.get(table_name).cloned())
200                .unwrap_or_default();
201
202            let mut old_values: Option<Vec<(String, Value)>> = None;
203            let mut new_values: Option<Vec<(String, Value)>> = None;
204            let mut old_row_id = None;
205            let mut new_row_id = None;
206
207            match case {
208                PreUpdateCase::Insert(accessor) => {
209                    let mut values = Vec::new();
210                    for idx in 0..accessor.get_column_count() {
211                        if let Ok(v) = accessor.get_new_column_value(idx) {
212                            values.push((column_name(&schema, idx), value_ref_to_json(v)));
213                        }
214                    }
215                    new_values = Some(values);
216                    new_row_id = Some(accessor.get_new_row_id());
217                }
218                PreUpdateCase::Delete(accessor) => {
219                    let mut values = Vec::new();
220                    for idx in 0..accessor.get_column_count() {
221                        if let Ok(v) = accessor.get_old_column_value(idx) {
222                            values.push((column_name(&schema, idx), value_ref_to_json(v)));
223                        }
224                    }
225                    old_values = Some(values);
226                    old_row_id = Some(accessor.get_old_row_id());
227                }
228                PreUpdateCase::Update {
229                    old_value_accessor,
230                    new_value_accessor,
231                } => {
232                    let mut old = Vec::new();
233                    let mut new = Vec::new();
234                    for idx in 0..old_value_accessor.get_column_count() {
235                        let name = column_name(&schema, idx);
236                        if let Ok(v) = old_value_accessor.get_old_column_value(idx) {
237                            old.push((name.clone(), value_ref_to_json(v)));
238                        }
239                        if let Ok(v) = new_value_accessor.get_new_column_value(idx) {
240                            new.push((name, value_ref_to_json(v)));
241                        }
242                    }
243                    old_values = Some(old);
244                    new_values = Some(new);
245                    old_row_id = Some(old_value_accessor.get_old_row_id());
246                    new_row_id = Some(new_value_accessor.get_new_row_id());
247                }
248                PreUpdateCase::Unknown => {}
249            }
250
251            let event = ChangeEvent {
252                action,
253                table: table_name.to_string(),
254                old_values,
255                new_values,
256                old_row_id,
257                new_row_id,
258                table_pk_columns: schema.pk_columns,
259                timestamp_ms: chrono::Utc::now().timestamp_millis() as u64,
260            };
261
262            if let Ok(mut buffer) = tx_buffer_pre.lock() {
263                buffer.push(event);
264            }
265        },
266    ));
267
268    let tx_buffer_commit = Arc::clone(&tx_buffer);
269    let event_tx_commit = event_tx.clone();
270    conn.commit_hook(Some(move || {
271        if let Ok(mut buffer) = tx_buffer_commit.lock() {
272            for event in buffer.take_all() {
273                let _ = event_tx_commit.send(event);
274            }
275        }
276        false
277    }));
278
279    let tx_buffer_rollback = Arc::clone(&tx_buffer);
280    conn.rollback_hook(Some(move || {
281        if let Ok(mut buffer) = tx_buffer_rollback.lock() {
282            buffer.clear();
283        }
284    }));
285
286    while let Some(command) = command_rx.blocking_recv() {
287        match command {
288            SqliteCommand::Execute { sql, response_tx } => {
289                let savepoint_command = parse_savepoint_command(sql.as_str());
290                let result = conn.execute(sql.as_str(), []).map_err(anyhow::Error::from);
291                if result.is_ok() {
292                    if let Some(command) = savepoint_command {
293                        apply_savepoint_command(&tx_buffer, command);
294                    }
295                    let _ = refresh_schema_cache_for_runtime(
296                        &conn,
297                        &schema_cache,
298                        allowed_tables.as_ref(),
299                    );
300                }
301                let _ = response_tx.send(result);
302            }
303            SqliteCommand::ExecuteParameterized {
304                sql,
305                params,
306                response_tx,
307            } => {
308                let rusqlite_params: Vec<Box<dyn rusqlite::types::ToSql>> =
309                    params.iter().map(sqlite_param_to_rusqlite).collect();
310                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
311                    rusqlite_params.iter().map(|p| p.as_ref()).collect();
312                let result = conn
313                    .execute(sql.as_str(), param_refs.as_slice())
314                    .map_err(anyhow::Error::from);
315                if result.is_ok() {
316                    let _ = refresh_schema_cache_for_runtime(
317                        &conn,
318                        &schema_cache,
319                        allowed_tables.as_ref(),
320                    );
321                }
322                let _ = response_tx.send(result);
323            }
324            SqliteCommand::ExecuteBatch { sql, response_tx } => {
325                let result = conn
326                    .execute_batch(sql.as_str())
327                    .map_err(anyhow::Error::from);
328                if result.is_ok() {
329                    let _ = refresh_schema_cache_for_runtime(
330                        &conn,
331                        &schema_cache,
332                        allowed_tables.as_ref(),
333                    );
334                }
335                let _ = response_tx.send(result);
336            }
337            SqliteCommand::QueryRows { sql, response_tx } => {
338                let _ = response_tx.send(query_rows(&conn, sql.as_str(), &[]));
339            }
340            SqliteCommand::QueryRowsParameterized {
341                sql,
342                params,
343                response_tx,
344            } => {
345                let rusqlite_params: Vec<Box<dyn rusqlite::types::ToSql>> =
346                    params.iter().map(sqlite_param_to_rusqlite).collect();
347                let param_refs: Vec<&dyn rusqlite::types::ToSql> =
348                    rusqlite_params.iter().map(|p| p.as_ref()).collect();
349                let _ = response_tx.send(query_rows(&conn, sql.as_str(), param_refs.as_slice()));
350            }
351            SqliteCommand::BeginTransaction { response_tx } => {
352                let result = conn
353                    .execute("BEGIN TRANSACTION", [])
354                    .map(|_| ())
355                    .map_err(anyhow::Error::from);
356                let _ = response_tx.send(result);
357            }
358            SqliteCommand::CommitTransaction { response_tx } => {
359                let result = conn
360                    .execute("COMMIT", [])
361                    .map(|_| ())
362                    .map_err(anyhow::Error::from);
363                let _ = response_tx.send(result);
364            }
365            SqliteCommand::RollbackTransaction { response_tx } => {
366                let result = conn
367                    .execute("ROLLBACK", [])
368                    .map(|_| ())
369                    .map_err(anyhow::Error::from);
370                let _ = response_tx.send(result);
371            }
372            SqliteCommand::Shutdown => break,
373        }
374    }
375
376    Ok(())
377}
378
379fn refresh_schema_cache_for_runtime(
380    conn: &Connection,
381    schema_cache: &Arc<Mutex<HashMap<String, TableSchema>>>,
382    allowed_tables: Option<&HashSet<String>>,
383) -> Result<()> {
384    let mut cache = schema_cache
385        .lock()
386        .map_err(|_| anyhow!("schema cache poisoned"))?;
387    refresh_schema_cache(conn, &mut cache, allowed_tables)
388}
389
390fn refresh_schema_cache(
391    conn: &Connection,
392    cache: &mut HashMap<String, TableSchema>,
393    allowed_tables: Option<&HashSet<String>>,
394) -> Result<()> {
395    let mut stmt = conn.prepare(
396        "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name",
397    )?;
398    let table_names = stmt
399        .query_map([], |row| row.get::<_, String>(0))?
400        .collect::<std::result::Result<Vec<_>, _>>()?;
401
402    for table in table_names {
403        if let Some(allowed) = allowed_tables {
404            if !allowed.contains(&table) {
405                continue;
406            }
407        }
408        let schema = load_table_schema(conn, &table)?;
409        cache.insert(table, schema);
410    }
411    Ok(())
412}
413
414fn load_table_schema(conn: &Connection, table: &str) -> Result<TableSchema> {
415    let escaped = table.replace('\'', "''");
416    let sql = format!("PRAGMA table_info('{escaped}')");
417    let mut stmt = conn.prepare(sql.as_str())?;
418
419    let mut columns = Vec::new();
420    let mut pk_pairs: Vec<(i64, String)> = Vec::new();
421
422    let mut rows = stmt.query([])?;
423    while let Some(row) = rows.next()? {
424        let name: String = row.get(1)?;
425        let pk_order: i64 = row.get(5)?;
426        columns.push(name.clone());
427        if pk_order > 0 {
428            pk_pairs.push((pk_order, name));
429        }
430    }
431
432    pk_pairs.sort_by_key(|(order, _)| *order);
433    let pk_columns = pk_pairs
434        .into_iter()
435        .map(|(_, name)| name)
436        .collect::<Vec<_>>();
437
438    Ok(TableSchema {
439        columns,
440        pk_columns,
441    })
442}
443
444fn column_name(schema: &TableSchema, index: i32) -> String {
445    schema
446        .columns
447        .get(index as usize)
448        .cloned()
449        .unwrap_or_else(|| {
450            log::warn!(
451                "Column index {index} out of range (schema has {} columns), using fallback name",
452                schema.columns.len()
453            );
454            format!("col_{index}")
455        })
456}
457
458fn value_ref_to_json(value: ValueRef<'_>) -> Value {
459    match value {
460        ValueRef::Null => Value::Null,
461        ValueRef::Integer(v) => Value::Number(v.into()),
462        ValueRef::Real(v) => serde_json::Number::from_f64(v)
463            .map(Value::Number)
464            .unwrap_or(Value::Null),
465        ValueRef::Text(v) => Value::String(String::from_utf8_lossy(v).to_string()),
466        ValueRef::Blob(v) => Value::String(base64::engine::general_purpose::STANDARD.encode(v)),
467    }
468}
469
470fn sqlite_param_to_rusqlite(p: &SqliteParam) -> Box<dyn rusqlite::types::ToSql> {
471    match p {
472        SqliteParam::Null => Box::new(rusqlite::types::Null) as Box<dyn rusqlite::types::ToSql>,
473        SqliteParam::Integer(v) => Box::new(*v),
474        SqliteParam::Real(v) => Box::new(*v),
475        SqliteParam::Text(v) => Box::new(v.clone()),
476        SqliteParam::Bool(v) => Box::new(*v),
477    }
478}
479
480fn query_rows(
481    conn: &Connection,
482    sql: &str,
483    params: &[&dyn rusqlite::types::ToSql],
484) -> Result<Vec<JsonRow>> {
485    let mut stmt = conn.prepare(sql)?;
486    let columns = stmt
487        .column_names()
488        .iter()
489        .map(|name| (*name).to_string())
490        .collect::<Vec<_>>();
491
492    let mut rows = stmt.query(params)?;
493    let mut result = Vec::new();
494    while let Some(row) = rows.next()? {
495        let mut item = serde_json::Map::new();
496        for (index, name) in columns.iter().enumerate() {
497            let value = row.get_ref(index)?;
498            item.insert(name.clone(), value_ref_to_json(value));
499        }
500        result.push(item);
501    }
502    Ok(result)
503}
504
505fn parse_savepoint_command(sql: &str) -> Option<SavepointCommand> {
506    let trimmed = sql.trim().trim_end_matches(';').trim();
507    if trimmed.is_empty() {
508        return None;
509    }
510
511    let lower = trimmed.to_ascii_lowercase();
512    if let Some(name) = lower.strip_prefix("savepoint ") {
513        return extract_identifier(trimmed, name.len()).map(SavepointCommand::Savepoint);
514    }
515    if let Some(name) = lower.strip_prefix("rollback to savepoint ") {
516        return extract_identifier(trimmed, name.len()).map(SavepointCommand::RollbackTo);
517    }
518    if let Some(name) = lower.strip_prefix("rollback to ") {
519        return extract_identifier(trimmed, name.len()).map(SavepointCommand::RollbackTo);
520    }
521    if let Some(name) = lower.strip_prefix("release savepoint ") {
522        return extract_identifier(trimmed, name.len()).map(SavepointCommand::Release);
523    }
524    if let Some(name) = lower.strip_prefix("release ") {
525        return extract_identifier(trimmed, name.len()).map(SavepointCommand::Release);
526    }
527    None
528}
529
530fn extract_identifier(original_sql: &str, suffix_len: usize) -> Option<String> {
531    let sql = original_sql.trim().trim_end_matches(';').trim();
532    let start = sql.len().checked_sub(suffix_len)?;
533    let remainder = sql.get(start..)?.trim();
534    let token = remainder
535        .split_whitespace()
536        .next()
537        .map(|value| value.trim_matches('"').trim_matches('\'').to_string())?;
538    if token.is_empty() {
539        None
540    } else {
541        Some(token)
542    }
543}
544
545fn apply_savepoint_command(tx_buffer: &Arc<Mutex<TransactionBuffer>>, command: SavepointCommand) {
546    if let Ok(mut buffer) = tx_buffer.lock() {
547        match command {
548            SavepointCommand::Savepoint(name) => buffer.begin_savepoint(name),
549            SavepointCommand::RollbackTo(name) => buffer.rollback_to_savepoint(&name),
550            SavepointCommand::Release(name) => buffer.release_savepoint(&name),
551        }
552    }
553}
554
555#[cfg(test)]
556mod tests {
557    use super::*;
558    use tokio::time::{timeout, Duration};
559
560    fn start_thread() -> (
561        mpsc::UnboundedSender<SqliteCommand>,
562        mpsc::UnboundedReceiver<ChangeEvent>,
563        std::thread::JoinHandle<()>,
564    ) {
565        let (command_tx, command_rx) = mpsc::unbounded_channel();
566        let (event_tx, event_rx) = mpsc::unbounded_channel();
567        let join_handle = std::thread::spawn(move || {
568            run_sqlite_thread(
569                SqliteThreadConfig {
570                    path: None,
571                    tables: None,
572                },
573                command_rx,
574                event_tx,
575            )
576            .expect("sqlite thread failed");
577        });
578        (command_tx, event_rx, join_handle)
579    }
580
581    async fn exec(command_tx: &mpsc::UnboundedSender<SqliteCommand>, sql: &str) -> Result<usize> {
582        let (response_tx, response_rx) = oneshot::channel();
583        command_tx
584            .send(SqliteCommand::Execute {
585                sql: sql.to_string(),
586                response_tx,
587            })
588            .expect("failed to send execute command");
589        response_rx.await.expect("response channel closed")
590    }
591
592    async fn begin_tx(command_tx: &mpsc::UnboundedSender<SqliteCommand>) {
593        let (response_tx, response_rx) = oneshot::channel();
594        command_tx
595            .send(SqliteCommand::BeginTransaction { response_tx })
596            .expect("failed to send begin transaction");
597        response_rx
598            .await
599            .expect("begin response closed")
600            .expect("begin failed");
601    }
602
603    async fn commit_tx(command_tx: &mpsc::UnboundedSender<SqliteCommand>) {
604        let (response_tx, response_rx) = oneshot::channel();
605        command_tx
606            .send(SqliteCommand::CommitTransaction { response_tx })
607            .expect("failed to send commit transaction");
608        response_rx
609            .await
610            .expect("commit response closed")
611            .expect("commit failed");
612    }
613
614    async fn rollback_tx(command_tx: &mpsc::UnboundedSender<SqliteCommand>) {
615        let (response_tx, response_rx) = oneshot::channel();
616        command_tx
617            .send(SqliteCommand::RollbackTransaction { response_tx })
618            .expect("failed to send rollback transaction");
619        response_rx
620            .await
621            .expect("rollback response closed")
622            .expect("rollback failed");
623    }
624
625    #[tokio::test]
626    async fn committed_transaction_dispatches_events() {
627        let (command_tx, mut event_rx, join_handle) = start_thread();
628
629        exec(
630            &command_tx,
631            "CREATE TABLE sensors(id INTEGER PRIMARY KEY, name TEXT)",
632        )
633        .await
634        .expect("create table failed");
635
636        begin_tx(&command_tx).await;
637        exec(&command_tx, "INSERT INTO sensors(id, name) VALUES (1, 'a')")
638            .await
639            .expect("insert 1 failed");
640        exec(&command_tx, "INSERT INTO sensors(id, name) VALUES (2, 'b')")
641            .await
642            .expect("insert 2 failed");
643        commit_tx(&command_tx).await;
644
645        let first = timeout(Duration::from_secs(2), event_rx.recv())
646            .await
647            .expect("timeout waiting first event")
648            .expect("missing first event");
649        let second = timeout(Duration::from_secs(2), event_rx.recv())
650            .await
651            .expect("timeout waiting second event")
652            .expect("missing second event");
653
654        assert_eq!(first.action, Action::SQLITE_INSERT);
655        assert_eq!(second.action, Action::SQLITE_INSERT);
656
657        command_tx
658            .send(SqliteCommand::Shutdown)
659            .expect("failed to send shutdown");
660        join_handle.join().expect("thread join failed");
661    }
662
663    #[tokio::test]
664    async fn rolled_back_transaction_discards_events() {
665        let (command_tx, mut event_rx, join_handle) = start_thread();
666
667        exec(
668            &command_tx,
669            "CREATE TABLE sensors(id INTEGER PRIMARY KEY, name TEXT)",
670        )
671        .await
672        .expect("create table failed");
673
674        begin_tx(&command_tx).await;
675        exec(&command_tx, "INSERT INTO sensors(id, name) VALUES (1, 'a')")
676            .await
677            .expect("insert failed");
678        rollback_tx(&command_tx).await;
679
680        let receive = timeout(Duration::from_millis(200), event_rx.recv()).await;
681        assert!(receive.is_err(), "expected no event after rollback");
682
683        command_tx
684            .send(SqliteCommand::Shutdown)
685            .expect("failed to send shutdown");
686        join_handle.join().expect("thread join failed");
687    }
688
689    #[tokio::test]
690    async fn rollback_to_savepoint_discards_partial_events() {
691        let (command_tx, mut event_rx, join_handle) = start_thread();
692
693        exec(
694            &command_tx,
695            "CREATE TABLE sensors(id INTEGER PRIMARY KEY, name TEXT)",
696        )
697        .await
698        .expect("create table failed");
699
700        begin_tx(&command_tx).await;
701        exec(&command_tx, "INSERT INTO sensors(id, name) VALUES (1, 'a')")
702            .await
703            .expect("insert 1 failed");
704        exec(&command_tx, "SAVEPOINT s1")
705            .await
706            .expect("savepoint failed");
707        exec(&command_tx, "INSERT INTO sensors(id, name) VALUES (2, 'b')")
708            .await
709            .expect("insert 2 failed");
710        exec(&command_tx, "ROLLBACK TO SAVEPOINT s1")
711            .await
712            .expect("rollback savepoint failed");
713        exec(&command_tx, "INSERT INTO sensors(id, name) VALUES (3, 'c')")
714            .await
715            .expect("insert 3 failed");
716        commit_tx(&command_tx).await;
717
718        let first = timeout(Duration::from_secs(2), event_rx.recv())
719            .await
720            .expect("timeout waiting first event")
721            .expect("missing first event");
722        let second = timeout(Duration::from_secs(2), event_rx.recv())
723            .await
724            .expect("timeout waiting second event")
725            .expect("missing second event");
726        let third = timeout(Duration::from_millis(200), event_rx.recv()).await;
727
728        let first_id = first
729            .new_values
730            .as_ref()
731            .and_then(|values| values.iter().find(|(k, _)| k == "id").map(|(_, v)| v))
732            .cloned()
733            .unwrap_or_default();
734        let second_id = second
735            .new_values
736            .as_ref()
737            .and_then(|values| values.iter().find(|(k, _)| k == "id").map(|(_, v)| v))
738            .cloned()
739            .unwrap_or_default();
740
741        assert_eq!(first_id, serde_json::json!(1));
742        assert_eq!(second_id, serde_json::json!(3));
743        assert!(third.is_err(), "expected no third committed event");
744
745        command_tx
746            .send(SqliteCommand::Shutdown)
747            .expect("failed to send shutdown");
748        join_handle.join().expect("thread join failed");
749    }
750}
drasi_source_sqlite/thread.rs

drasi_source_sqlite/
thread.rs
