wasm-dbms 0.10.0

// Rust guideline compliant 2026-02-28

//! Transaction management for the DBMS engine.

pub(crate) mod journal;
mod overlay;
pub mod session;

use wasm_dbms_api::prelude::{
    ColumnDef, DbmsResult, DeleteBehavior, Filter, TableSchema, UpdateRecord as _, Value,
};

pub(crate) use self::overlay::TableOverlay;
pub use self::overlay::{DatabaseOverlay, IndexOverlay};

/// A transaction represents a sequence of operations performed as a single
/// logical unit of work.
#[derive(Debug, Default)]
pub struct Transaction {
    /// Stack of operations performed in this transaction.
    pub(crate) operations: Vec<TransactionOp>,
    /// Overlay to track uncommitted changes.
    overlay: DatabaseOverlay,
}

impl Transaction {
    /// Inserts a new insert operation into the transaction.
    pub fn insert<T>(&mut self, values: Vec<(ColumnDef, Value)>) -> DbmsResult<()>
    where
        T: TableSchema,
    {
        self.overlay.insert::<T>(values.clone())?;
        self.operations.push(TransactionOp::Insert {
            table: T::table_name(),
            values,
        });
        Ok(())
    }

    /// Inserts a new update operation into the transaction.
    ///
    /// `rows` is a list of `(primary_key, current_row)` pairs for each affected record.
    /// The current row is needed to track old indexed values in the overlay.
    ///
    /// The operation recorded for commit is restricted to the complete row versions in `rows`,
    /// so that commit updates exactly the rows captured when the update was staged. A row that
    /// changed or was replaced after staging no longer matches and causes a commit conflict.
    /// See <https://github.com/veeso/wasm-dbms/issues/118>.
    pub fn update<T>(
        &mut self,
        patch: T::Update,
        rows: Vec<(Value, Vec<(ColumnDef, Value)>)>,
    ) -> DbmsResult<()>
    where
        T: TableSchema,
    {
        let patch_values = patch.update_values();
        let overlay_patch: Vec<_> = patch_values
            .iter()
            .map(|(col, val)| (col.name, val.clone()))
            .collect();

        let mut filters = Vec::with_capacity(rows.len());
        for (pk, current_row) in rows {
            self.overlay
                .update::<T>(pk.clone(), overlay_patch.clone(), &current_row);
            filters.push(Self::captured_row_filter::<T>(pk, &current_row));
        }

        self.operations.push(TransactionOp::Update {
            table: T::table_name(),
            patch: patch_values,
            filters,
        });
        Ok(())
    }

    /// Inserts a new delete operation into the transaction.
    ///
    /// `rows` is a list of `(primary_key, current_row)` pairs for each affected record.
    /// The current row is needed to track removed indexed values in the overlay.
    ///
    /// The operation recorded for commit is restricted to the complete row versions in `rows`,
    /// so that commit deletes exactly the rows captured when the delete was staged. A row that
    /// changed or was replaced after staging no longer matches and causes a commit conflict.
    /// See <https://github.com/veeso/wasm-dbms/issues/118>.
    pub fn delete<T>(
        &mut self,
        behaviour: DeleteBehavior,
        rows: Vec<(Value, Vec<(ColumnDef, Value)>)>,
    ) -> DbmsResult<()>
    where
        T: TableSchema,
    {
        let mut filters = Vec::with_capacity(rows.len());
        for (pk, current_row) in rows {
            self.overlay.delete::<T>(pk.clone(), &current_row);
            filters.push(Self::captured_row_filter::<T>(pk, &current_row));
        }

        self.operations.push(TransactionOp::Delete {
            table: T::table_name(),
            behaviour,
            filters,
        });
        Ok(())
    }

    /// Builds an optimistic-concurrency filter for one captured row version.
    fn captured_row_filter<T>(primary_key: Value, row: &[(ColumnDef, Value)]) -> Filter
    where
        T: TableSchema,
    {
        row.iter()
            .filter(|(column, _)| column.name != T::primary_key())
            .fold(
                Filter::eq(T::primary_key(), primary_key),
                |filter, (column, value)| filter.and(Filter::eq(column.name, value.clone())),
            )
    }

    /// Returns a reference to the overlay.
    pub fn overlay(&self) -> &DatabaseOverlay {
        &self.overlay
    }

    /// Returns a mutable reference to the overlay.
    pub fn overlay_mut(&mut self) -> &mut DatabaseOverlay {
        &mut self.overlay
    }
}

/// An operation within a transaction.
#[derive(Debug)]
pub enum TransactionOp {
    Insert {
        table: &'static str,
        values: Vec<(ColumnDef, Value)>,
    },
    Delete {
        table: &'static str,
        behaviour: DeleteBehavior,
        filters: Vec<Filter>,
    },
    Update {
        table: &'static str,
        patch: Vec<(ColumnDef, Value)>,
        filters: Vec<Filter>,
    },
}

#[cfg(test)]
mod tests {

    use wasm_dbms_api::prelude::{Database as _, InsertRecord as _, Query, Text, Uint32, Value};
    use wasm_dbms_macros::{DatabaseSchema, Table};
    use wasm_dbms_memory::prelude::HeapMemoryProvider;

    use super::*;
    use crate::prelude::{DbmsContext, WasmDbmsDatabase};

    #[derive(Debug, Table, Clone, PartialEq, Eq)]
    #[table = "items"]
    pub struct Item {
        #[primary_key]
        pub id: Uint32,
        pub name: Text,
    }

    #[derive(DatabaseSchema)]
    #[tables(Item = "items")]
    pub struct TestSchema;

