basic_usage.rs

//! Basic usage example for Timeplus Rust client
//! 
//! This example demonstrates:
//! - Connecting to Timeplus
//! - Creating a stream
//! - Inserting data
//! - Querying data
//! - Cleaning up
//!
//! Run with: `cargo run --example basic_usage`

use timeplus_client::{TimeplusClient, utils};
use serde_json::json;
use std::collections::HashMap;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    println!("🚀 Timeplus Basic Usage Example");
    println!("{}", "=".repeat(50));

    // Create client using utility function
    let client = TimeplusClient::with_config(utils::local_config())?;

    // Test connection
    println!("\n🔍 Testing connection...");
    match client.health_check().await {
        Ok(true) => {
            let version = client.version().await?;
            println!("✅ Connected to Timeplus version: {}", version);
        }
        Ok(false) => {
            println!("❌ Health check failed");
            return Ok(());
        }
        Err(e) => {
            println!("❌ Connection error: {}", e);
            println!("💡 Make sure Timeplus server is running:");
            println!("   docker run -d --name timeplus-server \\");
            println!("     -p 8000:8000 -p 8123:8123 -p 8463:8463 -p 3218:3218 \\");
            println!("     -v /tmp/timeplus_data:/timeplus/data \\");
            println!("     docker.timeplus.com/timeplus/timeplus-enterprise:2.7.9");
            return Ok(());
        }
    }

    // Create a stream
    println!("\n📊 Creating stream...");
    let stream_name = "basic_example_stream";
    let schema = "id uint32, name string, value float64, category string, timestamp datetime64(9) DEFAULT now64()";

    // Clean up first (in case stream exists)
    let _ = client.streams().drop(stream_name).await;

    client.streams().create(stream_name, schema).await?;
    println!("✅ Stream '{}' created", stream_name);

    // Insert some data
    println!("\n📝 Inserting data...");
    
    // Method 1: Insert single record with key-value pairs
    let data = vec![
        ("id", json!(1)),
        ("name", json!("Alice")),
        ("value", json!(10.5)),
        ("category", json!("A")),
    ];
    client.streams().insert_data(stream_name, &data).await?;
    println!("✅ Inserted single record");

    // Method 2: Insert batch records
    let records = vec![
        HashMap::from([
            ("id".to_string(), json!(2)),
            ("name".to_string(), json!("Bob")),
            ("value".to_string(), json!(20.3)),
            ("category".to_string(), json!("B")),
        ]),
        HashMap::from([
            ("id".to_string(), json!(3)),
            ("name".to_string(), json!("Charlie")),
            ("value".to_string(), json!(15.7)),
            ("category".to_string(), json!("A")),
        ]),
        HashMap::from([
            ("id".to_string(), json!(4)),
            ("name".to_string(), json!("Diana")),
            ("value".to_string(), json!(25.1)),
            ("category".to_string(), json!("C")),
        ]),
    ];
    client.streams().insert_batch(stream_name, &records).await?;

    // Query the data
    println!("\n🔍 Querying data...");
    
    // Get all records
    let result = client.streams().query(stream_name, None).await?;
    println!("✅ Found {} records in stream", result.row_count());

    // Display first few records
    for (i, row) in result.rows().iter().take(3).enumerate() {
        println!("   Record {}: {}", i + 1, serde_json::to_string_pretty(row)?);
    }

    // Get stream statistics
    println!("\n📊 Stream statistics...");
    let stats = client.streams().stats(stream_name).await?;
    for (key, value) in stats {
        println!("   {}: {}", key, value);
    }

    // Advanced queries
    println!("\n🔍 Advanced queries...");
    
    // Aggregation query
    let agg_result = client.query(&format!(
        "SELECT category, count(*) as count, avg(value) as avg_value FROM table({}) GROUP BY category ORDER BY count DESC",
        stream_name
    )).await?;
    
    println!("✅ Aggregation results:");
    for row in agg_result.rows() {
        if let (Some(cat), Some(count), Some(avg)) = (
            row.get("category").and_then(|v| v.as_str()),
            row.get("count").and_then(|v| v.as_u64()),
            row.get("avg_value").and_then(|v| v.as_f64())
        ) {
            println!("   Category {}: {} records, avg value: {:.2}", cat, count, avg);
        }
    }

    // Filter query
    let filter_result = client.query(&format!(
        "SELECT * FROM table({}) WHERE value > 20.0 ORDER BY value DESC",
        stream_name
    )).await?;
    
    println!("✅ High-value records (value > 20.0): {} found", filter_result.row_count());
    for row in filter_result.rows() {
        if let (Some(name), Some(value)) = (
            row.get("name").and_then(|v| v.as_str()),
            row.get("value").and_then(|v| v.as_f64())
        ) {
            println!("   {}: {:.1}", name, value);
        }
    }

    // List all streams
    println!("\n📋 Listing all streams...");
    let streams = client.streams().list().await?;
    println!("✅ Found {} streams:", streams.len());
    for stream in streams.iter().take(5) {
        println!("   - {}: {}", stream.name, 
                stream.schema.chars().take(50).collect::<String>());
    }

    // Clean up
    println!("\n🧹 Cleaning up...");
    client.streams().drop(stream_name).await?;
    println!("✅ Stream '{}' dropped", stream_name);

    println!("\n🎉 Basic usage example completed successfully!");
    println!("💡 Next steps:");
    println!("   - Try the streaming_callbacks example for real-time processing");
    println!("   - Try the python_udfs example for custom functions");
    println!("   - Try the advanced_queries example for complex SQL");

    Ok(())
}