test: ADR-0006 §8 step 7 — full-stack undo across DSL + SQL (R21/R22)

Three Tier-3 flows through the real worker:
- undo/redo steps back across interleaved DSL insert, SQL insert,
  and SQL delete — proving SQL DML snapshots like DSL (R22)
- undo restores the database read model AND the on-disk CSV
  together (consistent (db, csv) pair)
- the snapshot ring persists across a close + reopen of the project
  (undo works in a later session)

1696 passed / 0 failed / 1 ignored; clippy clean.

tests/undo_snapshots.rs

@@ -9,7 +9,7 @@
 use std::path::Path;
 
 use rdbms_playground::db::Database;
-use rdbms_playground::dsl::{ColumnSpec, RowFilter, Type, Value};
+use rdbms_playground::dsl::{ColumnSpec, Command, RowFilter, Type, Value, parse_command};
 use rdbms_playground::persistence::Persistence;
 use rdbms_playground::project;
 
@@ -234,3 +234,189 @@ fn empty_undo_and_redo_are_no_ops() {
         assert!(db.peek_redo().await.unwrap().is_none());
     });
 }
+
+// ---- Step 7: full-stack flow across DSL *and* SQL (R21 / R22) ----
+//
+// R22: the snapshot hook lives in the worker dispatch, so SQL DML
+// (SqlInsert/SqlUpdate/SqlDelete) is snapshotted exactly like DSL.
+
+async fn make_t(db: &Database) {
+    db.create_table(
+        "T".to_string(),
+        vec![
+            ColumnSpec::new("id".to_string(), Type::Int),
+            ColumnSpec::new("n".to_string(), Type::Int),
+        ],
+        vec!["id".to_string()],
+        Some("create table T with pk id(int), n(int)".to_string()),
+    )
+    .await
+    .unwrap();
+}
+
+async fn dsl_insert(db: &Database, id: i64, n: i64) {
+    db.insert(
+        "T".to_string(),
+        Some(vec!["id".to_string(), "n".to_string()]),
+        vec![Value::Number(id.to_string()), Value::Number(n.to_string())],
+        Some(format!("insert into T (id, n) values ({id}, {n})")),
+    )
+    .await
+    .unwrap();
+}
+
+async fn sql_insert(db: &Database, input: &str) {
+    match parse_command(input).unwrap() {
+        Command::SqlInsert {
+            sql,
+            target_table,
+            listed_columns,
+            row_source,
+            returning,
+        } => {
+            db.run_sql_insert(
+                sql,
+                Some(input.to_string()),
+                target_table,
+                listed_columns,
+                row_source,
+                returning,
+            )
+            .await
+            .unwrap();
+        }
+        other => panic!("expected SqlInsert from {input:?}, got {other:?}"),
+    }
+}
+
+async fn sql_delete(db: &Database, input: &str) {
+    match parse_command(input).unwrap() {
+        Command::SqlDelete {
+            sql,
+            target_table,
+            returning,
+        } => {
+            db.run_sql_delete(sql, Some(input.to_string()), target_table, returning)
+                .await
+                .unwrap();
+        }
+        other => panic!("expected SqlDelete from {input:?}, got {other:?}"),
+    }
+}
+
+async fn count_t(db: &Database) -> usize {
+    db.query_data("T".to_string(), None, None, None)
+        .await
+        .unwrap()
+        .rows
+        .len()
+}
+
+#[test]
+fn undo_steps_back_across_dsl_and_sql_mutations() {
+    let data = tempdir();
+    let (_p, db, _path) = open_project(&data, true);
+
+    rt().block_on(async {
+        make_t(&db).await; // snapshot 1: create
+        dsl_insert(&db, 1, 10).await; // snapshot 2: DSL insert
+        sql_insert(&db, "insert into T (id, n) values (2, 20)").await; // snapshot 3: SQL insert
+        assert_eq!(count_t(&db).await, 2);
+        sql_delete(&db, "delete from T where id = 1").await; // snapshot 4: SQL delete
+        assert_eq!(count_t(&db).await, 1);
+
+        // Walk back through SQL then DSL boundaries.
+        db.undo().await.unwrap();
+        assert_eq!(count_t(&db).await, 2, "SQL delete undone");
+        db.undo().await.unwrap();
+        assert_eq!(count_t(&db).await, 1, "SQL insert undone");
+        db.undo().await.unwrap();
+        assert_eq!(count_t(&db).await, 0, "DSL insert undone");
+
+        // Redo re-applies the DSL insert.
+        db.redo().await.unwrap();
+        assert_eq!(count_t(&db).await, 1, "DSL insert redone");
+    });
+}
+
+#[test]
+fn undo_restores_db_and_csv_consistently() {
+    let data = tempdir();
+    let (_p, db, path) = open_project(&data, true);
+
+    rt().block_on(async {
+        db.create_table(
+            "T".to_string(),
+            vec![
+                ColumnSpec::new("id".to_string(), Type::Int),
+                ColumnSpec::new("name".to_string(), Type::Text),
+            ],
+            vec!["id".to_string()],
+            Some("create table T".to_string()),
+        )
+        .await
+        .unwrap();
+        db.insert(
+            "T".to_string(),
+            Some(vec!["id".to_string(), "name".to_string()]),
+            vec![Value::Number("1".to_string()), Value::Text("Alice".to_string())],
+            Some("insert Alice".to_string()),
+        )
+        .await
+        .unwrap();
+        sql_insert(&db, "insert into T (id, name) values (2, 'Bob')").await;
+        sql_delete(&db, "delete from T where id = 2").await;
+
+        let csv = std::fs::read_to_string(path.join("data").join("T.csv")).unwrap();
+        assert!(
+            csv.contains("Alice") && !csv.contains("Bob"),
+            "post-delete csv: {csv}"
+        );
+
+        db.undo().await.unwrap();
+        // Both the database read model and the on-disk CSV are
+        // restored — the (db, csv) pair stays consistent.
+        assert_eq!(
+            db.query_data("T".to_string(), None, None, None)
+                .await
+                .unwrap()
+                .rows
+                .len(),
+            2
+        );
+        let csv2 = std::fs::read_to_string(path.join("data").join("T.csv")).unwrap();
+        assert!(csv2.contains("Bob"), "csv restored on disk: {csv2}");
+    });
+}
+
+#[test]
+fn undo_ring_persists_across_reopen() {
+    let data = tempdir();
+    let (project, db, path) = open_project(&data, true);
+    let db_path = project.db_path();
+
+    rt().block_on(async {
+        make_t(&db).await;
+        dsl_insert(&db, 1, 10).await;
+        sql_delete(&db, "delete from T where id = 1").await;
+        assert_eq!(count_t(&db).await, 0);
+    });
+
+    // Close the worker, then reopen the *same* project (lock still
+    // held by `project`). The persisted ring must survive.
+    drop(db);
+    let persistence = Persistence::new(path);
+    let db2 = Database::open_with_persistence_and_undo(&db_path, persistence, true)
+        .expect("reopen db");
+
+    rt().block_on(async {
+        let peek = db2
+            .peek_undo()
+            .await
+            .unwrap()
+            .expect("ring persisted across reopen");
+        assert_eq!(peek.command, "delete from T where id = 1");
+        db2.undo().await.unwrap();
+        assert_eq!(count_t(&db2).await, 1, "row restored after reopen + undo");
+    });
+}