use announcement::Announcement;
use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
use std::hint::black_box;
use std::sync::Arc;
use std::time::Duration;

// ============================================================================
// SHUTDOWN SIGNAL PATTERN - Notify all workers to shut down
// ============================================================================

fn bench_shutdown_signal(c: &mut Criterion) {
    let mut group = c.benchmark_group("pattern/shutdown_signal");

    // Simulate shutdown signal to various numbers of workers
    for num_workers in [10, 100, 1_000, 10_000].iter() {
        group.throughput(Throughput::Elements(*num_workers as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_workers),
            num_workers,
            |b, &n| {
                b.iter(|| {
                    let (shutdown_tx, shutdown_announcement) = Announcement::new();

                    // Create N worker listeners
                    let mut workers = Vec::with_capacity(n);
                    for _ in 0..n {
                        workers.push(shutdown_announcement.listener());
                    }

                    // Send shutdown signal
                    shutdown_tx.announce(Arc::new(())).unwrap();

                    // All workers check for shutdown
                    for worker in &workers {
                        black_box(worker.try_listen());
                    }
                });
            },
        );
    }

    group.finish();
}

fn bench_shutdown_with_tokio_tasks(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("pattern/shutdown_async");
    group.sample_size(10);
    group.measurement_time(Duration::from_secs(10));

    for num_workers in [10, 100].iter() {
        group.throughput(Throughput::Elements(*num_workers as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_workers),
            num_workers,
            |b, &n| {
                b.to_async(&rt).iter(|| async move {
                    let (shutdown_tx, shutdown_announcement) = Announcement::new();

                    // Spawn N worker tasks
                    let mut handles = Vec::with_capacity(n);
                    for i in 0..n {
                        let listener = shutdown_announcement.listener();
                        handles.push(tokio::spawn(async move {
                            // Wait for shutdown
                            listener.listen().await.unwrap();
                            i
                        }));
                    }

                    // Small delay to let tasks start
                    tokio::time::sleep(Duration::from_millis(1)).await;

                    // Send shutdown signal
                    shutdown_tx.announce(Arc::new(())).unwrap();

                    // Wait for all workers to complete
                    for handle in handles {
                        black_box(handle.await.unwrap());
                    }
                });
            },
        );
    }

    group.finish();
}

// ============================================================================
// CONFIG BROADCAST PATTERN - Share configuration to all services
// ============================================================================

#[derive(Clone, Debug)]
#[allow(dead_code)]
struct Config {
    database_url: String,
    api_key: String,
    max_connections: usize,
    timeout_ms: u64,
}

fn bench_config_broadcast(c: &mut Criterion) {
    let mut group = c.benchmark_group("pattern/config_broadcast");

    for num_services in [10, 100].iter() {
        group.throughput(Throughput::Elements(*num_services as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_services),
            num_services,
            |b, &n| {
                b.iter(|| {
                    let (config_tx, config_announcement) = Announcement::new();

                    // Create N service listeners
                    let mut services = Vec::with_capacity(n);
                    for _ in 0..n {
                        services.push(config_announcement.listener());
                    }

                    // Broadcast configuration
                    let config = Arc::new(Config {
                        database_url: "postgres://localhost/db".to_string(),
                        api_key: "secret-key-12345".to_string(),
                        max_connections: 100,
                        timeout_ms: 5000,
                    });

                    config_tx.announce(config).unwrap();

                    // All services receive config
                    for service in &services {
                        black_box(service.try_listen());
                    }
                });
            },
        );
    }

    group.finish();
}

fn bench_config_broadcast_async(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("pattern/config_broadcast_async");
    group.sample_size(10);
    group.measurement_time(Duration::from_secs(10));

    for num_services in [10, 100].iter() {
        group.throughput(Throughput::Elements(*num_services as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_services),
            num_services,
            |b, &n| {
                b.to_async(&rt).iter(|| async move {
                    let (config_tx, config_announcement) = Announcement::new();

                    // Spawn N service tasks waiting for config
                    let mut handles = Vec::with_capacity(n);
                    for i in 0..n {
                        let listener = config_announcement.listener();
                        handles.push(tokio::spawn(async move {
                            // Wait for config
                            let config: Arc<Config> = listener.listen().await.unwrap();
                            // Use config
                            (i, config.max_connections)
                        }));
                    }

                    // Small delay to let tasks start
                    tokio::time::sleep(Duration::from_millis(1)).await;

                    // Broadcast configuration
                    let config = Arc::new(Config {
                        database_url: "postgres://localhost/db".to_string(),
                        api_key: "secret-key-12345".to_string(),
                        max_connections: 100,
                        timeout_ms: 5000,
                    });

                    config_tx.announce(config).unwrap();

                    // Wait for all services to process config
                    for handle in handles {
                        black_box(handle.await.unwrap());
                    }
                });
            },
        );
    }

    group.finish();
}

