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

use crate::tpl::column::ColumnGen;
use seedfaker_core::ctx::Identity;
use seedfaker_core::field::{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 ground_truth: bool,
    pub tz_offset_minutes: i32,
    pub since: i64,
    pub until: i64,
}

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

pub fn run(gen_config: &GenConfig, opts: &RunOptions) -> Result<(), Box<dyn std::error::Error>> {
    let stdout = std::io::stdout();
    let mut out = BufWriter::with_capacity(256 * 1024, stdout.lock());
    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);

    // Pre-compute native locales
    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 };

    // Pre-resolve per-var ranges
    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 { .. } => None,
        })
        .collect();

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

    // Find birthdate range for Identity
    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,
        });

    // Template compilation — "serial" is an implicit var, always available
    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();

    // CSV/TSV separators
    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();

    // SQL prefix
    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
    };

    // Header
    if !opts.no_header && !is_text && !opts.ground_truth {
        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 tpl_buf = String::new();

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

        // Per-record script selection (Script::Both = random per record)
        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
        };

        // Per-record locale locking for ctx (stack-only, no heap alloc)
        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
        };

        // Identity for ctx
        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();

        // Generate all columns (reuse pre-allocated buffers, hoist constant ctx fields)
        let mut ctx = seedfaker_core::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: seedfaker_core::field::Ordering::None,
            numeric: None,
        };
        for i in 0..col_count {
            values[i].clear();
            raw_values[i] = None;
        }
        for &i in &gen_config.eval_order {
            match &gen_config.columns[i].gen {
                ColumnGen::Field { field, modifier, transform, ordering, .. } => {
                    ctx.rng = Rng::derive_fast(domain_hashes[i], serial);
                    ctx.modifier = modifier;
                    ctx.range = resolved_ranges[i];
                    ctx.ordering = *ordering;
                    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() {
                            let src = values[src_idx].clone();
                            values[i].push_str(&src);
                        } else if let Some(raw) = raw_values[src_idx] {
                            format_ref(raw, modifier, &gen_config.columns, src_idx, &mut values[i]);
                        } else {
                            let src = values[src_idx].clone();
                            values[i].push_str(&src);
                        }
                    }
                }
                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);
                    // Date: delta is days → convert to seconds
                    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, // mul never adjusts
                    };
                    raw_values[i] = Some(result);
                    format_raw_typed(result, *result_type, &mut values[i]);
                }
            }
        }

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

        // Serial var (always last, for template mode)
        if compiled_tpl.is_some() {
            let idx = col_count;
            values[idx].clear();
            let mut tmp = itoa::Buffer::new();
            values[idx].push_str(tmp.format(serial));
        }

        // Snapshot originals before corruption (if ground_truth mode)
        let originals: Option<Vec<String>> =
            if opts.ground_truth { Some(values.clone()) } 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());
        }

        // Ground truth mode: JSONL with value + original (when corrupted)
        if let Some(ref orig) = originals {
            write_ground_truth(&mut out, serial, &col_names, &values, orig)?;
            serial += 1;
            if let Some(interval) = rate_interval {
                if out.flush().is_err() {
                    return Ok(());
                }
                std::thread::sleep(interval);
            }
            continue;
        }

        // Output
        let ok = if let Some(ref tpl) = compiled_tpl {
            tpl_buf.clear();
            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,
            };
            crate::tpl::render::render(tpl, &mut rctx, &mut tpl_buf);
            out.write_all(tpl_buf.as_bytes()).is_ok() && out.write_all(b"\n").is_ok()
        } 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),
                OutputMode::Sql(_) => {
                    if let Some(ref prefix) = sql_prefix {
                        writers::write_sql(&mut out, prefix, &values)
                    } 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(())
}

fn write_ground_truth(
    out: &mut impl Write,
    serial: u64,
    names: &[String],
    values: &[String],
    originals: &[String],
) -> std::io::Result<()> {
    out.write_all(b"{\"serial\":")?;
    let mut tmp = itoa::Buffer::new();
    out.write_all(tmp.format(serial).as_bytes())?;
    for ((name, val), orig) in names.iter().zip(values).zip(originals) {
        out.write_all(b",\"")?;
        out.write_all(name.as_bytes())?;
        out.write_all(b"\":\"")?;
        writers::json_escape_write(out, val.as_bytes())?;
        out.write_all(b"\"")?;
        if val != orig {
            out.write_all(b",\"")?;
            out.write_all(name.as_bytes())?;
            out.write_all(b"__original\":\"")?;
            writers::json_escape_write(out, orig.as_bytes())?;
            out.write_all(b"\"")?;
        }
    }
    out.write_all(b"}\n")
}

