#![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 sasl_plain_admin_producer_and_consumer_round_trip() -> Result<()> {
    let kafka = KafkaHarness::start_sasl_plain().await?;
    let topic = unique_topic("sasl-plain-roundtrip");
    let username = kafka.sasl_username()?.to_owned();
    let password = kafka.sasl_password()?.to_owned();

    let admin = KafkaAdmin::connect(
        AdminConfig::new(kafka.bootstrap_server())
            .with_client_id("integration-sasl-admin")
            .with_sasl_plain(username.clone(), password.clone()),
    )
    .await?;
    admin
        .create_topics([NewTopic::new(topic.clone(), 1, 1)])
        .await?;

    let producer = KafkaProducer::connect(
        ProducerConfig::new(kafka.bootstrap_server())
            .with_client_id("integration-sasl-producer")
            .with_sasl_plain(username.clone(), password.clone()),
    )
    .await?;
    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(
            kafka.bootstrap_server(),
            unique_group("sasl-plain-roundtrip"),
        )
        .with_client_id("integration-sasl-consumer")
        .with_sasl_plain(username, password)
        .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    consumer.subscribe(vec![topic.clone()]).await?;

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

    let records = poll_until(
        &consumer,
        |records| {
            records.iter().any(|record| {
                record.topic == topic && record.value.as_deref() == Some(&b"sasl payload"[..])
            })
        },
        Duration::from_secs(20),
    )
    .await?;
    assert!(records.iter().any(|record| {
        record.topic == topic && record.value.as_deref() == Some(&b"sasl payload"[..])
    }));

    consumer.shutdown().await?;
    producer.shutdown().await?;
    Ok(())
}
#[tokio::test(flavor = "multi_thread")]
async fn sasl_scram_admin_producer_and_consumer_round_trip() -> Result<()> {
    let kafka = KafkaHarness::start_sasl_scram_sha_256().await?;
    let topic = unique_topic("sasl-scram-roundtrip");
    let username = kafka.sasl_username()?.to_owned();
    let password = kafka.sasl_password()?.to_owned();

    let admin = KafkaAdmin::connect(
        AdminConfig::new(kafka.bootstrap_server())
            .with_client_id("integration-scram-admin")
            .with_sasl_scram_sha_256(username.clone(), password.clone()),
    )
    .await?;
    admin
        .create_topics([NewTopic::new(topic.clone(), 1, 1)])
        .await?;

    let producer = KafkaProducer::connect(
        ProducerConfig::new(kafka.bootstrap_server())
            .with_client_id("integration-scram-producer")
            .with_sasl_scram_sha_256(username.clone(), password.clone()),
    )
    .await?;
    let consumer = KafkaConsumer::connect(
        ConsumerConfig::new(
            kafka.bootstrap_server(),
            unique_group("sasl-scram-roundtrip"),
        )
        .with_client_id("integration-scram-consumer")
        .with_sasl_scram_sha_256(username, password)
        .with_auto_offset_reset(AutoOffsetReset::Earliest),
    )
    .await?;
    consumer.subscribe(vec![topic.clone()]).await?;

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

    let records = poll_until(
        &consumer,
        |records| {
            records.iter().any(|record| {
                record.topic == topic && record.value.as_deref() == Some(&b"scram payload"[..])
            })
        },
        Duration::from_secs(20),
    )
    .await?;
    assert!(records.iter().any(|record| {
        record.topic == topic && record.value.as_deref() == Some(&b"scram payload"[..])
    }));

    consumer.shutdown().await?;
    producer.shutdown().await?;
    Ok(())
}
