use std::{
    fs::{self, File},
    path::Path,
    sync::Arc,
};

use anyhow::Result;
use aws_sdk_s3::Client as S3Client;
use lazy_static::lazy_static;
use storage_backend::{
    s3::s3_upload_object,
    storage::{BackendStorage, create_backend_storage_for_test, s3_storage::create_s3_client},
};
use tici_common::config::S3Config;
use zip::ZipArchive;

use crate::{
    config::FragBatchUploaderConfig,
    utils::s3_frag::{list_all_files_in_relative, to_s3_path},
    writer::frag_batch_uploader::FragBatchUploader,
};

pub const TEST_S3_BUCKET: &str = "cdc-test-bucket";
pub const TEST_S3_BUCKET_PARTITION: &str = "cdc-test-bucket-partition";
lazy_static! {
    pub static ref TEST_S3_PREFIX: String = {
        Path::new(file!())
            .with_extension("")
            .to_str()
            .unwrap()
            .to_string()
    };
}

const TEST_DATA_ZIP_ARCHIVE: &str = "../../ci/cdc_test_data.zip";
const TEST_DATA_ZIP_ARCHIVE_PARTITION: &str = "../../ci/cdc_test_partition_data.zip";

fn test_prepare_data(zip_path: &str) -> Result<String> {
    let zip_path = Path::new(zip_path);
    let parent_dir = zip_path.parent().unwrap_or_else(|| Path::new("."));
    let file = File::open(zip_path)?;
    let mut archive = ZipArchive::new(file)?;

    let target_dir = {
        assert!(!archive.is_empty());
        let file = archive.by_index(0)?;
        let target_dir = parent_dir.join(file.mangled_name());
        assert!(file.is_dir());
        match fs::create_dir_all(&target_dir) {
            Ok(_) => {}
            Err(e) => {
                if e.kind() != std::io::ErrorKind::AlreadyExists {
                    return Err(e.into());
                } else {
                    return Ok(target_dir.to_str().unwrap().to_string());
                }
            }
        }
        target_dir
    };

    for i in 1..archive.len() {
        let mut file = archive.by_index(i)?;
        let outpath = parent_dir.join(file.mangled_name());
        if file.is_dir() {
            fs::create_dir_all(&outpath)?;
        } else {
            if let Some(p) = outpath.parent() {
                if !p.exists() {
                    fs::create_dir_all(p)?;
                }
            }
            let mut outfile = File::create(&outpath)?;
            std::io::copy(&mut file, &mut outfile)?;
        }
    }

    Ok(target_dir.to_str().unwrap().to_string())
}

async fn test_create_bucket(s3_client: &S3Client, bucket_name: &str) -> Result<()> {
    let create_bucket = s3_client
        .create_bucket()
        .bucket(bucket_name.to_string())
        .send()
        .await;
    if let Err(e) = create_bucket {
        let sdk_error = e.into_service_error();
        if !sdk_error.is_bucket_already_exists() && !sdk_error.is_bucket_already_owned_by_you() {
            return Err(sdk_error.into());
        }
    }
    Ok(())
}

async fn upload_dir(
    s3_client: &S3Client,
    s3_bucket: &str,
    s3_prefix: &str,
    local_prefix_path: &str,
    local_path: &str,
) -> Result<Vec<String>> {
    let s3_path = to_s3_path(s3_prefix, local_prefix_path, local_path)?;
    let relative_fnames = list_all_files_in_relative(local_path)?;
    let mut s3_keys = Vec::new();
    for fname in relative_fnames {
        let local_fname = format!("{}/{}", local_path, fname);
        let s3_key = format!("{}/{}", s3_path, fname);
        s3_upload_object(s3_client, s3_bucket, &s3_key, &local_fname).await?;
        s3_keys.push(s3_key);
    }
    Ok(s3_keys)
}

async fn ensure_test_fixture_uploaded(
    s3_client: &S3Client,
    bucket: &str,
    zip_path: &str,
    required_prefix: &str,
) -> Result<()> {
    test_create_bucket(s3_client, bucket).await?;
    let list_resp = s3_client
        .list_objects_v2()
        .bucket(bucket)
        .prefix(required_prefix.to_string())
        .send()
        .await?;
    if list_resp.contents().is_empty() {
        let path = test_prepare_data(zip_path)?;
        upload_dir(s3_client, bucket, &TEST_S3_PREFIX, &path, &path).await?;
        fs::remove_dir_all(path)?;
    }
    Ok(())
}

async fn test_init_s3_directory(s3_client: &S3Client) -> Result<()> {
    // Prepare normal table data. Check the concrete fixture sub-prefix instead
    // of the shared TEST_S3_PREFIX root, because other tests may already have
    // written fragment data under the same prefix.
    ensure_test_fixture_uploaded(
        s3_client,
        TEST_S3_BUCKET,
        TEST_DATA_ZIP_ARCHIVE,
        &format!("{}/123", TEST_S3_PREFIX.as_str()),
    )
    .await?;

    // Prepare partition table data
    ensure_test_fixture_uploaded(
        s3_client,
        TEST_S3_BUCKET_PARTITION,
        TEST_DATA_ZIP_ARCHIVE_PARTITION,
        &format!("{}/437", TEST_S3_PREFIX.as_str()),
    )
    .await?;
    Ok(())
}

static TEST_S3_DIR_INITIALIZED: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new();

async fn test_create_s3_client() -> Result<Arc<S3Client>> {
    let mut config = S3Config::default_for_test();
    config.bucket = TEST_S3_BUCKET.to_string();
    config.prefix = TEST_S3_PREFIX.to_string();
    let s3_client = Arc::new(create_s3_client(config).await?);

    TEST_S3_DIR_INITIALIZED
        .get_or_init(|| async {
            test_init_s3_directory(&s3_client).await.unwrap();
        })
        .await;

    Ok(s3_client)
}

pub async fn test_create_backend_storage() -> Result<Arc<dyn BackendStorage>> {
    // Ensure S3 test data is initialized first
    let _ = test_create_s3_client().await?;

    let mut config = S3Config::default_for_test();
    config.bucket = TEST_S3_BUCKET.to_string();
    config.prefix = TEST_S3_PREFIX.to_string();
    Ok(create_backend_storage_for_test(&config).await?)
}

pub async fn test_get_frag_batch_uploader() -> Result<Arc<FragBatchUploader>> {
    let backend_storage = test_create_backend_storage().await?;
    Ok(Arc::new(FragBatchUploader::new(
        backend_storage,
        FragBatchUploaderConfig::default(),
    )))
}