tiny-counter 0.1.0

// Concurrent access and thread safety.
//
// This example shows how to:
// - Share EventStore across multiple threads
// - Record events from concurrent threads
// - Query from a separate thread
// - Use reservations to prevent race conditions in concurrent scenarios

use std::sync::Arc;
use std::thread;
use tiny_counter::{EventStore, TimeUnit};

fn main() {
    println!("=== Concurrent Access Examples ===\n");

    // 1. Basic concurrent recording
    println!("1. Recording from multiple threads:");
    let store = Arc::new(EventStore::new());

    let mut handles = vec![];

    // Spawn 5 threads, each recording 100 events
    for thread_id in 0..5 {
        let store = Arc::clone(&store);
        let handle = thread::spawn(move || {
            for _ in 0..100 {
                store.record("concurrent_event");
            }
            println!("  Thread {} recorded 100 events", thread_id);
        });
        handles.push(handle);
    }

    // Wait for all threads to complete
    for handle in handles {
        handle.join().unwrap();
    }

    // Query the total from main thread
    let total = store
        .query("concurrent_event")
        .last_days(1)
        .sum()
        .unwrap_or(0);

    println!("  Total events recorded: {} (expected 500)\n", total);

    // 2. Concurrent queries
    println!("2. Querying from multiple threads:");
    let store = Arc::new(EventStore::new());

    // Pre-populate with data
    store.record_count("shared_data", 1000);

    let mut handles = vec![];

    // Spawn multiple reader threads
    for thread_id in 0..3 {
        let store = Arc::clone(&store);
        let handle = thread::spawn(move || {
            let count = store.query("shared_data").last_days(7).sum().unwrap_or(0);
            println!("  Thread {} read: {} events", thread_id, count);
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    // 3. Prevent race conditions with reservations
    println!("\n3. Race condition prevention with reservations:");
    let store = Arc::new(EventStore::new());

    let mut handles = vec![];

    // Spawn 20 threads all trying to reserve from a limit of 10
    for thread_id in 0..20 {
        let store = Arc::clone(&store);
        let handle = thread::spawn(move || {
            let result = store
                .limit()
                .at_most("limited_api", 10, TimeUnit::Hours)
                .reserve("limited_api");

            match result {
                Ok(reservation) => {
                    // Success - commit it
                    reservation.commit();
                    (thread_id, true)
                }
                Err(_) => (thread_id, false),
            }
        });
        handles.push(handle);
    }

    let mut successes = 0;
    let mut failures = 0;

    for handle in handles {
        let (thread_id, success) = handle.join().unwrap();
        if success {
            successes += 1;
            println!("  Thread {}: ✓ Reserved slot", thread_id);
        } else {
            failures += 1;
        }
    }

    println!("  Total: {} succeeded, {} failed", successes, failures);
    println!("  (Expected: exactly 10 succeeded, 10 failed)\n");

    // 4. Producer-consumer pattern
    println!("4. Producer-consumer pattern:");
    let store = Arc::new(EventStore::new());

    // Producer thread
    let producer_store = Arc::clone(&store);
    let producer = thread::spawn(move || {
        for i in 0..50 {
            producer_store.record("message");
            if (i + 1) % 10 == 0 {
                println!("  Producer: Sent {} messages", i + 1);
            }
            thread::sleep(std::time::Duration::from_millis(10));
        }
    });

    // Consumer thread
    let consumer_store = Arc::clone(&store);
    let consumer = thread::spawn(move || {
        thread::sleep(std::time::Duration::from_millis(100));

        let mut last_count = 0;
        for _ in 0..5 {
            let count = consumer_store
                .query("message")
                .last_days(1)
                .sum()
                .unwrap_or(0);

            let new_messages = count - last_count;
            if new_messages > 0 {
                println!(
                    "  Consumer: Processed {} messages (total: {})",
                    new_messages, count
                );
            }
            last_count = count;

            thread::sleep(std::time::Duration::from_millis(100));
        }
    });

    producer.join().unwrap();
    consumer.join().unwrap();

    // 5. Concurrent updates to different events
    println!("\n5. Concurrent updates to different events:");
    let store = Arc::new(EventStore::new());

    let mut handles = vec![];

    let events = ["event_a", "event_b", "event_c", "event_d"];

    for (thread_id, event_name) in events.iter().enumerate() {
        let store = Arc::clone(&store);
        let event_name = event_name.to_string();
        let handle = thread::spawn(move || {
            for _ in 0..25 {
                store.record(&event_name);
            }
            println!("  Thread {} recorded 25 {} events", thread_id, event_name);
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    // Query all events
    let total = store
        .query_many(&["event_a", "event_b", "event_c", "event_d"])
        .last_days(1)
        .sum()
        .unwrap_or(0);

    println!(
        "  Total events across all types: {} (expected 100)\n",
        total
    );

    // 6. Stress test
    println!("6. Stress test (100 threads, 10 events each):");
    let store = Arc::new(EventStore::new());

    let mut handles = vec![];

    for _ in 0..100 {
        let store = Arc::clone(&store);
        let handle = thread::spawn(move || {
            for _ in 0..10 {
                store.record("stress_test");
            }
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    let final_count = store.query("stress_test").last_days(1).sum().unwrap_or(0);

    println!("  Final count: {} (expected 1000)", final_count);

    if final_count == 1000 {
        println!("  ✓ No events lost to race conditions");
    } else {
        println!("  ✗ Lost {} events", 1000 - final_count);
    }

    println!("\n✓ Concurrent access complete!");
}