use std::io::{BufWriter, Write};

use crate::tpl::column::{ColumnGen, FkDistribution, ParentCtx};
use seedfaker_core::ctx::{GenContext, Identity};
use seedfaker_core::field::{Ordering, RangeSpec, Transform};
use seedfaker_core::locale::Locale;
use seedfaker_core::rng::Rng;
use seedfaker_core::script::{Corrupt, Ctx, Script};

use crate::config::GenConfig;
use crate::format;
use crate::writers;

pub struct RunOptions {
    pub master_seed: u64,
    pub count: u64,
    pub locales: Vec<&'static Locale>,
    pub script: Script,
    pub ctx: Ctx,
    pub corrupt: Corrupt,
    pub rate: Option<u64>,
    pub no_header: bool,
    pub output: OutputMode,
    pub delim: Option<String>,
    pub validate: bool,
    pub annotated: bool,
    pub tz_offset_minutes: i32,
    pub since: i64,
    pub until: i64,
}

pub enum OutputMode {
    Default,
    Csv,
    Tsv,
    Jsonl,
    Sql(String),
    Template,
}

pub fn format_ext(mode: &OutputMode) -> &'static str {
    match mode {
        OutputMode::Csv => "csv",
        OutputMode::Default | OutputMode::Tsv => "tsv",
        OutputMode::Jsonl => "jsonl",
        OutputMode::Sql(_) => "sql",
        OutputMode::Template => "txt",
    }
}

// ═══════════════════════════════════════════════════════════════════
// FK orchestration
// ═══════════════════════════════════════════════════════════════════

/// Fill `parent_domain_hash` / `deref_domain_hash` / `parent_ctx` on all FK columns.
/// Call on a **clone** of the table's `GenConfig` — never mutate the stored original.
pub fn finalize_fk_columns(
    config: &mut GenConfig,
    global_seed: u64,
    all_tables: &[(String, GenConfig)],
) {
    // Collect (col_index, parent_table) for FkDeref resolution below.
    // We need this snapshot because we'll iterate `config.columns` mutably.
    let anchor_parents: Vec<(String, String)> = config
        .columns
        .iter()
        .filter_map(|c| {
            if let ColumnGen::Fk { parent_table, .. } = &c.gen {
                Some((c.name.clone(), parent_table.clone()))
            } else {
                None
            }
        })
        .collect();

    for col in &mut config.columns {
        match &mut col.gen {
            ColumnGen::Fk {
                parent_table,
                parent_col_name,
                parent_field,
                parent_modifier,
                parent_domain_hash,
                parent_ctx,
                ..
            } => {
                let parent_seed = seedfaker_core::rng::domain_hash(global_seed, parent_table);
                *parent_domain_hash = seedfaker_core::eval::column_domain_hash(
                    parent_seed,
                    parent_col_name,
                    parent_field,
                    parent_modifier,
                );
                **parent_ctx = build_parent_ctx(all_tables, parent_table, global_seed);
            }
            ColumnGen::FkDeref {
                anchor_col,
                deref_col_name,
                deref_field,
                deref_modifier,
                deref_domain_hash,
                parent_ctx,
                ..
            } => {
                let parent_table = anchor_parents
                    .iter()
                    .find(|(name, _)| name == anchor_col)
                    .map_or("", |(_, pt)| pt.as_str());

                let parent_seed = seedfaker_core::rng::domain_hash(global_seed, parent_table);
                *deref_domain_hash = seedfaker_core::eval::column_domain_hash(
                    parent_seed,
                    deref_col_name,
                    deref_field,
                    deref_modifier,
                );
                **parent_ctx = build_parent_ctx(all_tables, parent_table, global_seed);
            }
            _ => {}
        }
    }
}

/// Build a `ParentCtx` by resolving a parent table's stored options.
/// All resolution is done with core functions — no CLI layer needed.
fn build_parent_ctx(
    all_tables: &[(String, GenConfig)],
    parent_table: &str,
    global_seed: u64,
) -> ParentCtx {
    let table_seed = seedfaker_core::rng::domain_hash(global_seed, parent_table);

    let Some((_, cfg)) = all_tables.iter().find(|(n, _)| n == parent_table) else {
        return ParentCtx { table_seed, ..Default::default() };
    };

    let opts = &cfg.options;

    let locales: Vec<&'static Locale> =
        seedfaker_core::locale::resolve(&opts.locale).unwrap_or_default();

    let ctx = match opts.ctx.as_deref() {
        Some(s) => Ctx::parse(s),
        None => Ctx::None,
    };

    let script = match opts.abc.as_deref() {
        Some("native") => Script::Native,
        Some("mixed") => Script::Both,
        _ => Script::Latin,
    };

    let tz_offset = opts.tz.as_deref().and_then(|s| seedfaker_core::tz::parse(s).ok()).unwrap_or(0);

    let since = opts
        .since
        .as_deref()
        .and_then(|s| seedfaker_core::temporal::parse(s).ok())
        .unwrap_or(seedfaker_core::temporal::DEFAULT_SINCE);

    let until = opts
        .until
        .as_deref()
        .and_then(|s| seedfaker_core::temporal::parse_until(s).ok())
        .unwrap_or_else(seedfaker_core::temporal::default_until);

    ParentCtx {
        table_seed,
        locales,
        script,
        ctx,
        tz_offset_minutes: tz_offset,
        since,
        until,
        parent_count: opts.count.unwrap_or(0),
    }
}

