sqlitegraph 2.0.7

//! Node operation handlers for V2 WAL recovery replay
//!
//! This module provides node-specific replay operations including
//! node insertion, update, and deletion with proper rollback support.

use crate::backend::native::v2::wal::recovery::errors::RecoveryError;
use crate::backend::native::v2::wal::recovery::replayer::types::RollbackOperation;
use crate::backend::native::v2::{
    EdgeCluster, FreeSpaceManager, NodeRecordV2, free_space::AllocationStrategy,
};
use crate::backend::native::{EdgeStore, NativeNodeId, NodeStore, adjacency::Direction};
use crate::debug::{debug_log, warn_log};

impl super::DefaultReplayOperations {
    /// Handle node insertion during replay
    pub fn handle_node_insert(
        &self,
        node_id: u64,
        slot_offset: u64,
        node_data: &[u8],
        rollback_data: &mut Vec<RollbackOperation>,
    ) -> Result<(), RecoveryError> {
        debug_log!(
            "Replaying node insert: node_id={}, slot_offset={}, data_size={}",
            node_id,
            slot_offset,
            node_data.len()
        );

        // Deserialize the node data
        let node_record =
            crate::backend::native::v2::node_record_v2::NodeRecordV2::deserialize(node_data)
                .map_err(|e| {
                    RecoveryError::io_error(format!("Failed to deserialize node data: {}", e))
                })?;

        // Validate node ID consistency
        if node_record.id != node_id as crate::backend::native::NativeNodeId {
            return Err(RecoveryError::validation(format!(
                "Node ID mismatch: expected {}, got {}",
                node_id, node_record.id
            )));
        }

        // Add rollback operation before making changes
        rollback_data.push(RollbackOperation::NodeInsert {
            node_id: node_id as crate::backend::native::NativeNodeId,
            node_data: node_data.to_vec(),
        });

        // Create NodeStore for this operation following proper SME methodology
        {
            let mut graph_file = self.graph_file.write().map_err(|e| {
                RecoveryError::io_error(format!("Failed to lock graph file: {}", e))
            })?;

            let mut node_store = NodeStore::new(&mut *graph_file);
            node_store
                .write_node_v2(&node_record)
                .map_err(|e| RecoveryError::io_error(format!("Failed to write node: {}", e)))?;
        } // graph_file lock and node_store are released here

        // Update statistics (lock-free)
        self.statistics.record_node_operation();
        self.statistics.record_bytes_written(node_data.len() as u64);

        debug_log!("Successfully replayed node insert: node_id={}", node_id);
        Ok(())
    }

    /// Handle node update during replay
    pub fn handle_node_update(
        &self,
        node_id: u64,
        _slot_offset: u64,
        new_data: &[u8],
        old_data: Option<&Vec<u8>>,
        rollback_data: &mut Vec<RollbackOperation>,
    ) -> Result<(), RecoveryError> {
        debug_log!(
            "Replaying node update: node_id={}, data_size={}",
            node_id,
            new_data.len()
        );

        // Validate input data
        if new_data.is_empty() {
            return Err(RecoveryError::validation(
                "Node update data cannot be empty".to_string(),
            ));
        }

        // Deserialize the new node data
        let node_record =
            crate::backend::native::v2::node_record_v2::NodeRecordV2::deserialize(new_data)
                .map_err(|e| {
                    RecoveryError::io_error(format!("Failed to deserialize node data: {}", e))
                })?;

        // Validate node ID consistency
        if node_record.id != node_id as crate::backend::native::NativeNodeId {
            return Err(RecoveryError::validation(format!(
                "Node ID mismatch: expected {}, got {}",
                node_id, node_record.id
            )));
        }

        // Add rollback operation before making changes
        if let Some(old_data_vec) = old_data {
            rollback_data.push(RollbackOperation::NodeUpdate {
                node_id: node_id as crate::backend::native::NativeNodeId,
                old_data: old_data_vec.clone(),
            });
        }

        // Create NodeStore for this operation following proper SME methodology
        {
            let mut graph_file = self.graph_file.write().map_err(|e| {
                RecoveryError::io_error(format!("Failed to lock graph file: {}", e))
            })?;

            let mut node_store = NodeStore::new(&mut *graph_file);
            node_store.write_node_v2(&node_record).map_err(|e| {
                RecoveryError::io_error(format!("Failed to write node update: {}", e))
            })?;
        } // graph_file lock and node_store are released here

        // Update statistics (lock-free)
        self.statistics.record_node_operation();
        self.statistics.record_bytes_written(new_data.len() as u64);

        debug_log!("Successfully replayed node update: node_id={}", node_id);
        Ok(())
    }

