// Keep this as a separate integration-test binary because failpoints are
// process-global. If this test panics or leaves a failpoint configured, the
// contamination stays within this binary instead of breaking unrelated unit
// tests in tici_shard.
use std::{fs, path::Path};

use anyhow::{Context, Result};
use fail::FailScenario;
use tantivy::{Index, TantivyDocument};
use tempfile::TempDir;
use tici_shard::{
    compactor::merge_utils::rebuild_index,
    test_utils::TEST_INDEX_META_OVERLAP,
    utils::{
        cmp_doc::sort_docs,
        create_index_writer, frag_index_path,
        schema::{HANDLE_STORE_NAME, IndexMeta, VERSION_STORE_NAME},
    },
};

struct MergeTestDoc {
    handle: String,
    name: String,
    description: String,
    version: u64,
}

impl MergeTestDoc {
    fn new(id: usize, description: String) -> Self {
        let handle = format!("handle_{id:06}");
        Self {
            name: format!("name_{id:06}"),
            description,
            handle,
            version: 1,
        }
    }
}

fn get_schema_field(
    schema: &tantivy::schema::Schema,
    field_name: &str,
) -> Result<tantivy::schema::Field> {
    schema
        .get_field(field_name)
        .with_context(|| format!("Field {} not found in schema", field_name))
}

fn build_test_doc(index_meta: &IndexMeta, test_doc: &MergeTestDoc) -> Result<TantivyDocument> {
    let schema = index_meta.get_tantivy_schema();
    let mut doc = TantivyDocument::default();
    doc.add_bytes(
        get_schema_field(schema, HANDLE_STORE_NAME)?,
        test_doc.handle.as_bytes(),
    );
    doc.add_u64(
        get_schema_field(schema, VERSION_STORE_NAME)?,
        test_doc.version,
    );
    doc.add_text(get_schema_field(schema, "column_1")?, &test_doc.name);
    doc.add_text(get_schema_field(schema, "column_2")?, &test_doc.description);
    doc.add_text(get_schema_field(schema, "column_3")?, &test_doc.handle);
    doc.add_text(get_schema_field(schema, "column_4")?, &test_doc.handle);
    Ok(doc)
}

fn write_input_frag(
    parent_dir: &Path,
    frag_name: &str,
    index_meta: &IndexMeta,
    docs: &[MergeTestDoc],
) -> Result<String> {
    let frag_path = parent_dir.join(frag_name);
    let frag_path = frag_path
        .to_str()
        .context("frag path to_str failed")?
        .to_string();
    let index_path = frag_index_path(&frag_path);
    fs::create_dir_all(&index_path)
        .with_context(|| format!("Failed to create index directory idx_path={}", index_path))?;

    let mut index = Index::create_in_dir(&index_path, index_meta.get_tantivy_schema().clone())
        .with_context(|| format!("Failed to create index idx_path={}", index_path))?;
    index.set_tokenizers(index_meta.get_analyzer_manager().clone());
    let mut index_writer = create_index_writer(&index, 1, 50_000_000, &frag_path)?;

    let tantivy_docs = docs
        .iter()
        .map(|doc| build_test_doc(index_meta, doc))
        .collect::<Result<Vec<_>>>()?;
    let tantivy_docs = sort_docs(tantivy_docs, index_meta.get_sort_fields());
    for doc in tantivy_docs {
        index_writer.add_document(doc)?;
    }
    index_writer.commit()?;
    index_writer.wait_merging_threads()?;
    Ok(frag_path)
}

fn prepare_output_frag(parent_dir: &Path, frag_name: &str) -> Result<String> {
    let frag_path = parent_dir.join(frag_name);
    let frag_path = frag_path
        .to_str()
        .context("output frag path to_str failed")?
        .to_string();
    fs::create_dir_all(frag_index_path(&frag_path))
        .with_context(|| format!("Failed to create output index dir for {}", frag_path))?;
    Ok(frag_path)
}

#[test]
fn test_rebuild_index_propagates_add_document_error() -> Result<()> {
    let _fail_scenario_guard = FailScenario::setup();
    let temp_dir = TempDir::new()?;
    let docs = (0..20_000)
        .map(|i| {
            MergeTestDoc::new(
                i,
                format!(
                    "hellohappytaxpayerlongtokenblabla{i} \
                     anotheruniquetokenblabla{i} \
                     yetanotheruniquetokenblabla{i}"
                ),
            )
        })
        .collect::<Vec<_>>();
    let input_frag = write_input_frag(temp_dir.path(), "input", &TEST_INDEX_META_OVERLAP, &docs)?;
    let output_frag = prepare_output_frag(temp_dir.path(), "output")?;
    fail::cfg("FieldSerializer::close_term", "return(simulatederror)").unwrap();

    let err = rebuild_index(
        &[input_frag],
        &TEST_INDEX_META_OVERLAP,
        &output_frag,
        1,
        15_000_000,
        Vec::new(),
    )
    .expect_err("failpoint should make add_document fail during rebuild");
    assert!(
        err.to_string().contains("index add_document failed"),
        "err={err:?}"
    );
    Ok(())
}