// ============================================================================
// EVENT FANOUT PATTERN - Broadcast events to many subscribers
// ============================================================================

#[derive(Clone, Debug)]
#[allow(dead_code)]
struct Event {
    event_type: String,
    timestamp: u64,
    data: Vec<u8>,
}

fn bench_event_fanout(c: &mut Criterion) {
    let mut group = c.benchmark_group("pattern/event_fanout");

    for num_subscribers in [10, 100, 1_000].iter() {
        group.throughput(Throughput::Elements(*num_subscribers as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_subscribers),
            num_subscribers,
            |b, &n| {
                b.iter(|| {
                    let (event_tx, event_announcement) = Announcement::new();

                    // Create N subscriber listeners
                    let mut subscribers = Vec::with_capacity(n);
                    for _ in 0..n {
                        subscribers.push(event_announcement.listener());
                    }

                    // Broadcast event
                    let event = Arc::new(Event {
                        event_type: "user.created".to_string(),
                        timestamp: 1234567890,
                        data: vec![1, 2, 3, 4, 5],
                    });

                    event_tx.announce(event).unwrap();

                    // All subscribers receive event
                    for subscriber in &subscribers {
                        black_box(subscriber.try_listen());
                    }
                });
            },
        );
    }

    group.finish();
}

fn bench_event_fanout_async(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("pattern/event_fanout_async");
    group.sample_size(10);
    group.measurement_time(Duration::from_secs(10));

    for num_subscribers in [10, 100].iter() {
        group.throughput(Throughput::Elements(*num_subscribers as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_subscribers),
            num_subscribers,
            |b, &n| {
                b.to_async(&rt).iter(|| async move {
                    let (event_tx, event_announcement) = Announcement::new();

                    // Spawn N subscriber tasks
                    let mut handles = Vec::with_capacity(n);
                    for i in 0..n {
                        let listener = event_announcement.listener();
                        handles.push(tokio::spawn(async move {
                            let event: Arc<Event> = listener.listen().await.unwrap();
                            (i, event.timestamp)
                        }));
                    }

                    // Small delay
                    tokio::time::sleep(Duration::from_millis(1)).await;

                    // Broadcast event
                    let event = Arc::new(Event {
                        event_type: "user.created".to_string(),
                        timestamp: 1234567890,
                        data: vec![1, 2, 3, 4, 5],
                    });

                    event_tx.announce(event).unwrap();

                    // Wait for all subscribers
                    for handle in handles {
                        black_box(handle.await.unwrap());
                    }
                });
            },
        );
    }

    group.finish();
}

// ============================================================================
// REQUEST-RESPONSE MULTIPLEXING - Multiple waiters for single response
// ============================================================================

#[derive(Clone, Debug)]
#[allow(dead_code)]
struct Response {
    status: u16,
    body: String,
}

fn bench_request_response_multiplex(c: &mut Criterion) {
    let mut group = c.benchmark_group("pattern/request_response_multiplex");

    // Multiple tasks waiting for same HTTP response
    for num_waiters in [2, 5, 10, 50, 100].iter() {
        group.throughput(Throughput::Elements(*num_waiters as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_waiters),
            num_waiters,
            |b, &n| {
                b.iter(|| {
                    let (response_tx, response_announcement) = Announcement::new();

                    // Create N waiters for the same response
                    let mut waiters = Vec::with_capacity(n);
                    for _ in 0..n {
                        waiters.push(response_announcement.listener());
                    }

                    // Simulate response arrives
                    let response = Arc::new(Response {
                        status: 200,
                        body: "OK".to_string(),
                    });

                    response_tx.announce(response).unwrap();

                    // All waiters get response
                    for waiter in &waiters {
                        black_box(waiter.try_listen());
                    }
                });
            },
        );
    }

    group.finish();
}

fn bench_request_response_multiplex_async(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("pattern/request_response_multiplex_async");
    group.sample_size(10);

    for num_waiters in [2, 5, 10, 50, 100].iter() {
        group.throughput(Throughput::Elements(*num_waiters as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_waiters),
            num_waiters,
            |b, &n| {
                b.to_async(&rt).iter(|| async move {
                    let (response_tx, response_announcement) = Announcement::new();

                    // Spawn N tasks waiting for response
                    let mut handles = Vec::with_capacity(n);
                    for i in 0..n {
                        let listener = response_announcement.listener();
                        handles.push(tokio::spawn(async move {
                            let response: Arc<Response> = listener.listen().await.unwrap();
                            (i, response.status)
                        }));
                    }

                    // Response arrives
                    let response = Arc::new(Response {
                        status: 200,
                        body: "OK".to_string(),
                    });

                    response_tx.announce(response).unwrap();

                    // Wait for all waiters
                    for handle in handles {
                        black_box(handle.await.unwrap());
                    }
                });
            },
        );
    }

    group.finish();
}