pub fn compute_domain_hashes(columns: &[crate::tpl::column::Column], master_seed: u64) -> Vec<u64> {
    columns
        .iter()
        .map(|v| match &v.gen {
            ColumnGen::Field { field, modifier, .. } => {
                let base =
                    seedfaker_core::pipeline::field_domain_hash(master_seed, field, modifier);
                if is_alias(&v.name, field.name, modifier) {
                    seedfaker_core::rng::domain_hash(base, &v.name)
                } else {
                    base
                }
            }
            ColumnGen::Expr { .. } => {
                // Expr columns need a domain hash for inline field operands.
                seedfaker_core::rng::domain_hash(master_seed, &format!("_expr_{}", v.name))
            }
            ColumnGen::Literal(_) | ColumnGen::Aggr { .. } | ColumnGen::Ref { .. } => 0,
        })
        .collect()
}

/// Check if a column name is a user-provided alias (not auto-derived from field + modifier).
/// Auto-derived names: `first-name`, `first_name`, `phone_e164`, `phone:e164`.
fn is_alias(col_name: &str, field_name: &str, modifier: &str) -> bool {
    if col_name == field_name {
        return false;
    }
    let normalized = field_name.replace('-', "_");
    if col_name == normalized {
        return false;
    }
    if !modifier.is_empty()
        && (col_name == format!("{field_name}_{modifier}")
            || col_name == format!("{normalized}_{modifier}")
            || col_name == format!("{field_name}:{modifier}"))
    {
        return false;
    }
    true
}

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)
}

/// Shared context for expression evaluation within one record.
pub struct ExprEnv<'a> {
    pub raw_values: &'a [Option<f64>],
    pub col_names: &'a [String],
    pub domain_hashes: &'a [u64],
    pub serial: u64,
}

/// Evaluate one side of an expression. Column refs read from `raw_values`;
/// field specs generate via RNG derived from the expression's domain hash.
/// `is_left` selects sub-domain to keep left/right deterministically independent.
pub fn eval_operand<'a>(
    operand: &'a crate::tpl::ExprOperand,
    env: &ExprEnv<'_>,
    ctx: &mut seedfaker_core::ctx::GenContext<'a>,
    expr_idx: usize,
    is_left: bool,
) -> f64 {
    match operand {
        crate::tpl::ExprOperand::Col(name) => env
            .col_names
            .iter()
            .position(|n| n == name)
            .and_then(|idx| env.raw_values[idx])
            .unwrap_or(0.0),
        crate::tpl::ExprOperand::Field { field, modifier, range } => {
            let sub = if is_left { "L" } else { "R" };
            let sub_hash = seedfaker_core::rng::domain_hash(env.domain_hashes[expr_idx], sub);
            ctx.rng = Rng::derive_fast(sub_hash, env.serial);
            ctx.modifier = modifier;
            ctx.range = range.as_ref().and_then(|r| {
                seedfaker_core::field::resolve_range(&Some(*r), field.name, ctx.since, ctx.until)
            });
            let mut buf = String::new();
            let raw = field.generate(ctx, &mut buf);
            raw.unwrap_or(0.0)
        }
    }
}

/// Format a raw f64 according to expression result type.
pub fn format_raw_typed(value: f64, result_type: crate::tpl::ExprResultType, buf: &mut String) {
    use crate::tpl::ExprResultType;
    match result_type {
        ExprResultType::Date => {
            let epoch = value as i64;
            let (y, m, d, _, _, _) = seedfaker_core::gen::timestamp::epoch_to_parts(epoch);
            use std::fmt::Write;
            let _ = write!(buf, "{y:04}-{m:02}-{d:02}");
        }
        ExprResultType::Timestamp => {
            let epoch = value as i64;
            let (y, m, d, h, min, s) = seedfaker_core::gen::timestamp::epoch_to_parts(epoch);
            use std::fmt::Write;
            let _ = write!(buf, "{y:04}-{m:02}-{d:02}T{h:02}:{min:02}:{s:02}Z");
        }
        ExprResultType::Money => {
            use std::fmt::Write;
            let _ = write!(buf, "{value:.2}");
        }
        ExprResultType::Float => {
            use std::fmt::Write;
            let _ = write!(buf, "{value:.2}");
        }
        ExprResultType::Int => {
            let mut tmp = itoa::Buffer::new();
            buf.push_str(tmp.format(value as i64));
        }
    }
}

