#![allow(unused_imports)]

mod common;

use std::sync::Arc;
use std::time::{Duration, Instant};

use anyhow::{Result, anyhow, bail};
use bytes::Bytes;
use kafkit_client::{
    AdminConfig, AlterConfigOp, AutoOffsetReset, ConfigResource, ConsumerConfig, ConsumerError,
    IsolationLevel, KafkaAdmin, KafkaClient, KafkaConsumer, KafkaMessage, KafkaProducer,
    NewPartitions, NewTopic, ProduceRecord, ProducerCompression, ProducerConfig, RecordHeader,
    SubscriptionPattern, TopicPartition, TopicPartitionTimestamp,
};

use crate::common::{
    KafkaHarness, collect_values, expected_assignment, poll_describe_until, poll_until,
    poll_until_assignment, poll_until_consumer_error, poll_until_with_admin, poll_until_with_api,
    unique_group, unique_topic,
};

#[tokio::test(flavor = "multi_thread")]
async fn consumer_query_apis_expose_offsets_metadata_and_seek_by_timestamp() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let topic = unique_topic("consumer-query-apis");
    let admin = KafkaAdmin::connect(AdminConfig::new(kafka.bootstrap_server())).await?;
    admin
        .create_topics([NewTopic::new(topic.clone(), 2, 1)])
        .await?;

    let producer = KafkaProducer::connect(ProducerConfig::new(kafka.bootstrap_server())).await?;
    producer
        .send(
            ProduceRecord::new(topic.clone(), 0, Bytes::from_static(b"first"))
                .with_timestamp(1_700_000_000_100),
        )
        .await?;
    producer
        .send(
            ProduceRecord::new(topic.clone(), 0, Bytes::from_static(b"second"))
                .with_timestamp(1_700_000_000_200),
        )
        .await?;
    producer
        .send(
            ProduceRecord::new(topic.clone(), 0, Bytes::from_static(b"third"))
                .with_timestamp(1_700_000_000_300),
        )
        .await?;

    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(
            kafka.bootstrap_server(),
            unique_group("consumer-query-apis"),
        )
        .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    let partition = TopicPartition::new(topic.clone(), 0);
    consumer.assign(vec![partition.clone()]).await?;

    let beginning = consumer.beginning_offsets(vec![partition.clone()]).await?;
    assert_eq!(beginning.len(), 1);
    assert_eq!(beginning[0].offset, 0);

    let end = consumer.end_offsets(vec![partition.clone()]).await?;
    assert_eq!(end.len(), 1);
    assert_eq!(end[0].offset, 3);

    let looked_up = consumer
        .offsets_for_times(vec![TopicPartitionTimestamp::new(
            topic.clone(),
            0,
            1_700_000_000_200,
        )])
        .await?;
    assert_eq!(looked_up.len(), 1);
    assert_eq!(looked_up[0].offset, 1);
    assert_eq!(looked_up[0].timestamp, 1_700_000_000_200);

    consumer
        .seek_to_timestamp(vec![TopicPartitionTimestamp::new(
            topic.clone(),
            0,
            1_700_000_000_200,
        )])
        .await?;
    let replayed = poll_until(
        &consumer,
        |records| {
            records
                .iter()
                .any(|record| record.value.as_deref() == Some(&b"second"[..]))
        },
        Duration::from_secs(20),
    )
    .await?;
    assert!(
        replayed
            .iter()
            .any(|record| record.value.as_deref() == Some(&b"second"[..]))
    );

    let topics = consumer.list_topics().await?;
    assert!(topics.iter().any(|listed| listed == &topic));

    let partitions = consumer.partitions_for(topic.clone()).await?;
    assert_eq!(partitions.len(), 2);
    assert!(partitions.iter().all(|partition| partition.leader_id >= 0));

    consumer.shutdown().await?;
    producer.shutdown().await?;
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn consumer_committed_offsets_are_visible_after_restart() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let topic = unique_topic("consumer-committed-query");
    let group = unique_group("consumer-committed-query");
    kafka.create_topic(&topic).await?;

    let producer = KafkaProducer::connect(ProducerConfig::new(kafka.bootstrap_server())).await?;
    producer
        .send(ProduceRecord::new(
            topic.clone(),
            0,
            Bytes::from_static(b"first"),
        ))
        .await?;

    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), group.clone())
            .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    consumer.subscribe(vec![topic.clone()]).await?;
    let first_batch = poll_until(
        &consumer,
        |records| {
            records
                .iter()
                .any(|record| record.value.as_deref() == Some(&b"first"[..]))
        },
        Duration::from_secs(20),
    )
    .await?;
    consumer.commit(&first_batch).await?;

    let partition = TopicPartition::new(topic.clone(), 0);
    let committed = consumer.committed(vec![partition.clone()]).await?;
    assert_eq!(committed.len(), 1);
    assert_eq!(committed[0].offset, 1);
    consumer.shutdown().await?;

    let restarted = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), group)
            .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    let committed = restarted.committed(vec![partition]).await?;
    assert_eq!(committed.len(), 1);
    assert_eq!(committed[0].offset, 1);
    restarted.shutdown().await?;
    producer.shutdown().await?;
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn consumer_wakeup_interrupts_poll_and_unsubscribe_clears_delivery() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let topic = unique_topic("consumer-wakeup");
    let group = unique_group("consumer-wakeup");
    kafka.create_topic(&topic).await?;

    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), group)
            .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    consumer.subscribe(vec![topic.clone()]).await?;

    let (poll_result, wakeup_result) =
        tokio::join!(consumer.poll_for(Duration::from_secs(30)), async {
            tokio::time::sleep(Duration::from_millis(100)).await;
            consumer.wakeup().await
        });
    wakeup_result?;
    let error = poll_result.expect_err("poll should be interrupted by wakeup");
    assert!(error.to_string().contains("wakeup"));

    let producer = KafkaProducer::connect(ProducerConfig::new(kafka.bootstrap_server())).await?;
    producer
        .send(ProduceRecord::new(
            topic.clone(),
            0,
            Bytes::from_static(b"before-unsubscribe"),
        ))
        .await?;
    let delivered = poll_until(
        &consumer,
        |records| {
            records
                .iter()
                .any(|record| record.value.as_deref() == Some(&b"before-unsubscribe"[..]))
        },
        Duration::from_secs(20),
    )
    .await?;
    assert_eq!(delivered.len(), 1);

    consumer.unsubscribe().await?;
    producer
        .send(ProduceRecord::new(
            topic.clone(),
            0,
            Bytes::from_static(b"after-unsubscribe"),
        ))
        .await?;
    let drained = consumer.poll_for(Duration::from_secs(1)).await?;
    assert!(drained.is_empty());

    consumer.shutdown().await?;
    producer.shutdown().await?;
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn consumer_manual_assignment_supports_seek_position_pause_and_resume() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let topic = unique_topic("manual-assign");
    let partition = TopicPartition::new(topic.clone(), 0);
    kafka.create_topic(&topic).await?;

    let producer = KafkaProducer::connect(ProducerConfig::new(kafka.bootstrap_server())).await?;
    for value in [b"zero".as_slice(), b"one".as_slice(), b"two".as_slice()] {
        producer
            .send(ProduceRecord::new(
                topic.clone(),
                0,
                Bytes::copy_from_slice(value),
            ))
            .await?;
    }

    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), unique_group("manual-assign"))
            .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    consumer.assign(vec![partition.clone()]).await?;

    assert_eq!(consumer.position(partition.clone()).await?, 0);

    consumer.seek(partition.clone(), 1).await?;
    assert_eq!(consumer.position(partition.clone()).await?, 1);

    let records = poll_until(
        &consumer,
        |records| {
            records
                .iter()
                .any(|record| record.value.as_deref() == Some(&b"one"[..]))
        },
        Duration::from_secs(20),
    )
    .await?;
    let values = records
        .iter()
        .filter_map(|record| record.value.as_deref())
        .map(|value| String::from_utf8_lossy(value).into_owned())
        .collect::<Vec<_>>();
    assert_eq!(values, vec!["one".to_owned(), "two".to_owned()]);
    assert_eq!(consumer.position(partition.clone()).await?, 3);

    consumer.seek_to_beginning(vec![partition.clone()]).await?;
    assert_eq!(consumer.position(partition.clone()).await?, 0);
    let replayed = poll_until(
        &consumer,
        |records| {
            records
                .iter()
                .any(|record| record.value.as_deref() == Some(&b"zero"[..]))
        },
        Duration::from_secs(20),
    )
    .await?;
    assert!(
        replayed
            .iter()
            .any(|record| record.value.as_deref() == Some(&b"zero"[..]))
    );

    consumer.seek_to_end(vec![partition.clone()]).await?;
    assert_eq!(consumer.position(partition.clone()).await?, 3);
    consumer.pause(vec![partition.clone()]).await?;
    producer
        .send(ProduceRecord::new(
            topic.clone(),
            0,
            Bytes::from_static(b"after-pause"),
        ))
        .await?;

    let paused_batch = consumer.poll_for(Duration::from_secs(1)).await?;
    assert!(paused_batch.is_empty());
    assert_eq!(consumer.position(partition.clone()).await?, 3);

    consumer.resume(vec![partition.clone()]).await?;
    let resumed = poll_until(
        &consumer,
        |records| {
            records
                .iter()
                .any(|record| record.value.as_deref() == Some(&b"after-pause"[..]))
        },
        Duration::from_secs(20),
    )
    .await?;
    assert!(
        resumed
            .iter()
            .any(|record| record.value.as_deref() == Some(&b"after-pause"[..]))
    );
    assert_eq!(consumer.position(partition.clone()).await?, 4);

    consumer.shutdown().await?;
    producer.shutdown().await?;
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn committed_offsets_survive_consumer_restart() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let topic = unique_topic("commit-restart");
    let group = unique_group("commit-restart");
    kafka.create_topic(&topic).await?;

    let producer = KafkaProducer::connect(
        ProducerConfig::new(kafka.bootstrap_server()).with_compression(ProducerCompression::None),
    )
    .await?;
    producer
        .send(ProduceRecord::new(
            topic.clone(),
            0,
            Bytes::from_static(b"first"),
        ))
        .await?;

    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), group.clone())
            .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    consumer.subscribe(vec![topic.clone()]).await?;
    let first_batch = poll_until(
        &consumer,
        |records| {
            records
                .iter()
                .any(|record| record.value.as_deref() == Some(&b"first"[..]))
        },
        Duration::from_secs(20),
    )
    .await?;
    consumer.commit(&first_batch).await?;
    consumer.shutdown().await?;

    producer
        .send(ProduceRecord::new(
            topic.clone(),
            0,
            Bytes::from_static(b"second"),
        ))
        .await?;

    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), group)
            .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    consumer.subscribe(vec![topic.clone()]).await?;
    let second_batch = poll_until(
        &consumer,
        |records| {
            records
                .iter()
                .any(|record| record.value.as_deref() == Some(&b"second"[..]))
        },
        Duration::from_secs(20),
    )
    .await?;

    assert!(
        second_batch
            .iter()
            .all(|record| record.value.as_deref() != Some(&b"first"[..]))
    );
    assert!(
        second_batch
            .iter()
            .any(|record| record.value.as_deref() == Some(&b"second"[..]))
    );

    consumer.shutdown().await?;
    producer.shutdown().await?;
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn consumer_regex_subscription_lifecycle_matches_topic_updates() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let admin = KafkaAdmin::connect(AdminConfig::new(kafka.bootstrap_server())).await?;
    let group = unique_group("regex-lifecycle");
    let matching_a = unique_topic("regex-match");
    let matching_b = unique_topic("regex-match");
    let other = unique_topic("regex-other");

    admin
        .create_topics([
            NewTopic::new(matching_a.clone(), 2, 1),
            NewTopic::new(other.clone(), 1, 1),
        ])
        .await?;

    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), group)
            .with_metadata_max_age(Duration::from_millis(200)),
    )
    .await?;

    consumer
        .subscribe_regex("regex-match-.*".to_owned())
        .await?;
    poll_until_assignment(
        &consumer,
        expected_assignment(&matching_a, 2),
        Duration::from_secs(20),
    )
    .await?;

    admin
        .create_topics([NewTopic::new(matching_b.clone(), 1, 1)])
        .await?;
    let mut expected = expected_assignment(&matching_a, 2);
    expected.extend(expected_assignment(&matching_b, 1));
    poll_until_assignment(&consumer, expected, Duration::from_secs(20)).await?;

    consumer
        .subscribe_regex("regex-other-.*".to_owned())
        .await?;
    poll_until_assignment(
        &consumer,
        expected_assignment(&other, 1),
        Duration::from_secs(20),
    )
    .await?;

    consumer.unsubscribe().await?;
    let assignment = consumer.assignment().await?;
    assert!(assignment.is_empty());
    consumer.shutdown().await?;
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn consumer_regex_subscription_rejects_invalid_patterns() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let consumer = KafkaConsumer::connect(ConsumerConfig::new(
        kafka.bootstrap_server(),
        unique_group("regex-invalid"),
    ))
    .await?;

    consumer
        .subscribe_pattern(SubscriptionPattern::new("(t.*c"))
        .await?;
    let error = poll_until_consumer_error(&consumer, Duration::from_secs(10)).await?;
    assert!(matches!(
        error,
        kafkit_client::Error::Consumer(ConsumerError::InvalidRegularExpression { .. })
    ));

    let _ = consumer.shutdown().await;
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn consumer_rejects_unsupported_assignor() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let topic = unique_topic("unsupported-assignor");
    kafka.create_topic(&topic).await?;

    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(
            kafka.bootstrap_server(),
            unique_group("unsupported-assignor"),
        )
        .with_server_assignor("not-a-real-assignor"),
    )
    .await?;
    consumer.subscribe(vec![topic]).await?;

    let error = poll_until_consumer_error(&consumer, Duration::from_secs(10)).await?;
    assert!(matches!(
        error,
        kafkit_client::Error::Consumer(ConsumerError::UnsupportedAssignor { .. })
    ));

    let _ = consumer.shutdown().await;
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn consumer_rejects_duplicate_static_members_until_the_original_leaves() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let admin = KafkaAdmin::connect(AdminConfig::new(kafka.bootstrap_server())).await?;
    let topic = unique_topic("static-member-fence");
    let group = unique_group("static-member-fence");
    let instance_id = unique_group("static-instance");

    admin
        .create_topics([NewTopic::new(topic.clone(), 2, 1)])
        .await?;

    let original = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), group.clone())
            .with_instance_id(instance_id.clone()),
    )
    .await?;
    original.subscribe(vec![topic.clone()]).await?;
    let expected = expected_assignment(&topic, 2);
    poll_until_assignment(&original, expected.clone(), Duration::from_secs(20)).await?;

    let duplicate = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), group).with_instance_id(instance_id.clone()),
    )
    .await?;
    duplicate.subscribe(vec![topic]).await?;
    let error = poll_until_consumer_error(&duplicate, Duration::from_secs(10)).await?;
    assert!(matches!(
        error,
        kafkit_client::Error::Consumer(ConsumerError::UnreleasedInstanceId { .. })
    ));

    poll_until_assignment(&original, expected, Duration::from_secs(10)).await?;

    let _ = duplicate.shutdown().await;
    original.shutdown().await?;
    Ok(())
}