// ============================================================================
// LAZY INITIALIZATION PATTERN - Signal when resource is ready
// ============================================================================

#[derive(Clone, Debug)]
#[allow(dead_code)]
struct Database {
    connection_pool: Arc<Vec<String>>,
    max_connections: usize,
}

fn bench_lazy_initialization(c: &mut Criterion) {
    let mut group = c.benchmark_group("pattern/lazy_initialization");

    for num_clients in [10, 100].iter() {
        group.throughput(Throughput::Elements(*num_clients as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_clients),
            num_clients,
            |b, &n| {
                b.iter(|| {
                    let (db_ready_tx, db_ready_announcement) = Announcement::new();

                    // N clients waiting for database to be ready
                    let mut clients = Vec::with_capacity(n);
                    for _ in 0..n {
                        clients.push(db_ready_announcement.listener());
                    }

                    // Simulate database initialization
                    let db = Arc::new(Database {
                        connection_pool: Arc::new(vec![
                            "conn1".to_string(),
                            "conn2".to_string(),
                            "conn3".to_string(),
                        ]),
                        max_connections: 10,
                    });

                    // Signal ready
                    db_ready_tx.announce(db).unwrap();

                    // All clients can now proceed
                    for client in &clients {
                        black_box(client.try_listen());
                    }
                });
            },
        );
    }

    group.finish();
}

fn bench_lazy_initialization_async(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("pattern/lazy_initialization_async");
    group.sample_size(10);

    for num_clients in [10, 100].iter() {
        group.throughput(Throughput::Elements(*num_clients as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_clients),
            num_clients,
            |b, &n| {
                b.to_async(&rt).iter(|| async move {
                    let (db_ready_tx, db_ready_announcement) = Announcement::new();

                    // Spawn N client tasks
                    let mut handles = Vec::with_capacity(n);
                    for i in 0..n {
                        let listener = db_ready_announcement.listener();
                        handles.push(tokio::spawn(async move {
                            // Wait for database
                            let db: Arc<Database> = listener.listen().await.unwrap();
                            (i, db.max_connections)
                        }));
                    }

                    let db = Arc::new(Database {
                        connection_pool: Arc::new(vec![
                            "conn1".to_string(),
                            "conn2".to_string(),
                            "conn3".to_string(),
                        ]),
                        max_connections: 10,
                    });

                    db_ready_tx.announce(db).unwrap();

                    // Wait for all clients
                    for handle in handles {
                        black_box(handle.await.unwrap());
                    }
                });
            },
        );
    }

    group.finish();
}

// ============================================================================
// BARRIER PATTERN - Multiple announcements that all must complete
// ============================================================================

fn bench_barrier_pattern(c: &mut Criterion) {
    let mut group = c.benchmark_group("pattern/barrier");

    // N tasks each announce completion, all wait for all to complete
    for num_tasks in [5, 10, 25].iter() {
        group.throughput(Throughput::Elements(*num_tasks as u64));
        group.bench_with_input(
            BenchmarkId::from_parameter(num_tasks),
            num_tasks,
            |b, &n| {
                b.iter(|| {
                    let mut announcements = Vec::with_capacity(n);
                    let mut announcers = Vec::with_capacity(n);

                    // Create N announcement channels (one per task)
                    for _ in 0..n {
                        let (tx, announcement) = Announcement::<Arc<i32>>::new();
                        announcers.push(tx);
                        announcements.push(announcement);
                    }

                    // Each task listens to all other tasks
                    let mut all_listeners = Vec::with_capacity(n * n);
                    for announcement in &announcements {
                        for _ in 0..n {
                            all_listeners.push(announcement.listener());
                        }
                    }

                    // All tasks complete
                    for (i, announcer) in announcers.into_iter().enumerate() {
                        announcer.announce(Arc::new(i as i32)).unwrap();
                    }

                    // Verify all received
                    for listener in &all_listeners {
                        black_box(listener.try_listen());
                    }
                });
            },
        );
    }

    group.finish();
}

criterion_group!(
    shutdown,
    bench_shutdown_signal,
    bench_shutdown_with_tokio_tasks,
);

criterion_group!(config, bench_config_broadcast, bench_config_broadcast_async,);

criterion_group!(events, bench_event_fanout, bench_event_fanout_async,);

criterion_group!(
    multiplex,
    bench_request_response_multiplex,
    bench_request_response_multiplex_async,
);

criterion_group!(
    lazy_init,
    bench_lazy_initialization,
    bench_lazy_initialization_async,
);

criterion_group!(barrier, bench_barrier_pattern,);

criterion_main!(shutdown, config, events, multiplex, lazy_init, barrier);