Blockchain-CRDT Examples

Practical examples for using the Blockchain-CRDT Hybrid Sync system.

Example 1: Distributed Counter

use heliosdb_blockchain_crdt::*;
use std::sync::Arc;

async fn distributed_counter_example() -> Result<()> {
    // Setup
    let mut validator_set = ValidatorSet::new(0.33);
    let keypair = ValidatorKeypair::generate(NodeId::new("node1"));
    validator_set.add_validator(keypair.to_validator());

    let consensus = Arc::new(ProofOfAuthority::with_keypair(validator_set, keypair));
    let blockchain = Arc::new(Blockchain::new(consensus));

    let config = BlockchainCrdtConfig::default();
    let manager = BlockchainCrdtManager::new(config, blockchain);

    // Initialize counter
    let counter = GCounter::new();
    manager.protocol().init_crdt_state(
        "page_views".to_string(),
        CrdtState::GCounter(counter),
    );

    // Increment from multiple nodes
    for i in 0..10 {
        manager.protocol().apply_operation(
            "page_views".to_string(),
            CrdtOperation::Increment {
                node_id: NodeId::new(format!("node{}", i % 3)),
                amount: 1,
            },
        )?;
    }

    // Commit to blockchain
    manager.protocol().commit_operations().await?;

    // Get value
    let state = manager.protocol().get_crdt_state("page_views").unwrap();
    if let CrdtState::GCounter(counter) = state {
        println!("Total page views: {}", counter.value());
    }

    Ok(())
}

Example 2: Collaborative Shopping Cart

async fn shopping_cart_example() -> Result<()> {
    let manager = setup_manager().await?;

    // Initialize OR-Set for shopping cart
    let cart = ORSet::new();
    manager.protocol().init_crdt_state(
        "shopping_cart".to_string(),
        CrdtState::ORSet(cart),
    );

    // Add items (different users)
    let items = vec![
        ("laptop", "Product A"),
        ("mouse", "Product B"),
        ("keyboard", "Product C"),
    ];

    for (item_id, item_name) in items {
        manager.protocol().apply_operation(
            "shopping_cart".to_string(),
            CrdtOperation::AddElement {
                element_id: item_id.to_string(),
                timestamp: LogicalTimestamp::new(1, &NodeId::new("user1")),
                value: item_name.as_bytes().to_vec(),
            },
        )?;
    }

    // Remove item
    manager.protocol().apply_operation(
        "shopping_cart".to_string(),
        CrdtOperation::RemoveElement {
            element_id: "mouse".to_string(),
            timestamp: LogicalTimestamp::new(2, &NodeId::new("user1")),
        },
    )?;

    // Commit
    manager.protocol().commit_operations().await?;

    // Audit trail is automatically maintained in blockchain
    assert!(manager.protocol().verify_integrity().await?);

    Ok(())
}

Example 3: Multi-Node Convergence

async fn multi_node_convergence_example() -> Result<()> {
    // Create 3 nodes
    let node1 = create_node("node1").await?;
    let node2 = create_node("node2").await?;
    let node3 = create_node("node3").await?;

    // Each node tracks inventory
    for node in &[&node1, &node2, &node3] {
        let counter = PNCounter::new();
        node.protocol().init_crdt_state(
            "inventory".to_string(),
            CrdtState::PNCounter(counter),
        );
    }

    // Node 1: Add 100 items
    node1.protocol().apply_operation(
        "inventory".to_string(),
        CrdtOperation::Increment {
            node_id: NodeId::new("node1"),
            amount: 100,
        },
    )?;

    // Node 2: Sell 20 items
    node2.protocol().apply_operation(
        "inventory".to_string(),
        CrdtOperation::Decrement {
            node_id: NodeId::new("node2"),
            amount: 20,
        },
    )?;

    // Node 3: Add 50 items
    node3.protocol().apply_operation(
        "inventory".to_string(),
        CrdtOperation::Increment {
            node_id: NodeId::new("node3"),
            amount: 50,
        },
    )?;

    // Sync nodes
    let state1 = node1.protocol().get_crdt_state("inventory").unwrap();
    let state2 = node2.protocol().get_crdt_state("inventory").unwrap();
    let state3 = node3.protocol().get_crdt_state("inventory").unwrap();

    node1.protocol().merge_crdt_state("inventory".to_string(), state2.clone())?;
    node1.protocol().merge_crdt_state("inventory".to_string(), state3.clone())?;

    // All nodes converge to 130 (100 - 20 + 50)
    let final_state = node1.protocol().get_crdt_state("inventory").unwrap();
    if let CrdtState::PNCounter(counter) = final_state {
        assert_eq!(counter.value(), 130);
    }

    Ok(())
}

Example 4: GDPR Compliance

async fn gdpr_compliance_example() -> Result<()> {
    let manager = setup_manager().await?;

    // Register data subject
    manager.gdpr().register_subject("user@example.com".to_string(), true)?;

    // Link data to subject
    manager.gdpr().link_data("user@example.com", "profile_data".to_string())?;
    manager.gdpr().link_data("user@example.com", "purchase_history".to_string())?;

    // User requests data export (data portability)
    let data_ids = manager.gdpr().export_data("user@example.com")?;
    println!("Exported data IDs: {:?}", data_ids);

    // User requests deletion (right to be forgotten)
    use heliosdb_blockchain_crdt::security::gdpr::RedactionReason;
    let request_id = manager.gdpr().request_redaction(
        "user@example.com".to_string(),
        RedactionReason::RightToBeForgotten,
    )?;

    // Process redaction request
    manager.gdpr().process_redaction(request_id).await?;

    // Verify in audit log
    let audit_entries = manager.audit_logger().get_all();
    assert!(audit_entries.iter().any(|e| e.operation_type == "DataRedacted"));

    Ok(())
}

