#![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 transactional_producer_commits_multiple_v2_transactions() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let topic = unique_topic("txn-commit");
    kafka.create_topic(&topic).await?;

    let producer = KafkaProducer::connect(
        ProducerConfig::new(kafka.bootstrap_server())
            .with_client_id("integration-transactional-producer")
            .with_transactional_id(unique_group("txn-producer")),
    )
    .await?;

    producer.begin_transaction().await?;
    producer
        .send(ProduceRecord::new(
            topic.clone(),
            0,
            Bytes::from_static(b"first transaction"),
        ))
        .await?;
    producer.commit_transaction().await?;

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

    let values = kafka
        .consume_values(&topic, "read_committed", 2, 10_000)
        .await?;
    assert_eq!(
        values,
        vec![
            "first transaction".to_owned(),
            "second transaction".to_owned()
        ]
    );
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn transactional_producer_supports_explicit_init_transactions() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let topic = unique_topic("txn-init");
    kafka.create_topic(&topic).await?;

    let producer = KafkaProducer::connect(
        ProducerConfig::new(kafka.bootstrap_server())
            .with_client_id("integration-transactional-init-producer")
            .with_transactional_id(unique_group("txn-init-producer")),
    )
    .await?;

    producer.init_transactions().await?;
    producer.begin_transaction().await?;
    producer
        .send(ProduceRecord::new(
            topic.clone(),
            0,
            Bytes::from_static(b"initialized transaction"),
        ))
        .await?;
    producer.commit_transaction().await?;
    producer.shutdown().await?;

    let values = kafka
        .consume_values(&topic, "read_committed", 1, 10_000)
        .await?;
    assert_eq!(values, vec!["initialized transaction".to_owned()]);
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn transactional_producer_can_send_offsets_to_transaction_for_modern_groups() -> Result<()> {
    let kafka = KafkaHarness::start().await?;
    let input_topic = unique_topic("txn-offset-input");
    let output_topic = unique_topic("txn-offset-output");
    let consumer_group = unique_group("txn-offset-group");
    kafka.create_topic(&input_topic).await?;
    kafka.create_topic(&output_topic).await?;

    let seed_producer =
        KafkaProducer::connect(ProducerConfig::new(kafka.bootstrap_server())).await?;
    seed_producer
        .send(ProduceRecord::new(
            input_topic.clone(),
            0,
            Bytes::from_static(b"input-record"),
        ))
        .await?;
    seed_producer.shutdown().await?;

    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), consumer_group.clone())
            .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    consumer.subscribe(vec![input_topic.clone()]).await?;
    let consumed = poll_until(
        &consumer,
        |records| {
            records
                .iter()
                .any(|record| record.value.as_deref() == Some(&b"input-record"[..]))
        },
        Duration::from_secs(20),
    )
    .await?;
    let group_metadata = consumer.group_metadata().await?;
    assert!(group_metadata.generation_id > 0);
    assert!(!group_metadata.member_id.is_empty());

    let transactional_producer = KafkaProducer::connect(
        ProducerConfig::new(kafka.bootstrap_server())
            .with_transactional_id(unique_group("txn-offset-producer")),
    )
    .await?;
    transactional_producer.init_transactions().await?;
    transactional_producer.begin_transaction().await?;
    transactional_producer
        .send(ProduceRecord::new(
            output_topic.clone(),
            0,
            Bytes::from_static(b"transaction-output"),
        ))
        .await?;
    transactional_producer
        .send_offsets_to_transaction(consumed.commit_offsets(), group_metadata)
        .await?;
    transactional_producer.commit_transaction().await?;

    let values = kafka
        .consume_values(&output_topic, "read_committed", 1, 10_000)
        .await?;
    assert_eq!(values, vec!["transaction-output".to_owned()]);

    consumer.shutdown().await?;
    transactional_producer.shutdown().await?;

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

    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        let records = verifying_consumer.poll_for(Duration::from_secs(1)).await?;
        if records
            .iter()
            .any(|record| record.value.as_deref() == Some(&b"input-record"[..]))
        {
            bail!("transactional offset commit did not advance the consumer group position");
        }
        if Instant::now() >= deadline {
            break;
        }
    }

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

    let producer = KafkaProducer::connect(
        ProducerConfig::new(kafka.bootstrap_server())
            .with_transactional_id(unique_group("read-committed-producer")),
    )
    .await?;
    producer.init_transactions().await?;

    producer.begin_transaction().await?;
    producer
        .send(ProduceRecord::new(
            topic.clone(),
            0,
            Bytes::from_static(b"committed-value"),
        ))
        .await?;
    producer.commit_transaction().await?;

    producer.begin_transaction().await?;
    producer
        .send(ProduceRecord::new(
            topic.clone(),
            0,
            Bytes::from_static(b"aborted-value"),
        ))
        .await?;
    producer.abort_transaction().await?;
    producer.shutdown().await?;

    let read_uncommitted = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), unique_group("read-uncommitted"))
            .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    read_uncommitted.subscribe(vec![topic.clone()]).await?;
    let uncommitted_values = collect_values(&read_uncommitted, Duration::from_secs(10), 2).await?;
    assert!(
        uncommitted_values
            .iter()
            .any(|value| value == "committed-value")
    );
    assert!(
        uncommitted_values
            .iter()
            .any(|value| value == "aborted-value")
    );
    read_uncommitted.shutdown().await?;

    let read_committed = KafkaConsumer::connect(
        ConsumerConfig::new(kafka.bootstrap_server(), unique_group("read-committed"))
            .with_auto_offset_reset(AutoOffsetReset::Earliest)
            .with_isolation_level(IsolationLevel::ReadCommitted),
    )
    .await?;
    read_committed.subscribe(vec![topic.clone()]).await?;
    let committed_values = collect_values(&read_committed, Duration::from_secs(5), 2).await?;
    assert_eq!(committed_values, vec!["committed-value".to_owned()]);
    read_committed.shutdown().await?;
    Ok(())
}