fn generate_parent_field(
    field: &'static seedfaker_core::field::Field,
    modifier: &str,
    range: &Option<RangeSpec>,
    domain_hash: u64,
    row_idx: u64,
    parent_ctx: &ParentCtx,
    buf: &mut String,
) -> Option<f64> {
    let resolved =
        seedfaker_core::field::resolve_range(range, field.name, parent_ctx.since, parent_ctx.until);
    let mut ctx = GenContext {
        rng: Rng::derive_fast(domain_hash, row_idx),
        locales: &parent_ctx.locales,
        modifier,
        identity: None,
        tz_offset_minutes: parent_ctx.tz_offset_minutes,
        since: parent_ctx.since,
        until: parent_ctx.until,
        range: resolved,
        ordering: Ordering::None,
        zipf: None,
        numeric: None,
    };
    field.generate(&mut ctx, buf)
}

// ═══════════════════════════════════════════════════════════════════
// run<W: Write>
// ═══════════════════════════════════════════════════════════════════

pub fn run<W: Write>(
    writer: W,
    gen_config: &GenConfig,
    opts: &RunOptions,
) -> Result<(), Box<dyn std::error::Error>> {
    let mut out = BufWriter::with_capacity(256 * 1024, writer);
    let unlimited = opts.count == 0;
    let mut serial: u64 = 0;

    let rate_interval = opts.rate.map(|r| {
        if r == 0 {
            std::time::Duration::ZERO
        } else {
            std::time::Duration::from_secs_f64(1.0 / r as f64)
        }
    });

    let col_names: Vec<String> = gen_config.columns.iter().map(|v| v.name.clone()).collect();
    let col_count = gen_config.columns.len();

    let domain_hashes = compute_domain_hashes(&gen_config.columns, opts.master_seed);

    let mut script_rng = Rng::derive(opts.master_seed, 0, seedfaker_core::DOMAIN_SCRIPT);
    let effective_locales =
        format::apply_script(&opts.locales.clone(), opts.script, &mut script_rng);
    let effective_refs: Vec<&Locale> = effective_locales.iter().collect();
    let locales: &[&Locale] =
        if opts.script == Script::Native { &effective_refs } else { &opts.locales };

    let resolved_ranges: Vec<Option<(i64, i64)>> = gen_config
        .columns
        .iter()
        .map(|v| match &v.gen {
            ColumnGen::Field { field, range, .. } => resolve_range(range, field.name, opts),
            ColumnGen::Literal(_)
            | ColumnGen::Aggr { .. }
            | ColumnGen::Ref { .. }
            | ColumnGen::Expr { .. }
            | ColumnGen::Fk { .. }
            | ColumnGen::FkDeref { .. } => None,
        })
        .collect();

    validate_resolved_ranges(&gen_config.columns, &resolved_ranges)?;

    let birthdate_range: Option<(i64, i64)> =
        gen_config.columns.iter().find_map(|v| match &v.gen {
            ColumnGen::Field { field, range, .. } if field.name == "birthdate" => {
                resolve_range(range, field.name, opts)
            }
            _ => None,
        });

    let compiled_tpl = match gen_config.template.as_ref() {
        Some(tpl) => {
            let mut name_refs: Vec<&str> = col_names.iter().map(String::as_str).collect();
            name_refs.push("serial");
            Some(crate::tpl::compile::compile(tpl, &name_refs)?)
        }
        None => None,
    };
    let is_text = compiled_tpl.is_some();

    let field_types: Vec<&'static str> = gen_config
        .columns
        .iter()
        .map(|c| match &c.gen {
            ColumnGen::Field { field, .. } => field.name,
            ColumnGen::Literal(_) => "literal",
            ColumnGen::Expr { .. } => "expr",
            ColumnGen::Aggr { .. } => "aggr",
            ColumnGen::Ref { .. } => "ref",
            ColumnGen::Fk { .. } => "fk",
            ColumnGen::FkDeref { .. } => "fk_deref",
        })
        .collect();

    let csv_sep_s = opts.delim.clone().unwrap_or_else(|| ",".to_string());
    let tsv_sep_s = opts.delim.clone().unwrap_or_else(|| "\t".to_string());
    let csv_sep = csv_sep_s.as_bytes();
    let tsv_sep = tsv_sep_s.as_bytes();

    let sql_prefix: Option<Vec<u8>> = if let OutputMode::Sql(ref table) = opts.output {
        let safe = writers::sanitize_identifier(table);
        Some(format!("INSERT INTO {} ({}) VALUES ", safe, col_names.join(", ")).into_bytes())
    } else {
        None
    };

    if !opts.no_header && !is_text && !opts.annotated {
        match opts.output {
            OutputMode::Csv => writers::write_header(&mut out, &col_names, csv_sep)?,
            OutputMode::Tsv => writers::write_header(&mut out, &col_names, tsv_sep)?,
            _ => {}
        }
    }

    let needs_ctx = opts.ctx != Ctx::None;

    let mut aggr = crate::aggr::AggrState::new(&gen_config.columns, &col_names)?;

    let extra = usize::from(compiled_tpl.is_some());
    let mut values: Vec<String> =
        (0..col_count + extra).map(|_| String::with_capacity(32)).collect();
    let mut raw_values: Vec<Option<f64>> = vec![None; col_count];
    let mut is_omitted: Vec<bool> = vec![false; col_count];
    let mut fk_row_indices: Vec<Option<u64>> = vec![None; col_count];
    let mut tpl_buf = String::new();
    let mut itoa_buf = itoa::Buffer::new();

    if opts.validate {
        return Ok(());
    }

    loop {
        if !unlimited && serial >= opts.count {
            break;
        }

        let both_locales: Vec<Locale>;
        let both_refs: Vec<&Locale>;
        let use_locales: &[&Locale] = if opts.script == Script::Both {
            let mut sr = Rng::derive(opts.master_seed, serial, seedfaker_core::DOMAIN_SCRIPT);
            both_locales = format::apply_script(&opts.locales, opts.script, &mut sr);
            both_refs = both_locales.iter().collect();
            &both_refs
        } else {
            locales
        };

        let locked_locale: Option<Locale> = match opts.ctx {
            Ctx::None => None,
            Ctx::Strict => {
                let mut lr = Rng::derive(opts.master_seed, serial, seedfaker_core::DOMAIN_LOCALE);
                Some(**lr.choice(use_locales))
            }
            Ctx::Loose => {
                let mut lr = Rng::derive(opts.master_seed, serial, seedfaker_core::DOMAIN_LOCALE);
                if lr.maybe(0.7) {
                    Some(**lr.choice(use_locales))
                } else {
                    None
                }
            }
        };
        let locked_ref: &Locale;
        let locked_slice: [&Locale; 1];
        let record_locales: &[&Locale] = if let Some(ref loc) = locked_locale {
            locked_ref = loc;
            locked_slice = [locked_ref];
            &locked_slice
        } else {
            use_locales
        };

        let identity = if needs_ctx {
            let mut ir = Rng::derive(opts.master_seed, serial, seedfaker_core::DOMAIN_IDENTITY);
            Some(Identity::new(&mut ir, record_locales, birthdate_range, opts.since, opts.until))
        } else {
            None
        };
        let id_ref = identity.as_ref();

        let mut ctx = GenContext {
            rng: Rng::new(0),
            locales: record_locales,
            modifier: "",
            identity: id_ref,
            tz_offset_minutes: opts.tz_offset_minutes,
            since: opts.since,
            until: opts.until,
            range: None,
            ordering: Ordering::None,
            zipf: None,
            numeric: None,
        };

        for i in 0..col_count {
            values[i].clear();
            raw_values[i] = None;
            is_omitted[i] = false;
            fk_row_indices[i] = None;
        }

        for &i in &gen_config.eval_order {
            match &gen_config.columns[i].gen {
                ColumnGen::Field {
                    field, modifier, transform, ordering, omit_pct, zipf, ..
                } => {
                    if let Some(pct) = omit_pct {
                        let mut or = Rng::derive(domain_hashes[i], serial, "omit");
                        if or.range(0, 100) < i64::from(*pct) {
                            is_omitted[i] = true;
                            continue;
                        }
                    }
                    ctx.rng = Rng::derive_fast(domain_hashes[i], serial);
                    ctx.modifier = modifier;
                    ctx.range = resolved_ranges[i];
                    ctx.ordering = *ordering;
                    ctx.zipf = *zipf;
                    raw_values[i] = field.generate(&mut ctx, &mut values[i]);
                    if *transform != Transform::None {
                        let s = std::mem::take(&mut values[i]);
                        values[i] = transform.apply(&s);
                    }
                }
                ColumnGen::Literal(s) => {
                    values[i].push_str(s);
                }
                ColumnGen::Aggr { .. } => {}
                ColumnGen::Ref { source_col, modifier } => {
                    if let Some(src_idx) = col_names.iter().position(|n| n == source_col) {
                        raw_values[i] = raw_values[src_idx];
                        if !modifier.is_empty() {
                            if let Some(raw) = raw_values[src_idx] {
                                format_ref(
                                    raw,
                                    modifier,
                                    &gen_config.columns,
                                    src_idx,
                                    &mut values[i],
                                );
                                continue;
                            }
                        }
                        if src_idx < i {
                            let (left, right) = values.split_at_mut(i);
                            right[0].push_str(&left[src_idx]);
                        } else {
                            let (left, right) = values.split_at_mut(src_idx);
                            left[i].push_str(&right[0]);
                        }
                    }
                }
                ColumnGen::Expr { left, op, right, result_type } => {
                    let env = ExprEnv {
                        raw_values: &raw_values,
                        col_names: &col_names,
                        domain_hashes: &domain_hashes,
                        serial,
                    };
                    let lv = eval_operand(left, &env, &mut ctx, i, true);
                    let rv = eval_operand(right, &env, &mut ctx, i, false);
                    let adjusted_rv = match result_type {
                        crate::tpl::ExprResultType::Date => rv * 86400.0,
                        _ => rv,
                    };
                    let result = match op {
                        crate::tpl::ExprOp::Add => lv + adjusted_rv,
                        crate::tpl::ExprOp::Sub => lv - adjusted_rv,
                        crate::tpl::ExprOp::Mul => lv * rv,
                    };
                    raw_values[i] = Some(result);
                    format_raw_typed(result, *result_type, &mut values[i]);
                }

                ColumnGen::Fk {
                    parent_field,
                    parent_modifier,
                    parent_range,
                    parent_count,
                    parent_domain_hash,
                    distribution,
                    parent_ctx,
                    ..
                } => {
                    let mut sample_rng = Rng::derive_fast(domain_hashes[i], serial);
                    let row_idx = match distribution {
                        FkDistribution::Uniform => {
                            sample_rng.range(0, *parent_count as i64 - 1) as u64
                        }
                        FkDistribution::Zipf(s) => sample_rng.zipf(*parent_count, *s) - 1,
                    };
                    fk_row_indices[i] = Some(row_idx);
                    raw_values[i] = generate_parent_field(
                        parent_field,
                        parent_modifier,
                        parent_range,
                        *parent_domain_hash,
                        row_idx,
                        parent_ctx,
                        &mut values[i],
                    );
                }

                ColumnGen::FkDeref {
                    anchor_col,
                    deref_field,
                    deref_modifier,
                    deref_range,
                    deref_domain_hash,
                    parent_ctx,
                    ..
                } => {
                    let Some(anchor_idx) = col_names.iter().position(|n| n == anchor_col) else {
                        continue;
                    };
                    if let Some(row_idx) = fk_row_indices[anchor_idx] {
                        raw_values[i] = generate_parent_field(
                            deref_field,
                            deref_modifier,
                            deref_range,
                            *deref_domain_hash,
                            row_idx,
                            parent_ctx,
                            &mut values[i],
                        );
                    }
                }
            }
        }

        aggr.update(&mut values, &raw_values)?;

        if compiled_tpl.is_some() {
            let idx = col_count;
            values[idx].clear();
            values[idx].push_str(itoa_buf.format(serial));
        }

        let ok = if let Some(ref tpl) = compiled_tpl {
            let mut tpl_rng = Rng::derive(opts.master_seed, serial, seedfaker_core::DOMAIN_TPL);
            let mut rctx = crate::tpl::RenderCtx {
                values: &values,
                rng: &mut tpl_rng,
                locales: record_locales,
                identity: id_ref,
                tz_offset_minutes: opts.tz_offset_minutes,
                since: opts.since,
                until: opts.until,
                field_types: &field_types,
            };
            let mut segments = crate::tpl::render::collect(tpl, &mut rctx);

            let tpl_originals: Option<Vec<String>> = if opts.corrupt == Corrupt::None {
                None
            } else {
                let mut gen_values: Vec<String> = segments
                    .iter()
                    .filter_map(|s| match s {
                        crate::tpl::render::Segment::Value { value, .. } => Some(value.clone()),
                        crate::tpl::render::Segment::Lit(_) => None,
                    })
                    .collect();
                let originals = gen_values.clone();
                let mut cr = Rng::derive(opts.master_seed, serial, seedfaker_core::DOMAIN_CORRUPT);
                seedfaker_core::corrupt::corrupt_values(
                    &mut cr,
                    &mut gen_values,
                    opts.corrupt.rate(),
                );
                let mut vi = 0;
                for seg in &mut segments {
                    if let crate::tpl::render::Segment::Value { value, .. } = seg {
                        *value = std::mem::take(&mut gen_values[vi]);
                        vi += 1;
                    }
                }
                Some(originals)
            };

            if opts.annotated {
                crate::meta::write_annotated_line(&mut out, &segments, tpl_originals.as_deref())?;
                true
            } else {
                tpl_buf.clear();
                crate::tpl::render::assemble_text(&segments, &mut tpl_buf);
                out.write_all(tpl_buf.as_bytes()).is_ok() && out.write_all(b"\n").is_ok()
            }
        } else {
            let struct_originals: Option<Vec<String>> =
                if opts.annotated && opts.corrupt != Corrupt::None {
                    Some(values[..col_count].to_vec())
                } else {
                    None
                };

            if opts.corrupt != Corrupt::None {
                let mut cr = Rng::derive(opts.master_seed, serial, seedfaker_core::DOMAIN_CORRUPT);
                seedfaker_core::corrupt::corrupt_values(&mut cr, &mut values, opts.corrupt.rate());
            }

            if opts.annotated {
                let delim_str = match opts.output {
                    OutputMode::Csv => csv_sep_s.as_str(),
                    _ => tsv_sep_s.as_str(),
                };
                let segments = crate::tpl::render::structured_segments(
                    &col_names,
                    &values,
                    &is_omitted,
                    &field_types,
                    &opts.output,
                    delim_str,
                    sql_prefix.as_deref(),
                );
                crate::meta::write_annotated_line(
                    &mut out,
                    &segments,
                    struct_originals.as_deref(),
                )?;
                true
            } else {
                match opts.output {
                    OutputMode::Default => writers::write_lines(&mut out, &values, tsv_sep),
                    OutputMode::Csv => writers::write_csv(&mut out, &values, csv_sep),
                    OutputMode::Tsv => writers::write_tsv(&mut out, &values, tsv_sep),
                    OutputMode::Jsonl => {
                        writers::write_jsonl(&mut out, &col_names, &values, &is_omitted)
                    }
                    OutputMode::Sql(_) => {
                        if let Some(ref prefix) = sql_prefix {
                            writers::write_sql(&mut out, prefix, &values, &is_omitted)
                        } else {
                            true
                        }
                    }
                    OutputMode::Template => {
                        for v in &values {
                            if out.write_all(v.as_bytes()).is_err() || out.write_all(b"\n").is_err()
                            {
                                break;
                            }
                        }
                        true
                    }
                }
            }
        };

        if !ok {
            break;
        }

        serial += 1;

        if let Some(interval) = rate_interval {
            if out.flush().is_err() {
                return Ok(());
            }
            std::thread::sleep(interval);
        }
    }

    if out.flush().is_err() {
        return Ok(());
    }
    Ok(())
}

pub use seedfaker_core::eval::compute_domain_hashes;

fn resolve_range(
    range: &Option<RangeSpec>,
    field_name: &str,
    opts: &RunOptions,
) -> Option<(i64, i64)> {
    seedfaker_core::field::resolve_range(range, field_name, opts.since, opts.until)
}

pub use seedfaker_core::eval::eval_operand;
pub use seedfaker_core::eval::format_raw_typed;
pub use seedfaker_core::eval::format_ref;
pub use seedfaker_core::eval::ExprEnv;

fn validate_resolved_ranges(
    columns: &[crate::tpl::column::Column],
    ranges: &[Option<(i64, i64)>],
) -> Result<(), Box<dyn std::error::Error>> {
    for (v, range) in columns.iter().zip(ranges.iter()) {
        if let Some((from, to)) = range {
            if from >= to {
                let name = match &v.gen {
                    ColumnGen::Field { field, .. } => field.name,
                    _ => "?",
                };
                return Err(format!("invalid resolved range for '{name}': {from}..{to}").into());
            }
        }
    }
    Ok(())
}