Example 5: Byzantine Fault Detection

async fn byzantine_detection_example() -> Result<()> {
    let manager = setup_manager().await?;

    // Create honest sync state
    let mut honest_clock = VectorClock::new();
    honest_clock.increment(NodeId::new("honest_node"));

    let honest_state = SyncState {
        node_id: NodeId::new("honest_node"),
        last_synced_block: 10,
        vector_clock: honest_clock,
        pending_operations: vec![],
    };

    // Sync succeeds
    manager.protocol()
        .sync_with_peer(NodeId::new("honest_node"), honest_state)
        .await?;

    // Create malicious sync state (invalid block reference)
    let mut malicious_clock = VectorClock::new();
    malicious_clock.increment(NodeId::new("malicious_node"));

    let malicious_state = SyncState {
        node_id: NodeId::new("malicious_node"),
        last_synced_block: 10_000_000, // Unreasonably high
        vector_clock: malicious_clock,
        pending_operations: vec![],
    };

    // Sync fails - byzantine behavior detected
    let result = manager.protocol()
        .sync_with_peer(NodeId::new("malicious_node"), malicious_state)
        .await;

    assert!(result.is_err());

    Ok(())
}

Example 6: Data Encryption

async fn encryption_example() -> Result<()> {
    let manager = setup_manager().await?;
    let crypto = manager.crypto();

    // Generate encryption key
    let key = crypto.generate_key()?;

    // Encrypt sensitive data before storing in CRDT
    let sensitive_data = b"credit card: 1234-5678-9012-3456";
    let encrypted = crypto.encrypt(sensitive_data, &key)?;

    // Store encrypted data in CRDT
    let mut set = LWWElementSet::new();
    let node = NodeId::new("node1");
    set.add(
        "payment_info".to_string(),
        LogicalTimestamp::new(1, &node),
        encrypted.ciphertext,
    );

    manager.protocol().init_crdt_state(
        "user_payment".to_string(),
        CrdtState::LWWSet(set),
    );

    // Later: retrieve and decrypt
    let state = manager.protocol().get_crdt_state("user_payment").unwrap();
    if let CrdtState::LWWSet(set) = state {
        if let Some(encrypted_bytes) = set.get("payment_info") {
            let encrypted_data = EncryptedData {
                ciphertext: encrypted_bytes.to_vec(),
                nonce: encrypted.nonce.clone(),
            };
            let decrypted = crypto.decrypt(&encrypted_data, &key)?;
            println!("Decrypted: {}", String::from_utf8_lossy(&decrypted));
        }
    }

    Ok(())
}

Example 7: Audit Trail Query

async fn audit_trail_example() -> Result<()> {
    let manager = setup_manager().await?;

    // Perform operations that generate audit events
    manager.protocol().apply_operation(
        "data1".to_string(),
        CrdtOperation::Increment {
            node_id: NodeId::new("node1"),
            amount: 10,
        },
    )?;

    manager.protocol().commit_operations().await?;

    // Query audit log
    let all_events = manager.audit_logger().get_all();
    println!("Total audit events: {}", all_events.len());

    // Query by block
    let block1_events = manager.audit_logger().get_by_block(1);
    println!("Block 1 events: {}", block1_events.len());

    // Query by node
    let node1_events = manager.audit_logger().get_by_node(&NodeId::new("node1"));
    println!("Node 1 events: {}", node1_events.len());

    // Query by type
    let operation_events = manager.audit_logger().get_by_type("OperationApplied");
    println!("Operation events: {}", operation_events.len());

    Ok(())
}

Helper Functions

use heliosdb_blockchain_crdt::*;
use std::sync::Arc;

async fn setup_manager() -> Result<BlockchainCrdtManager> {
    let mut validator_set = ValidatorSet::new(0.33);
    let keypair = ValidatorKeypair::generate(NodeId::new("node1"));
    validator_set.add_validator(keypair.to_validator());

    let consensus = Arc::new(ProofOfAuthority::with_keypair(validator_set, keypair));
    let blockchain = Arc::new(Blockchain::new(consensus));

    let config = BlockchainCrdtConfig::default();
    Ok(BlockchainCrdtManager::new(config, blockchain))
}

async fn create_node(node_id: &str) -> Result<BlockchainCrdtManager> {
    let mut validator_set = ValidatorSet::new(0.33);
    let keypair = ValidatorKeypair::generate(NodeId::new(node_id));
    validator_set.add_validator(keypair.to_validator());

    let consensus = Arc::new(ProofOfAuthority::with_keypair(validator_set, keypair));
    let blockchain = Arc::new(Blockchain::new(consensus));

    let config = BlockchainCrdtConfig {
        node_id: NodeId::new(node_id),
        ..Default::default()
    };

    Ok(BlockchainCrdtManager::new(config, blockchain))
}

Running Examples

# Run all examples
cargo run --example distributed_counter
cargo run --example shopping_cart
cargo run --example multi_node_sync
cargo run --example gdpr_compliance
cargo run --example byzantine_detection
cargo run --example encryption
cargo run --example audit_trail