    /// Handle node deletion during replay
    pub fn handle_node_delete(
        &self,
        node_id: u64,
        slot_offset: u64,
        old_data: Option<&Vec<u8>>,
        rollback_data: &mut Vec<RollbackOperation>,
    ) -> Result<(), RecoveryError> {
        debug_log!(
            "Replaying node delete: node_id={}, slot_offset={}",
            node_id,
            slot_offset
        );

        // Step 1: Validate input parameters
        if node_id == 0 {
            warn_log!("Invalid node_id=0 for node deletion - treating as no-op");
            return Ok(());
        }

        // Step 2: Parse existing node data if provided, or retrieve from storage
        let node_record = if let Some(data) = old_data {
            // Deserialize NodeRecordV2 from provided old_data using binary deserialization
            NodeRecordV2::deserialize(data).map_err(|e| {
                RecoveryError::replay_failure(format!(
                    "Failed to deserialize NodeRecordV2 data: {}",
                    e
                ))
            })?
        } else {
            // For now, create a minimal node record - in real implementation would retrieve from storage
            warn_log!("No old_data provided for node delete - creating minimal rollback record");
            NodeRecordV2::new(
                node_id as i64,
                "Unknown".to_string(),
                "deleted_node".to_string(),
                serde_json::Value::Null,
            )
        };

        // Step 3: Serialize the node_record to old_data for rollback
        // Use binary serialization (not JSON) for consistency with V2 format
        let old_data = node_record.serialize();

        let old_data_len = old_data.len();

        // Step 4: CAPTURE EDGES BEFORE DELETION (critical for rollback)
        // This must happen inside the graph_file lock since we need to read cluster data
        let mut captured_outgoing_edges = Vec::new();
        let mut captured_incoming_edges = Vec::new();

        // Create NodeStore and FreeSpaceManager for this operation following proper SME methodology
        {
            let mut graph_file = self.graph_file.write().map_err(|e| {
                RecoveryError::io_error(format!("Failed to lock graph file: {}", e))
            })?;

            // Step 4.5: CAPTURE OUTGOING EDGES BEFORE DELETION
            if node_record.outgoing_edge_count > 0 {
                debug_log!(
                    "Capturing {} outgoing edges before deletion",
                    node_record.outgoing_edge_count
                );
                // Read cluster data and deserialize to get edge records
                if node_record.outgoing_cluster_offset != 0 && node_record.outgoing_cluster_size > 0
                {
                    let mut cluster_buffer = vec![0u8; node_record.outgoing_cluster_size as usize];
                    graph_file
                        .read_bytes(node_record.outgoing_cluster_offset, &mut cluster_buffer)
                        .map_err(|e| {
                            RecoveryError::io_error(format!(
                                "Failed to read outgoing cluster: {}",
                                e
                            ))
                        })?;

                    let cluster = EdgeCluster::deserialize(&cluster_buffer).map_err(|e| {
                        RecoveryError::io_error(format!(
                            "Failed to deserialize outgoing cluster: {}",
                            e
                        ))
                    })?;

                    captured_outgoing_edges = cluster.edges().to_vec();
                    debug_log!(
                        "Captured {} outgoing edge records",
                        captured_outgoing_edges.len()
                    );
                }
            }

            // Step 4.6: CAPTURE INCOMING EDGES BEFORE DELETION
            if node_record.incoming_edge_count > 0 {
                debug_log!(
                    "Capturing {} incoming edges before deletion",
                    node_record.incoming_edge_count
                );
                // Read cluster data and deserialize to get edge records
                if node_record.incoming_cluster_offset != 0 && node_record.incoming_cluster_size > 0
                {
                    let mut cluster_buffer = vec![0u8; node_record.incoming_cluster_size as usize];
                    graph_file
                        .read_bytes(node_record.incoming_cluster_offset, &mut cluster_buffer)
                        .map_err(|e| {
                            RecoveryError::io_error(format!(
                                "Failed to read incoming cluster: {}",
                                e
                            ))
                        })?;

                    let cluster = EdgeCluster::deserialize(&cluster_buffer).map_err(|e| {
                        RecoveryError::io_error(format!(
                            "Failed to deserialize incoming cluster: {}",
                            e
                        ))
                    })?;

                    captured_incoming_edges = cluster.edges().to_vec();
                    debug_log!(
                        "Captured {} incoming edge records",
                        captured_incoming_edges.len()
                    );
                }
            }

            // Step 5: Handle edge cascade cleanup (if node has cluster references)
            // Do this BEFORE creating NodeStore to avoid borrow conflicts
            if node_record.outgoing_edge_count > 0 || node_record.incoming_edge_count > 0 {
                debug_log!(
                    "Node {} has edges - performing cascade cleanup: outgoing={}, incoming={}",
                    node_id,
                    node_record.outgoing_edge_count,
                    node_record.incoming_edge_count
                );

                // Create EdgeStore for edge deletion operations
                let mut edge_store = EdgeStore::new(&mut *graph_file);

                // Collect and delete outgoing edges (edges where from_id = node_id)
                if node_record.outgoing_edge_count > 0 {
                    let outgoing_edges: Vec<(NativeNodeId, NativeNodeId)> = edge_store
                        .iter_edges_with_ids(node_id as NativeNodeId, Direction::Outgoing)
                        .collect();

                    let outgoing_count = outgoing_edges.len();
                    for (edge_id, neighbor_id) in outgoing_edges {
                        // Mark edge as deleted (soft deletion)
                        if let Err(e) = edge_store.delete_edge(edge_id) {
                            warn_log!(
                                "Failed to delete outgoing edge {} for node {} -> neighbor {}: {:?}",
                                edge_id,
                                node_id,
                                neighbor_id,
                                e
                            );
                        } else {
                            debug_log!(
                                "Deleted outgoing edge {} for node {} -> neighbor {}",
                                edge_id,
                                node_id,
                                neighbor_id
                            );
                        }
                    }

                    debug_log!(
                        "Deleted {} outgoing edges for node {}",
                        outgoing_count,
                        node_id
                    );
                }

                // Collect and delete incoming edges (edges where to_id = node_id)
                if node_record.incoming_edge_count > 0 {
                    let incoming_edges: Vec<(NativeNodeId, NativeNodeId)> = edge_store
                        .iter_edges_with_ids(node_id as NativeNodeId, Direction::Incoming)
                        .collect();

                    let incoming_count = incoming_edges.len();
                    for (edge_id, neighbor_id) in incoming_edges {
                        // Mark edge as deleted (soft deletion)
                        if let Err(e) = edge_store.delete_edge(edge_id) {
                            warn_log!(
                                "Failed to delete incoming edge {} for node {} <- neighbor {}: {:?}",
                                edge_id,
                                node_id,
                                neighbor_id,
                                e
                            );
                        } else {
                            debug_log!(
                                "Deleted incoming edge {} for node {} <- neighbor {}",
                                edge_id,
                                node_id,
                                neighbor_id
                            );
                        }
                    }

                    debug_log!(
                        "Deleted {} incoming edges for node {}",
                        incoming_count,
                        node_id
                    );
                }

                debug_log!(
                    "Successfully completed edge cascade cleanup for node {}",
                    node_id
                );
            }

            // Now create NodeStore and FreeSpaceManager for remaining operations
            let mut node_store = NodeStore::new(&mut *graph_file);
            let mut free_space_manager = FreeSpaceManager::new(AllocationStrategy::FirstFit);

            // Step 6: Clean up cluster references if they exist
            if node_record.outgoing_cluster_offset != 0 || node_record.incoming_cluster_offset != 0
            {
                debug_log!(
                    "Cleaning up cluster references for node {}: outgoing_offset={}, incoming_offset={}",
                    node_id,
                    node_record.outgoing_cluster_offset,
                    node_record.incoming_cluster_offset
                );

                // Deallocate outgoing cluster if it exists
                if node_record.outgoing_cluster_offset != 0 && node_record.outgoing_cluster_size > 0
                {
                    free_space_manager.add_free_block(
                        node_record.outgoing_cluster_offset,
                        node_record.outgoing_cluster_size,
                    );
                    debug_log!(
                        "Deallocated outgoing cluster: node_id={}, offset={}, size={}",
                        node_id,
                        node_record.outgoing_cluster_offset,
                        node_record.outgoing_cluster_size
                    );
                }

                // Deallocate incoming cluster if it exists
                if node_record.incoming_cluster_offset != 0 && node_record.incoming_cluster_size > 0
                {
                    free_space_manager.add_free_block(
                        node_record.incoming_cluster_offset,
                        node_record.incoming_cluster_size,
                    );
                    debug_log!(
                        "Deallocated incoming cluster: node_id={}, offset={}, size={}",
                        node_id,
                        node_record.incoming_cluster_offset,
                        node_record.incoming_cluster_size
                    );
                }

                debug_log!(
                    "Successfully cleaned up cluster references for node {}",
                    node_id
                );
            }

            // Step 7: Deallocate node slot using FreeSpaceManager
            if slot_offset != 0 {
                // Estimate node size for deallocation (use reasonable default for now)
                let estimated_node_size = std::mem::size_of::<NodeRecordV2>() as u32;
                free_space_manager.add_free_block(slot_offset, estimated_node_size);
                debug_log!(
                    "Deallocated node slot: offset={}, size={}",
                    slot_offset,
                    estimated_node_size
                );
            }

            // Step 8: Remove node from node index using real NodeStore deletion
            // Use delete_node_with_offset with explicit slot_offset from WAL record
            // to skip the node existence check (needed for recovery scenarios)
            node_store
                .delete_node_with_offset(node_id as NativeNodeId, Some(slot_offset))
                .map_err(|e| {
                    RecoveryError::io_error(format!(
                        "Failed to delete node {} from NodeStore: {}",
                        node_id, e
                    ))
                })?;
        } // graph_file lock, node_store, and free_space_manager are released here

        // Step 8.5: Add rollback operation AFTER edge capture but AFTER lock release
        // This ensures we have the captured edges available for rollback
        rollback_data.push(RollbackOperation::NodeDelete {
            node_id: node_id as NativeNodeId,
            slot_offset,
            old_data,
            outgoing_edges: captured_outgoing_edges,
            incoming_edges: captured_incoming_edges,
        });

        // Step 9: Update statistics (lock-free)
        self.statistics.record_node_operation();
        self.statistics.record_bytes_written(old_data_len as u64);

        debug_log!(
            "Successfully completed node delete: node_id={}, rollback_data_count={}",
            node_id,
            rollback_data.len()
        );

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::native::v2::wal::recovery::replayer::types::RollbackOperation;
    use tempfile::tempdir;

    /// Helper to create test operations instance
    fn create_test_operations() -> super::super::DefaultReplayOperations {
        super::super::DefaultReplayOperations::create_test_operations()
    }

    #[test]
    fn test_handle_node_delete_basic() {
        let operations = create_test_operations();
        let mut rollback_data = Vec::new();

        // Test basic node deletion with minimal parameters
        let result = operations.handle_node_delete(42, 4096, None, &mut rollback_data);

        // Basic node delete should succeed (node_id != 0)
        assert!(result.is_ok(), "Basic node delete should succeed");

        // Should record rollback operation
        assert_eq!(rollback_data.len(), 1, "Should record rollback operation");

        // Verify rollback operation structure
        if let Some(RollbackOperation::NodeDelete {
            node_id,
            slot_offset,
            ..
        }) = rollback_data.first()
        {
            assert_eq!(*node_id, 42, "Rollback should preserve node ID");
            assert_eq!(*slot_offset, 4096, "Rollback should preserve slot offset");
        } else {
            panic!("Expected NodeDelete rollback operation");
        }
    }

    #[test]
    fn test_handle_node_delete_with_old_data() {
        let operations = create_test_operations();
        let mut rollback_data = Vec::new();

        // Create test node data for deletion
        let test_node = NodeRecordV2::new(
            123,
            "Document".to_string(),
            "test_doc".to_string(),
            serde_json::json!({"content": "test data", "version": 1}),
        );
        let serialized_data = test_node.serialize(); // Use binary serialization

        let result =
            operations.handle_node_delete(123, 8192, Some(&serialized_data), &mut rollback_data);

        assert!(result.is_ok(), "Node delete with old data should succeed");

        // Should record rollback with preserved data
        assert_eq!(rollback_data.len(), 1, "Should record rollback operation");

        if let Some(RollbackOperation::NodeDelete {
            node_id,
            slot_offset,
            old_data,
            ..
        }) = rollback_data.first()
        {
            assert_eq!(*node_id, 123);
            assert_eq!(*slot_offset, 8192);
            // Verify old_data was preserved using binary serialization
            let restored_node = NodeRecordV2::deserialize(old_data);
            assert!(
                restored_node.is_ok(),
                "Old data should be valid binary serialization"
            );
            if let Ok(node) = restored_node {
                assert_eq!(node.id, 123);
                assert_eq!(node.kind, "Document");
                assert_eq!(node.name, "test_doc");
            }
        } else {
            panic!("Expected NodeDelete rollback operation");
        }
    }

    #[test]
    fn test_handle_node_delete_nonexistent_node() {
        let operations = create_test_operations();
        let mut rollback_data = Vec::new();

        // Test deletion of node that doesn't exist
        let result = operations.handle_node_delete(
            999999, // Non-existent node ID
            4096,
            None,
            &mut rollback_data,
        );

        // Should handle gracefully - node deletion should succeed even if node doesn't exist
        assert!(result.is_ok(), "Should handle non-existent node gracefully");

        // Should still record rollback operation
        assert_eq!(
            rollback_data.len(),
            1,
            "Should record rollback operation even for non-existent node"
        );
    }

    #[test]
    fn test_handle_node_delete_with_cluster_references() {
        let operations = create_test_operations();
        let mut rollback_data = Vec::new();

        // Create node with cluster references (complex deletion scenario)
        let mut test_node = NodeRecordV2::new(
            456,
            "Function".to_string(),
            "complex_func".to_string(),
            serde_json::json!({"complex": "node"}),
        );
        // Set cluster references manually
        test_node.outgoing_cluster_offset = 1024;
        test_node.outgoing_cluster_size = 256;
        test_node.outgoing_edge_count = 5;
        test_node.incoming_cluster_offset = 2048;
        test_node.incoming_cluster_size = 128;
        test_node.incoming_edge_count = 3;
        let serialized_data = test_node.serialize();

        let result =
            operations.handle_node_delete(456, 4096, Some(&serialized_data), &mut rollback_data);

        // Handle cluster reference deletion - may fail due to missing actual cluster data
        // The implementation reads from graph_file which may not have valid cluster data
        // For test purposes, we expect either success or specific failure modes
        match &result {
            Ok(()) => {
                // This scenario requires edge cascade cleanup, cluster reference cleanup,
                // slot deallocation, and rollback operation recording
                assert_eq!(rollback_data.len(), 1, "Should record rollback operation");
            }
            Err(e) => {
                // If it fails, it should be due to I/O reading cluster data from empty graph file
                // This is acceptable for testing since we don't have actual cluster data written
                println!(
                    "Node delete with cluster references failed (expected for test): {}",
                    e.message
                );
            }
        }
    }

    #[test]
    fn test_handle_node_delete_malformed_old_data() {
        let operations = create_test_operations();
        let mut rollback_data = Vec::new();

        // Test with malformed node data (invalid binary serialization)
        let malformed_data = vec![1, 2, 3]; // Invalid serialization

        let result =
            operations.handle_node_delete(42, 4096, Some(&malformed_data), &mut rollback_data);

        // Should handle malformed data - deserialization will fail
        // The implementation should handle this gracefully
        assert!(result.is_err(), "Malformed data should cause error");

        // Should not record rollback operation for failed operation
        assert_eq!(
            rollback_data.len(),
            0,
            "Should not record rollback operation for malformed data"
        );
    }

    #[test]
    fn test_handle_node_delete_zero_node_id() {
        let operations = create_test_operations();
        let mut rollback_data = Vec::new();

        // Test with invalid node ID
        let result = operations.handle_node_delete(
            0, // Invalid node ID
            4096,
            None,
            &mut rollback_data,
        );

        // Should handle invalid node ID gracefully - implementation treats node_id=0 as no-op
        assert!(result.is_ok(), "Should handle invalid node ID gracefully");

        // Should NOT record rollback operation for node_id=0 (it's a no-op)
        assert_eq!(
            rollback_data.len(),
            0,
            "Should not record rollback operation for node_id=0 (no-op)"
        );
    }

    #[test]
    fn test_handle_node_delete_rollback_operation_preserves_slot_offset() {
        let operations = create_test_operations();
        let mut rollback_data = Vec::new();

        // Test that rollback operation correctly preserves slot offset for restoration
        let test_slot_offset = 16384;

        let result = operations.handle_node_delete(789, test_slot_offset, None, &mut rollback_data);

        assert!(result.is_ok(), "Node delete should succeed");

        assert_eq!(rollback_data.len(), 1, "Should record rollback operation");

        if let Some(RollbackOperation::NodeDelete {
            node_id,
            slot_offset,
            ..
        }) = rollback_data.first()
        {
            assert_eq!(*node_id, 789, "Should preserve correct node ID");
            assert_eq!(
                *slot_offset, test_slot_offset,
                "Should preserve exact slot offset for restoration"
            );
        } else {
            panic!("Expected NodeDelete rollback operation");
        }
    }

    #[test]
    fn test_handle_node_delete_edge_cleanup_required() {
        let operations = create_test_operations();
        let mut rollback_data = Vec::new();

        // Create node with edges (requires cascade cleanup)
        let mut node_with_edges = NodeRecordV2::new(
            555,
            "Module".to_string(),
            "test_module".to_string(),
            serde_json::json!({"has_edges": true}),
        );
        // Set cluster references manually
        node_with_edges.outgoing_cluster_offset = 512;
        node_with_edges.outgoing_cluster_size = 64;
        node_with_edges.outgoing_edge_count = 2;
        node_with_edges.incoming_cluster_offset = 1024;
        node_with_edges.incoming_cluster_size = 32;
        node_with_edges.incoming_edge_count = 1;
        let serialized_data = node_with_edges.serialize();

        let result =
            operations.handle_node_delete(555, 4096, Some(&serialized_data), &mut rollback_data);

        // Handle edge cleanup - may fail due to missing actual cluster data
        match &result {
            Ok(()) => {
                assert_eq!(rollback_data.len(), 1, "Should record rollback operation");
            }
            Err(e) => {
                // If it fails, it should be due to I/O reading cluster data from empty graph file
                // This is acceptable for testing since we don't have actual cluster data written
                println!(
                    "Node delete with edges failed (expected for test): {}",
                    e.message
                );
            }
        }
    }

    #[test]
    fn test_full_node_delete_and_restore_cycle() {
        use crate::backend::native::GraphFile;
        use crate::backend::native::v2::StringTable;
        use crate::backend::native::v2::free_space::AllocationStrategy;
        use crate::backend::native::v2::wal::recovery::replayer::rollback::RollbackSystem;
        use std::sync::{Arc, Mutex, RwLock};

        // Setup: Create test graph file and components
        let temp_file = tempfile::NamedTempFile::new().expect("Failed to create temp file");
        let graph_file = GraphFile::create(temp_file.path()).expect("Failed to create GraphFile");
        let graph_file = Arc::new(RwLock::new(graph_file));

        let node_store: Arc<Mutex<Option<NodeStore<'static>>>> = Arc::new(Mutex::new(None));
        let edge_store: Arc<Mutex<Option<EdgeStore<'static>>>> = Arc::new(Mutex::new(None));
        let string_table = Arc::new(Mutex::new(StringTable::new()));

        // Initialize FreeSpaceManager with initial free space
        let mut free_space_mgr = FreeSpaceManager::new(AllocationStrategy::FirstFit);
        free_space_mgr.add_free_block(2048, 1024 * 1024); // 1MB of free space
        let free_space_manager = Arc::new(Mutex::new(Some(free_space_mgr)));

        let statistics = Arc::new(
            crate::backend::native::v2::wal::recovery::replayer::types::ReplayStatistics::new(),
        );
        let kv_store = Arc::new(Mutex::new(crate::backend::native::v2::KvStore::new()));

        // Create operations handler
        let operations = super::super::DefaultReplayOperations::new(
            graph_file.clone(),
            node_store.clone(),
            edge_store,
            string_table,
            free_space_manager,
            kv_store,
            statistics,
        );

        // Step 1: Create a node with initial data
        let original_node = NodeRecordV2::new(
            1001,
            "TestClass".to_string(),
            "test_method".to_string(),
            serde_json::json!({"version": 1, "state": "initial"}),
        );

        // Write the node to storage
        {
            let mut graph_file_lock = graph_file.write().unwrap();
            let mut node_store = NodeStore::new(&mut *graph_file_lock);
            node_store
                .write_node_v2(&original_node)
                .expect("Failed to write initial node");
        }

        // Verify node exists
        {
            let mut graph_file_lock = graph_file.write().unwrap();
            let mut node_store = NodeStore::new(&mut *graph_file_lock);
            let read_result = node_store.read_node_v2(1001);
            assert!(read_result.is_ok(), "Node should exist after creation");
            let node = read_result.unwrap();
            assert_eq!(node.id, 1001);
            assert_eq!(node.name, "test_method");
        }

        // Step 2: Delete the node (capturing rollback data)
        let mut rollback_data = Vec::new();
        let serialized_old_data = original_node.serialize();

        let delete_result = operations.handle_node_delete(
            1001,
            4096,
            Some(&serialized_old_data),
            &mut rollback_data,
        );
        assert!(delete_result.is_ok(), "Node delete should succeed");
        assert_eq!(rollback_data.len(), 1, "Should record rollback operation");

        // Note: Node deletion may have side effects depending on the graph_file state
        // The key test is that rollback data is captured correctly

        // Step 3: Extract rollback operation data
        let (node_id, slot_offset, old_data, outgoing_edges, incoming_edges) =
            match rollback_data.first() {
                Some(RollbackOperation::NodeDelete {
                    node_id,
                    slot_offset,
                    old_data,
                    outgoing_edges,
                    incoming_edges,
                }) => (
                    *node_id,
                    *slot_offset,
                    old_data.clone(),
                    outgoing_edges.clone(),
                    incoming_edges.clone(),
                ),
                _ => panic!("Expected NodeDelete rollback operation"),
            };

        // Step 4: Rollback using the captured data
        let rollback_system = RollbackSystem::new(
            graph_file.clone(),
            node_store.clone(),
            Arc::new(Mutex::new(StringTable::new())),
            Arc::new(Mutex::new(Some(FreeSpaceManager::new(
                AllocationStrategy::FirstFit,
            )))),
        );

        let rollback_result = crate::backend::native::v2::wal::recovery::replayer::rollback::node_ops::rollback_node_delete(
            &rollback_system,
            node_id,
            slot_offset,
            old_data,
            outgoing_edges,
            incoming_edges,
        );
        assert!(rollback_result.is_ok(), "Rollback should succeed");

        // Step 5: Verify node is restored with correct data
        {
            let mut graph_file_lock = graph_file.write().unwrap();
            let mut node_store = NodeStore::new(&mut *graph_file_lock);
            let read_result = node_store.read_node_v2(1001);
            assert!(read_result.is_ok(), "Node should exist after rollback");
            let restored_node = read_result.unwrap();
            assert_eq!(restored_node.id, 1001, "Node ID should be restored");
            assert_eq!(
                restored_node.kind, "TestClass",
                "Node kind should be restored"
            );
            assert_eq!(
                restored_node.name, "test_method",
                "Node name should be restored"
            );
            assert_eq!(
                restored_node.data.get("version"),
                Some(&serde_json::json!(1)),
                "Node data should be preserved"
            );
        }

        // The test passes: full delete/rollback cycle completes correctly
    }
}