/// Reformat a raw value with a modifier. Used for `col:modifier` references.
pub fn format_ref(
    raw: f64,
    modifier: &str,
    columns: &[crate::tpl::column::Column],
    src_idx: usize,
    buf: &mut String,
) {
    use crate::tpl::column::{field_type, ColumnGen, FieldType};

    // Determine source field type for formatting context.
    let ft = match &columns[src_idx].gen {
        ColumnGen::Field { field, .. } => field_type(field.name),
        ColumnGen::Expr { result_type, .. } => match result_type {
            crate::tpl::ExprResultType::Date => FieldType::Date,
            crate::tpl::ExprResultType::Timestamp => FieldType::Timestamp,
            crate::tpl::ExprResultType::Money => FieldType::Money,
            crate::tpl::ExprResultType::Float => FieldType::Float,
            crate::tpl::ExprResultType::Int => FieldType::Int,
        },
        _ => FieldType::Int,
    };

    match ft {
        FieldType::Date => {
            let epoch = raw as i64;
            let (y, m, d, _, _, _) = seedfaker_core::gen::timestamp::epoch_to_parts(epoch);
            match modifier {
                "us" => {
                    seedfaker_core::gen::date::push_pad2(buf, m);
                    buf.push('/');
                    seedfaker_core::gen::date::push_pad2(buf, d);
                    buf.push('/');
                    buf.push_str(itoa::Buffer::new().format(y));
                }
                "eu" => {
                    seedfaker_core::gen::date::push_pad2(buf, d);
                    buf.push('.');
                    seedfaker_core::gen::date::push_pad2(buf, m);
                    buf.push('.');
                    buf.push_str(itoa::Buffer::new().format(y));
                }
                _ => format_raw_typed(raw, crate::tpl::ExprResultType::Date, buf),
            }
        }
        FieldType::Timestamp => {
            let epoch = raw as i64;
            match modifier {
                "unix" => buf.push_str(itoa::Buffer::new().format(epoch)),
                "ms" => buf.push_str(itoa::Buffer::new().format(epoch * 1000)),
                _ => format_raw_typed(raw, crate::tpl::ExprResultType::Timestamp, buf),
            }
        }
        FieldType::Money => {
            let v = raw;
            match modifier {
                "plain" => {
                    use std::fmt::Write;
                    let _ = write!(buf, "{v:.2}");
                }
                "usd" => {
                    buf.push('$');
                    push_money_formatted(v, buf, ',', '.');
                }
                "eur" => {
                    buf.push('€');
                    push_money_formatted(v, buf, '.', ',');
                }
                "gbp" => {
                    buf.push('£');
                    push_money_formatted(v, buf, ',', '.');
                }
                _ => {
                    use std::fmt::Write;
                    let _ = write!(buf, "{v:.2}");
                }
            }
        }
        _ => format_raw_typed(raw, crate::tpl::ExprResultType::Int, buf),
    }
}

fn push_money_formatted(v: f64, buf: &mut String, thousands: char, decimal: char) {
    let abs = v.abs();
    let whole = abs as i64;
    let cents = ((abs - whole as f64) * 100.0).round() as i64;
    if v < 0.0 {
        buf.push('-');
    }
    let mut ib = itoa::Buffer::new();
    let s = ib.format(whole);
    let len = s.len();
    for (i, ch) in s.chars().enumerate() {
        if i > 0 && (len - i) % 3 == 0 {
            buf.push(thousands);
        }
        buf.push(ch);
    }
    buf.push(decimal);
    if cents < 10 {
        buf.push('0');
    }
    buf.push_str(itoa::Buffer::new().format(cents));
}

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,
                    ColumnGen::Literal(_)
                    | ColumnGen::Aggr { .. }
                    | ColumnGen::Ref { .. }
                    | ColumnGen::Expr { .. } => "?",
                };
                return Err(format!("invalid resolved range for '{name}': {from}..{to}").into());
            }
        }
    }
    Ok(())
}