    fn setup() -> DbmsContext<HeapMemoryProvider> {
        let ctx = DbmsContext::new(HeapMemoryProvider::default());
        TestSchema::register_tables(&ctx).unwrap();
        ctx
    }

    #[test]
    fn test_transaction_insert_records_operation() {
        let mut tx = Transaction::default();
        let values = vec![
            (Item::columns()[0], Value::Uint32(Uint32(1))),
            (Item::columns()[1], Value::Text(Text("foo".to_string()))),
        ];
        tx.insert::<Item>(values).unwrap();
        assert_eq!(tx.operations.len(), 1);
        assert!(matches!(
            &tx.operations[0],
            TransactionOp::Insert { table: "items", .. }
        ));
    }

    #[test]
    fn test_transaction_update_records_operation() {
        let mut tx = Transaction::default();
        let patch = ItemUpdateRequest::from_values(
            &[(Item::columns()[1], Value::Text(Text("bar".to_string())))],
            Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        )
        .unwrap();
        let current_row = vec![
            (Item::columns()[0], Value::Uint32(Uint32(1))),
            (Item::columns()[1], Value::Text(Text("foo".to_string()))),
        ];
        tx.update::<Item>(patch, vec![(Value::Uint32(Uint32(1)), current_row)])
            .unwrap();
        assert_eq!(tx.operations.len(), 1);
        assert!(matches!(
            &tx.operations[0],
            TransactionOp::Update { table: "items", .. }
        ));
    }

    #[test]
    fn test_transaction_update_restricts_commit_filters_to_captured_row_versions() {
        let mut tx = Transaction::default();
        let patch = ItemUpdateRequest::from_values(
            &[(Item::columns()[1], Value::Text(Text("bar".to_string())))],
            Some(Filter::eq("name", Value::Text(Text("foo".to_string())))),
        )
        .unwrap();
        let rows = vec![
            (
                Value::Uint32(Uint32(1)),
                vec![
                    (Item::columns()[0], Value::Uint32(Uint32(1))),
                    (Item::columns()[1], Value::Text(Text("foo".to_string()))),
                ],
            ),
            (
                Value::Uint32(Uint32(3)),
                vec![
                    (Item::columns()[0], Value::Uint32(Uint32(3))),
                    (Item::columns()[1], Value::Text(Text("foo".to_string()))),
                ],
            ),
        ];
        tx.update::<Item>(patch, rows).unwrap();

        let TransactionOp::Update { filters, .. } = &tx.operations[0] else {
            panic!("expected an update operation");
        };
        assert_eq!(
            filters,
            &vec![
                Filter::eq("id", Value::Uint32(Uint32(1)))
                    .and(Filter::eq("name", Value::Text(Text("foo".to_string())),)),
                Filter::eq("id", Value::Uint32(Uint32(3)))
                    .and(Filter::eq("name", Value::Text(Text("foo".to_string())),)),
            ]
        );
    }

    #[test]
    fn test_transaction_delete_records_operation() {
        let mut tx = Transaction::default();
        let current_row = vec![
            (Item::columns()[0], Value::Uint32(Uint32(1))),
            (Item::columns()[1], Value::Text(Text("foo".to_string()))),
        ];
        tx.delete::<Item>(
            DeleteBehavior::Restrict,
            vec![(Value::Uint32(Uint32(1)), current_row)],
        )
        .unwrap();
        assert_eq!(tx.operations.len(), 1);
        let TransactionOp::Delete {
            table,
            behaviour,
            filters,
        } = &tx.operations[0]
        else {
            panic!("expected a delete operation");
        };
        assert_eq!(*table, "items");
        assert!(matches!(behaviour, DeleteBehavior::Restrict));
        assert_eq!(
            filters,
            &vec![
                Filter::eq("id", Value::Uint32(Uint32(1)))
                    .and(Filter::eq("name", Value::Text(Text("foo".to_string())),))
            ]
        );
    }

    #[test]
    fn test_transaction_overlay_accessors() {
        let mut tx = Transaction::default();
        // Overlay should start empty
        let overlay = tx.overlay();
        let overlay_str = format!("{overlay:?}");
        assert!(overlay_str.contains("DatabaseOverlay"));

        let _overlay_mut = tx.overlay_mut();
    }

    #[test]
    fn test_transaction_multiple_operations() {
        let mut tx = Transaction::default();
        let insert_values = vec![
            (Item::columns()[0], Value::Uint32(Uint32(1))),
            (Item::columns()[1], Value::Text(Text("a".to_string()))),
        ];
        tx.insert::<Item>(insert_values.clone()).unwrap();
        tx.delete::<Item>(
            DeleteBehavior::Cascade,
            vec![(Value::Uint32(Uint32(1)), insert_values)],
        )
        .unwrap();
        assert_eq!(tx.operations.len(), 2);
    }

    #[test]
    fn test_rollback_discards_transaction() {
        let ctx = setup();
        let tx_id = ctx.begin_transaction();
        let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);

        let insert = ItemInsertRequest::from_values(&[
            (Item::columns()[0], Value::Uint32(Uint32(42))),
            (
                Item::columns()[1],
                Value::Text(Text("rolled_back".to_string())),
            ),
        ])
        .unwrap();
        db.insert::<Item>(insert).unwrap();

        db.rollback().unwrap();

        // After rollback, the record should not exist
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        let rows = db.select::<Item>(Query::builder().build()).unwrap();
        assert!(rows.is_empty());
    }

    #[test]
    fn test_rollback_without_transaction_returns_error() {
        let ctx = setup();
        let mut db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        let result = db.rollback();
        assert!(result.is_err());
    }
}