use std::{
    collections::BTreeMap,
    fs,
    io::{self, Stdout},
    path::{Path, PathBuf},
    sync::mpsc,
};
use serde::{Serialize, Deserialize};

use anyhow::Result;
use crossterm::{
    event::{self, Event, KeyCode, KeyEvent, KeyModifiers},
    execute,
    terminal::{EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode},
};
use ratatui::{
    Frame, Terminal,
    backend::CrosstermBackend,
    layout::{Constraint, Direction, Layout, Rect},
    style::{Color, Modifier, Style},
    text::{Line, Span},
    widgets::{Block, Borders, Cell, Clear, Paragraph, Row, Table, TableState, Wrap},
};

use crate::{
    dataset::{DatasetPreview, FileFormat},
    profile::{ColumnProfile, DatasetProfile, profile_path_silent, profile_path_streaming},
};

type Tui = Terminal<CrosstermBackend<Stdout>>;
const PREFETCH_THRESHOLD: usize = 250;
const CELL_WIDTH: u16 = 24;
const ROW_GUTTER_WIDTH: u16 = 8;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Mode {
    Normal,
    Search,
    Help,
    RowDetail,
    ColumnJump,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum SearchScope {
    All,
    Column,
}

struct ProfileReceiver(mpsc::Receiver<Result<DatasetProfile, String>>);

impl std::fmt::Debug for ProfileReceiver {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Receiver")
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ThemeKind {
    RosePineMoon,
    Nord,
    Gruvbox,
    Dracula,
}

impl ThemeKind {
    fn name(&self) -> &'static str {
        match self {
            ThemeKind::RosePineMoon => "Rosé Pine Moon",
            ThemeKind::Nord => "Nord",
            ThemeKind::Gruvbox => "Gruvbox",
            ThemeKind::Dracula => "Dracula",
        }
    }

    fn cycle(&self) -> Self {
        match self {
            ThemeKind::RosePineMoon => ThemeKind::Nord,
            ThemeKind::Nord => ThemeKind::Gruvbox,
            ThemeKind::Gruvbox => ThemeKind::Dracula,
            ThemeKind::Dracula => ThemeKind::RosePineMoon,
        }
    }

    fn base_bg(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(35, 33, 54),
            ThemeKind::Nord => Color::Rgb(46, 52, 64),
            ThemeKind::Gruvbox => Color::Rgb(40, 40, 40),
            ThemeKind::Dracula => Color::Rgb(40, 42, 54),
        }
    }

    fn alternate_bg(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(42, 39, 63),
            ThemeKind::Nord => Color::Rgb(53, 60, 74),
            ThemeKind::Gruvbox => Color::Rgb(50, 48, 47),
            ThemeKind::Dracula => Color::Rgb(48, 50, 64),
        }
    }

    fn surface_bg(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(57, 53, 82),
            ThemeKind::Nord => Color::Rgb(59, 66, 82),
            ThemeKind::Gruvbox => Color::Rgb(60, 56, 54),
            ThemeKind::Dracula => Color::Rgb(68, 71, 90),
        }
    }

    fn text(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(224, 222, 244),
            ThemeKind::Nord => Color::Rgb(216, 222, 233),
            ThemeKind::Gruvbox => Color::Rgb(235, 219, 178),
            ThemeKind::Dracula => Color::Rgb(248, 248, 242),
        }
    }

    fn muted(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(110, 106, 134),
            ThemeKind::Nord => Color::Rgb(76, 86, 106),
            ThemeKind::Gruvbox => Color::Rgb(146, 131, 116),
            ThemeKind::Dracula => Color::Rgb(98, 114, 164),
        }
    }

    fn accent(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(156, 207, 216),
            ThemeKind::Nord => Color::Rgb(136, 192, 208),
            ThemeKind::Gruvbox => Color::Rgb(131, 165, 152),
            ThemeKind::Dracula => Color::Rgb(139, 233, 253),
        }
    }

    fn primary(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(246, 193, 119),
            ThemeKind::Nord => Color::Rgb(235, 203, 139),
            ThemeKind::Gruvbox => Color::Rgb(250, 189, 47),
            ThemeKind::Dracula => Color::Rgb(241, 250, 140),
        }
    }

    fn success(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(62, 143, 176),
            ThemeKind::Nord => Color::Rgb(163, 190, 140),
            ThemeKind::Gruvbox => Color::Rgb(184, 187, 38),
            ThemeKind::Dracula => Color::Rgb(80, 250, 123),
        }
    }

    fn error(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(235, 111, 146),
            ThemeKind::Nord => Color::Rgb(191, 97, 106),
            ThemeKind::Gruvbox => Color::Rgb(251, 73, 52),
            ThemeKind::Dracula => Color::Rgb(255, 85, 85),
        }
    }

    #[allow(dead_code)]
    fn warning(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(234, 154, 151),
            ThemeKind::Nord => Color::Rgb(208, 135, 112),
            ThemeKind::Gruvbox => Color::Rgb(254, 128, 25),
            ThemeKind::Dracula => Color::Rgb(255, 184, 108),
        }
    }

    fn iris(&self) -> Color {
        match self {
            ThemeKind::RosePineMoon => Color::Rgb(196, 167, 231),
            ThemeKind::Nord => Color::Rgb(180, 142, 173),
            ThemeKind::Gruvbox => Color::Rgb(211, 134, 155),
            ThemeKind::Dracula => Color::Rgb(189, 147, 249),
        }
    }
}

#[derive(Debug)]
struct App {
    dataset: DatasetPreview,
    selected_row: usize,
    selected_col: usize,
    row_offset: usize,
    col_offset: usize,
    search: String,
    search_scope: SearchScope,
    column_query: String,
    last_message: String,
    detail_col: usize,
    detail_offset: usize,
    column_browser_selected: usize,
    column_browser_offset: usize,
    profile_cache: BTreeMap<usize, ProfileCache>,
    full_profile: Option<DatasetProfile>,
    full_profile_error: Option<String>,
    mode: Mode,
    profile_rx: Option<ProfileReceiver>,
    theme: ThemeKind,
}

#[derive(Debug, Clone)]
struct ProfileCache {
    cache_version: u64,
    profile: ColumnProfile,
}

const DAP_CACHE_VERSION: u32 = 2;

#[derive(Debug, Serialize, Deserialize)]
struct DiskCacheEntry {
    #[serde(default)]
    dap_cache_version: u32,
    file_size: u64,
    modified_sec: u64,
    modified_nsec: u32,
    profile: DatasetProfile,
}

fn get_file_metadata(path: &Path) -> Option<(u64, u64, u32)> {
    let metadata = fs::metadata(path).ok()?;
    let size = metadata.len();
    let modified = metadata.modified().ok()?;
    let duration = modified.duration_since(std::time::SystemTime::UNIX_EPOCH).ok()?;
    Some((size, duration.as_secs(), duration.subsec_nanos()))
}

fn get_cache_path(path: &Path) -> Option<PathBuf> {
    let canonical = fs::canonicalize(path).ok().unwrap_or_else(|| path.to_path_buf());
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    use std::hash::{Hash, Hasher};
    canonical.hash(&mut hasher);
    let hash_val = hasher.finish();

    let base = std::env::var_os("XDG_CACHE_HOME")
        .map(PathBuf::from)
        .or_else(|| {
            std::env::var_os("HOME")
                .map(|home| PathBuf::from(home).join(".cache"))
        })?;
    let cache_dir = base.join("dap");
    let _ = fs::create_dir_all(&cache_dir);
    Some(cache_dir.join(format!("{:016x}.json", hash_val)))
}

fn load_cached_profile(path: &Path) -> Option<DatasetProfile> {
    let cache_path = get_cache_path(path)?;
    let content = fs::read_to_string(cache_path).ok()?;
    let entry: DiskCacheEntry = serde_json::from_str(&content).ok()?;
    
    // Check cache version
    if entry.dap_cache_version < DAP_CACHE_VERSION {
        return None;
    }
    
    // Validate that the file size and mtime match
    let (size, sec, nsec) = get_file_metadata(path)?;
    if entry.file_size == size && entry.modified_sec == sec && entry.modified_nsec == nsec {
        Some(entry.profile)
    } else {
        None
    }
}

fn save_cached_profile(path: &Path, profile: &DatasetProfile) -> Option<()> {
    let cache_path = get_cache_path(path)?;
    let (size, sec, nsec) = get_file_metadata(path)?;
    let entry = DiskCacheEntry {
        dap_cache_version: DAP_CACHE_VERSION,
        file_size: size,
        modified_sec: sec,
        modified_nsec: nsec,
        profile: profile.clone(),
    };
    let content = serde_json::to_string(&entry).ok()?;
    fs::write(cache_path, content).ok()?;
    Some(())
}

impl App {
    fn new(dataset: DatasetPreview) -> Self {
        let path_str = dataset.path.clone();
        let (tx, rx) = mpsc::channel();
        std::thread::spawn(move || {
            let path = Path::new(&path_str);
            if let Some(profile) = load_cached_profile(path) {
                let _ = tx.send(Ok(profile));
                return;
            }
            let tx_clone = tx.clone();
            match profile_path_streaming(path, None, move |partial| {
                let _ = tx_clone.send(Ok(partial.clone()));
            }) {
                Ok(profile) => {
                    save_cached_profile(path, &profile);
                    let _ = tx.send(Ok(profile));
                }
                Err(err) => {
                    let _ = tx.send(Err(err.to_string()));
                }
            }
        });

        Self {
            dataset,
            selected_row: 0,
            selected_col: 0,
            row_offset: 0,
            col_offset: 0,
            search: String::new(),
            search_scope: SearchScope::All,
            column_query: String::new(),
            last_message: "loading profile in background...".to_string(),
            detail_col: 0,
            detail_offset: 0,
            column_browser_selected: 0,
            column_browser_offset: 0,
            profile_cache: BTreeMap::new(),
            full_profile: None,
            full_profile_error: None,
            mode: Mode::Normal,
            profile_rx: Some(ProfileReceiver(rx)),
            theme: ThemeKind::Gruvbox,
        }
    }

    fn move_row(&mut self, delta: isize) {
        if delta.is_positive() {
            let target = self.selected_row.saturating_add(delta as usize);
            self.ensure_rows_for(target);
        }

        let max = self.dataset.row_count().saturating_sub(1);
        self.selected_row = add_delta(self.selected_row, delta).min(max);
        self.sync_offsets();
        self.prefetch_if_needed();
    }

    fn move_col(&mut self, delta: isize) {
        let max = self.dataset.column_count().saturating_sub(1);
        self.selected_col = add_delta(self.selected_col, delta).min(max);
        self.sync_offsets();
    }

    fn page_col(&mut self, delta: isize) {
        self.move_col(delta * 5);
    }

    fn move_detail_field(&mut self, delta: isize) {
        let max = self.dataset.column_count().saturating_sub(1);
        self.detail_col = add_delta(self.detail_col, delta).min(max);
    }

    fn page_detail_field(&mut self, delta: isize) {
        self.move_detail_field(delta * 10);
    }

    fn page(&mut self, delta: isize) {
        self.move_row(delta * 20);
    }

    fn half_page(&mut self, delta: isize) {
        self.move_row(delta * 10);
    }

    fn go_top(&mut self) {
        self.selected_row = 0;
        self.sync_offsets();
    }

    fn go_bottom_buffered(&mut self) {
        self.selected_row = self.dataset.row_count().saturating_sub(1);
        self.sync_offsets();
        self.prefetch_if_needed();
    }

    fn load_more_now(&mut self) {
        match self.dataset.load_more() {
            Ok(0) => self.last_message = "end of file".to_string(),
            Ok(loaded) => self.last_message = format!("loaded {loaded} more rows"),
            Err(err) => self.last_message = format!("load failed: {err}"),
        }
    }

    fn toggle_search_scope(&mut self) {
        self.search_scope = match self.search_scope {
            SearchScope::All => SearchScope::Column,
            SearchScope::Column => SearchScope::All,
        };
        self.last_message = format!("search scope: {}", self.search_scope.label());
    }

    fn open_column_browser(&mut self) {
        self.column_query.clear();
        self.column_browser_selected = self.selected_col;
        self.column_browser_offset = self.selected_col.saturating_sub(5);
        self.ensure_full_profile();
        self.mode = Mode::ColumnJump;
    }

    fn ensure_full_profile(&mut self) {
        if self.full_profile.is_some() || self.full_profile_error.is_some() {
            return;
        }

        if self.profile_rx.is_some() {
            return;
        }

        let path = Path::new(&self.dataset.path);
        if let Some(profile) = load_cached_profile(path) {
            self.last_message = format!(
                "loaded cached profile ({} rows)",
                profile.rows_profiled
            );
            self.full_profile = Some(profile);
            return;
        }

        match profile_path_silent(path, None) {
            Ok(profile) => {
                self.last_message =
                    format!("profiled {} rows for column browser", profile.rows_profiled);
                save_cached_profile(path, &profile);
                self.full_profile = Some(profile);
            }
            Err(err) => {
                let message = format!("full profile failed: {err}");
                self.last_message = message.clone();
                self.full_profile_error = Some(message);
            }
        }
    }

    fn check_background_profile(&mut self) {
        let Some(rx) = &self.profile_rx else {
            return;
        };

        let mut latest = None;
        let mut error = None;
        let mut finished = false;

        loop {
            match rx.0.try_recv() {
                Ok(Ok(profile)) => {
                    latest = Some(profile);
                }
                Ok(Err(err)) => {
                    error = Some(err);
                    finished = true;
                    break;
                }
                Err(mpsc::TryRecvError::Empty) => {
                    break;
                }
                Err(mpsc::TryRecvError::Disconnected) => {
                    finished = true;
                    break;
                }
            }
        }

        if let Some(profile) = latest {
            self.last_message = format!(
                "loaded/updated profile ({} rows)",
                profile.rows_profiled
            );
            self.full_profile = Some(profile);
        }

        if let Some(err) = error {
            self.last_message = format!("profile failed: {}", err);
            self.full_profile_error = Some(err);
        }

        if finished {
            self.profile_rx = None;
        }
    }

    fn full_column_profile(&self, col: usize) -> Option<&ColumnProfile> {
        self.full_profile.as_ref()?.columns.get(col)
    }

    fn normalize_column_browser_selection(&mut self) {
        let matches = filtered_columns(&self.dataset.headers, &self.column_query);
        if matches.is_empty() {
            self.column_browser_selected = 0;
            self.column_browser_offset = 0;
            return;
        }

        if !matches.contains(&self.column_browser_selected) {
            self.column_browser_selected = matches[0];
        }
    }

    fn move_column_browser(&mut self, delta: isize) {
        let matches = filtered_columns(&self.dataset.headers, &self.column_query);
        if matches.is_empty() {
            return;
        }

        let current = matches
            .iter()
            .position(|col| *col == self.column_browser_selected)
            .unwrap_or(0);
        let next = add_delta(current, delta).min(matches.len().saturating_sub(1));
        self.column_browser_selected = matches[next];
    }

    fn accept_column_browser(&mut self) {
        if self
            .dataset
            .headers
            .get(self.column_browser_selected)
            .is_none()
        {
            self.last_message = "no matching columns".to_string();
            return;
        }

        self.selected_col = self.column_browser_selected;
        self.sync_offsets();
        self.last_message = format!(
            "column {}: {}",
            self.selected_col + 1,
            self.dataset.headers[self.selected_col]
        );
    }

    fn ensure_rows_for(&mut self, row: usize) {
        while row >= self.dataset.row_count() && !self.dataset.exhausted {
            match self.dataset.load_more() {
                Ok(0) => {
                    self.last_message = "end of file".to_string();
                    break;
                }
                Ok(loaded) => {
                    self.last_message = format!("loaded {loaded} more rows");
                }
                Err(err) => {
                    self.last_message = format!("load failed: {err}");
                    break;
                }
            }
        }
    }

    fn prefetch_if_needed(&mut self) {
        if self.dataset.exhausted || self.dataset.row_count() == 0 {
            return;
        }

        let remaining = self
            .dataset
            .row_count()
            .saturating_sub(self.selected_row + 1);
        if remaining > PREFETCH_THRESHOLD {
            return;
        }

        match self.dataset.load_more() {
            Ok(0) => self.last_message = "end of file".to_string(),
            Ok(loaded) => self.last_message = format!("prefetched {loaded} more rows"),
            Err(err) => self.last_message = format!("prefetch failed: {err}"),
        }
    }

    fn sync_offsets(&mut self) {
        if self.selected_row < self.row_offset {
            self.row_offset = self.selected_row;
        } else if self.selected_row >= self.row_offset + 20 {
            self.row_offset = self.selected_row.saturating_sub(19);
        }

        if self.selected_col < self.col_offset {
            self.col_offset = self.selected_col;
        } else if self.selected_col >= self.col_offset + 5 {
            self.col_offset = self.selected_col.saturating_sub(4);
        }
    }

    fn next_search_match(&mut self) {
        if self.search.is_empty() {
            self.last_message = "search empty".to_string();
            return;
        }

        let needle = self.search.to_ascii_lowercase();
        let cols = self.dataset.column_count();
        if self.dataset.row_count() == 0 || cols == 0 {
            self.last_message = "no rows".to_string();
            return;
        }

        match self.search_scope {
            SearchScope::All => self.next_search_match_all(&needle, cols),
            SearchScope::Column => self.next_search_match_column(&needle),
        }
    }

    fn next_search_match_all(&mut self, needle: &str, cols: usize) {
        let start_row = self.selected_row;
        let start_col = self.selected_col;
        let mut row = self.selected_row;
        let mut col = self.selected_col + 1;

        loop {
            if col >= cols {
                col = 0;
                row += 1;
            }

            if row >= self.dataset.row_count() {
                if !self.dataset.exhausted {
                    match self.dataset.load_more() {
                        Ok(0) => {}
                        Ok(loaded) => {
                            self.last_message =
                                format!("loaded {loaded} more rows while searching");
                            continue;
                        }
                        Err(err) => {
                            self.last_message = format!("search load failed: {err}");
                            return;
                        }
                    }
                }

                row = 0;
            }

            if row == start_row && col == start_col {
                self.last_message = "no match".to_string();
                return;
            }

            if let Err(err) = self.dataset.ensure_row_loaded(row) {
                self.last_message = format!("search load failed: {err}");
                return;
            }

            if self
                .dataset
                .value(row, col)
                .contains_ignore_ascii_case(needle)
            {
                self.selected_row = row;
                self.selected_col = col;
                self.last_message = format!("match at row {}, col {}", row + 1, col + 1);
                self.sync_offsets();
                self.prefetch_if_needed();
                return;
            }

            col += 1;
        }
    }

    fn next_search_match_column(&mut self, needle: &str) {
        let start_row = self.selected_row;
        let col = self.selected_col;
        let mut row = self.selected_row + 1;

        loop {
            if row >= self.dataset.row_count() {
                if !self.dataset.exhausted {
                    match self.dataset.load_more() {
                        Ok(0) => {}
                        Ok(loaded) => {
                            self.last_message =
                                format!("loaded {loaded} more rows while searching");
                            continue;
                        }
                        Err(err) => {
                            self.last_message = format!("search load failed: {err}");
                            return;
                        }
                    }
                }

                row = 0;
            }

            if row == start_row {
                self.last_message = "no match".to_string();
                return;
            }

            if let Err(err) = self.dataset.ensure_row_loaded(row) {
                self.last_message = format!("search load failed: {err}");
                return;
            }

            if self
                .dataset
                .value(row, col)
                .contains_ignore_ascii_case(needle)
            {
                self.selected_row = row;
                self.last_message = format!("match at row {}, col {}", row + 1, col + 1);
                self.sync_offsets();
                self.prefetch_if_needed();
                return;
            }

            row += 1;
        }
    }

    fn selected_column_profile(&mut self) -> Option<ColumnProfile> {
        self.column_profile_cached(self.selected_col)
    }

    fn column_profile_cached(&mut self, col: usize) -> Option<ColumnProfile> {
        let cache_version = self.dataset.cache_version();
        self.profile_cache
            .retain(|_, cache| cache.cache_version == cache_version);

        if let Some(cache) = self.profile_cache.get(&col) {
            return Some(cache.profile.clone());
        }

        let profile = self.dataset.column_profile(col)?;
        self.profile_cache.insert(
            col,
            ProfileCache {
                cache_version,
                profile: profile.clone(),
            },
        );
        Some(profile)
    }
}

impl SearchScope {
    fn label(self) -> &'static str {
        match self {
            SearchScope::All => "all columns",
            SearchScope::Column => "selected column",
        }
    }
}

pub fn run(dataset: DatasetPreview) -> Result<()> {
    enable_raw_mode()?;
    let mut stdout = io::stdout();
    execute!(stdout, EnterAlternateScreen)?;
    let backend = CrosstermBackend::new(stdout);
    let mut terminal = Terminal::new(backend)?;

    let result = run_app(&mut terminal, App::new(dataset));

    disable_raw_mode()?;
    execute!(terminal.backend_mut(), LeaveAlternateScreen)?;
    terminal.show_cursor()?;

    result
}

fn run_app(terminal: &mut Tui, mut app: App) -> Result<()> {
    use std::time::Duration;
    loop {
        app.check_background_profile();
        terminal.draw(|frame| draw(frame, &mut app))?;

        if event::poll(Duration::from_millis(100))? {
            if let Event::Key(key) = event::read()?
                && handle_key(&mut app, key)
            {
                break;
            }
        }
    }

    Ok(())
}

fn handle_key(app: &mut App, key: KeyEvent) -> bool {
    match app.mode {
        Mode::Help => match key.code {
            KeyCode::Esc | KeyCode::Char('?') | KeyCode::Char('q') => app.mode = Mode::Normal,
            _ => {}
        },
        Mode::RowDetail => match key.code {
            KeyCode::Esc | KeyCode::Enter | KeyCode::Char('q') => app.mode = Mode::Normal,
            KeyCode::Down | KeyCode::Char('j') => {
                app.move_detail_field(1);
            }
            KeyCode::Up | KeyCode::Char('k') => {
                app.move_detail_field(-1);
            }
            KeyCode::PageDown => {
                app.page_detail_field(1);
            }
            KeyCode::PageUp => {
                app.page_detail_field(-1);
            }
            _ => {}
        },
        Mode::Search => match key.code {
            KeyCode::Esc => app.mode = Mode::Normal,
            KeyCode::Tab => app.toggle_search_scope(),
            KeyCode::Enter => {
                app.mode = Mode::Normal;
                app.next_search_match();
            }
            KeyCode::Backspace => {
                app.search.pop();
            }
            KeyCode::Char(c) => app.search.push(c),
            _ => {}
        },
        Mode::ColumnJump => match key.code {
            KeyCode::Esc => app.mode = Mode::Normal,
            KeyCode::Enter => {
                app.mode = Mode::Normal;
                app.accept_column_browser();
            }
            KeyCode::Down => app.move_column_browser(1),
            KeyCode::Up => app.move_column_browser(-1),
            KeyCode::Char('n') if key.modifiers.contains(KeyModifiers::CONTROL) => app.move_column_browser(1),
            KeyCode::Char('p') if key.modifiers.contains(KeyModifiers::CONTROL) => app.move_column_browser(-1),
            KeyCode::PageDown => app.move_column_browser(10),
            KeyCode::PageUp => app.move_column_browser(-10),
            KeyCode::Home => app.move_column_browser(-(app.dataset.column_count() as isize)),
            KeyCode::End => app.move_column_browser(app.dataset.column_count() as isize),
            KeyCode::Backspace => {
                app.column_query.pop();
                app.normalize_column_browser_selection();
            }
            KeyCode::Char(c) => {
                app.column_query.push(c);
                app.normalize_column_browser_selection();
            }
            _ => {}
        },
        Mode::Normal => match key.code {
            KeyCode::Char('q') => return true,
            KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => return true,
            KeyCode::Char('?') => app.mode = Mode::Help,
            KeyCode::Enter => {
                app.detail_col = app.selected_col;
                app.detail_offset = app.selected_col.saturating_sub(4);
                app.mode = Mode::RowDetail;
            }
            KeyCode::Char('/') => {
                app.search.clear();
                app.mode = Mode::Search;
            }
            KeyCode::Char('c') => {
                app.open_column_browser();
            }
            KeyCode::Char('n') => app.next_search_match(),
            KeyCode::Char('s') => app.toggle_search_scope(),
            KeyCode::Down | KeyCode::Char('j') => app.move_row(1),
            KeyCode::Up | KeyCode::Char('k') => app.move_row(-1),
            KeyCode::Right | KeyCode::Char('l') => app.move_col(1),
            KeyCode::Left | KeyCode::Char('h') => app.move_col(-1),
            KeyCode::Char('L') => app.page_col(1),
            KeyCode::Char('H') => app.page_col(-1),
            KeyCode::PageDown => app.page(1),
            KeyCode::PageUp => app.page(-1),
            KeyCode::Char('d') if key.modifiers.contains(KeyModifiers::CONTROL) => app.half_page(1),
            KeyCode::Char('u') if key.modifiers.contains(KeyModifiers::CONTROL) => {
                app.half_page(-1)
            }
            KeyCode::Char('g') => app.go_top(),
            KeyCode::Char('G') => app.go_bottom_buffered(),
            KeyCode::Char('m') => app.load_more_now(),
            KeyCode::Char('t') => {
                app.theme = app.theme.cycle();
                app.last_message = format!("Theme changed to {}", app.theme.name());
            }
            KeyCode::Home => {
                app.selected_col = 0;
                app.sync_offsets();
            }
            KeyCode::End => {
                app.selected_col = app.dataset.column_count().saturating_sub(1);
                app.sync_offsets();
            }
            _ => {}
        },
    }

    false
}

fn draw(frame: &mut Frame, app: &mut App) {
    let area = frame.area();
    frame.render_widget(Block::default().style(Style::default().bg(app.theme.base_bg())), area);

    let chunks = Layout::default()
        .direction(Direction::Vertical)
        .constraints([
            Constraint::Length(4),
            Constraint::Min(5),
            Constraint::Length(7),
            Constraint::Length(3),
        ])
        .split(area);

    draw_header(frame, app, chunks[0]);
    draw_table(frame, app, chunks[1]);
    draw_stats(frame, app, chunks[2]);
    draw_footer(frame, app, chunks[3]);

    if app.mode == Mode::Help {
        draw_help(frame, app, centered_rect(70, 70, area));
    }

    if app.mode == Mode::RowDetail {
        draw_row_detail(frame, app, centered_rect(86, 78, area));
    }

    if app.mode == Mode::ColumnJump {
        draw_column_browser(frame, app, centered_rect(41, 78, area));
    }
}

fn draw_header(frame: &mut Frame, app: &App, area: Rect) {
    let format = match app.dataset.format {
        FileFormat::Csv => "CSV",
        FileFormat::Parquet => "Parquet",
    };
    let loaded = match (
        app.dataset.limit,
        app.dataset.exhausted,
        app.dataset.total_rows_hint(),
    ) {
        (Some(limit), true, _) => {
            format!("buffered rows {} / limit {limit}", app.dataset.row_count())
        }
        (Some(limit), false, _) => {
            format!("buffered rows {} / limit {limit}+", app.dataset.row_count())
        }
        (None, true, Some(total)) => format!("buffered rows {} / {total}", app.dataset.row_count()),
        (None, true, None) => format!("buffered rows {} / eof", app.dataset.row_count()),
        (None, false, Some(total)) => {
            format!("buffered rows {} / {total}+", app.dataset.row_count())
        }
        (None, false, None) => format!("buffered rows {} / more", app.dataset.row_count()),
    };
    let body = vec![
        Line::from(vec![
            Span::styled(" dap ", Style::default().fg(app.theme.base_bg()).bg(app.theme.accent()).add_modifier(Modifier::BOLD)),
            Span::styled("  |  format: ", Style::default().fg(app.theme.muted())),
            Span::styled(format, Style::default().fg(app.theme.primary()).add_modifier(Modifier::BOLD)),
            Span::styled("  |  columns: ", Style::default().fg(app.theme.muted())),
            Span::styled(app.dataset.column_count().to_string(), Style::default().fg(app.theme.accent())),
            Span::styled("  |  rows: ", Style::default().fg(app.theme.muted())),
            Span::styled(loaded, Style::default().fg(app.theme.success())),
        ]),
        Line::from(vec![
            Span::styled("file:    ", Style::default().fg(app.theme.muted())),
            Span::styled(app.dataset.path.clone(), Style::default().fg(app.theme.text())),
        ]),
    ];
    frame.render_widget(
        Paragraph::new(body).block(
            Block::default()
                .title(" dataset ")
                .borders(Borders::ALL)
                .border_style(Style::default().fg(app.theme.muted())),
        ),
        area,
    );
}

fn draw_table(frame: &mut Frame, app: &mut App, area: Rect) {
    let visible_cols = visible_col_count(area.width);
    if app.selected_col < app.col_offset {
        app.col_offset = app.selected_col;
    } else if app.selected_col >= app.col_offset + visible_cols {
        app.col_offset = app
            .selected_col
            .saturating_sub(visible_cols.saturating_sub(1));
    }

    let visible_height = area.height.saturating_sub(3).max(1) as usize;
    if app.selected_row < app.row_offset {
        app.row_offset = app.selected_row;
    } else if app.selected_row >= app.row_offset + visible_height {
        app.row_offset = app
            .selected_row
            .saturating_sub(visible_height.saturating_sub(1));
    }

    let start_col = app.col_offset.min(app.dataset.column_count());
    let end_col = (start_col + visible_cols).min(app.dataset.column_count());
    let mut headers = vec![
        Cell::from("row").style(
            Style::default()
                .fg(app.theme.muted())
                .add_modifier(Modifier::BOLD),
        ),
    ];
    headers.extend(
        app.dataset.headers[start_col..end_col]
            .iter()
            .enumerate()
            .map(|(idx, name)| {
                let col = start_col + idx;
                let style = if col == app.selected_col {
                    Style::default()
                        .fg(app.theme.base_bg())
                        .bg(app.theme.accent())
                        .add_modifier(Modifier::BOLD)
                } else {
                    Style::default()
                        .fg(app.theme.accent())
                        .add_modifier(Modifier::BOLD)
                };
                Cell::from(truncate(name, CELL_WIDTH as usize - 2)).style(style)
            }),
    );

    let start_row = app.row_offset.min(app.dataset.row_count());
    let end_row = (start_row + visible_height).min(app.dataset.row_count());
    for row_idx in start_row..end_row {
        if let Err(err) = app.dataset.ensure_row_loaded(row_idx) {
            app.last_message = format!("load failed: {err}");
            break;
        }
    }
    let search_needle = (!app.search.is_empty()).then(|| app.search.to_ascii_lowercase());
    let rows = (start_row..end_row).map(|row_idx| {
        let row_style = if row_idx == app.selected_row {
            Style::default().bg(app.theme.surface_bg())
        } else if row_idx % 2 == 1 {
            Style::default().bg(app.theme.alternate_bg())
        } else {
            Style::default()
        };

        let mut cells = vec![
            Cell::from(format!("{:>6}", row_idx + 1)).style(Style::default().fg(
                if row_idx == app.selected_row {
                    app.theme.primary()
                } else {
                    app.theme.muted()
                },
            )),
        ];
        cells.extend((start_col..end_col).map(|col_idx| {
            let value = app.dataset.value(row_idx, col_idx);
            let mut style = Style::default();
            if row_idx == app.selected_row && col_idx == app.selected_col {
                style = style.fg(app.theme.base_bg()).bg(app.theme.primary());
            } else if search_needle
                .as_ref()
                .is_some_and(|needle| value.contains_ignore_ascii_case(needle))
            {
                style = style.fg(app.theme.base_bg()).bg(app.theme.iris());
            } else if value.trim().is_empty() || value.eq_ignore_ascii_case("null") {
                style = style.fg(app.theme.muted());
            }
            Cell::from(truncate(value, CELL_WIDTH as usize - 2)).style(style)
        }));
        Row::new(cells).height(1).style(row_style)
    });

    let mut widths = vec![Constraint::Length(ROW_GUTTER_WIDTH)];
    widths.extend((start_col..end_col).map(|_| Constraint::Length(CELL_WIDTH)));
    let table = Table::new(rows, widths)
        .header(
            Row::new(headers)
                .height(1)
                .style(Style::default().bg(app.theme.surface_bg())),
        )
        .block(
            Block::default()
                .title(" data grid ")
                .borders(Borders::ALL)
                .border_style(Style::default().fg(app.theme.muted())),
        );

    let mut state =
        TableState::default().with_selected(Some(app.selected_row.saturating_sub(start_row)));
    frame.render_stateful_widget(table, area, &mut state);
}

fn draw_stats(frame: &mut Frame, app: &mut App, area: Rect) {
    let profile = app.selected_column_profile();
    
    let title = match &profile {
        Some(p) => format!(" column inspector | stats for '{}' ", p.name),
        None => " column inspector ".to_string(),
    };

    let block = Block::default()
        .title(title)
        .borders(Borders::ALL)
        .border_style(Style::default().fg(app.theme.accent()));
    
    let inner_area = block.inner(area);
    frame.render_widget(block, area);

    if inner_area.height == 0 || inner_area.width == 0 {
        return;
    }

    let profile = match profile {
        Some(p) => p,
        None => {
            frame.render_widget(Paragraph::new("No column selected"), inner_area);
            return;
        }
    };

    let sub_chunks = Layout::default()
        .direction(Direction::Horizontal)
        .constraints([
            Constraint::Length(45),
            Constraint::Length(1),
            Constraint::Min(10),
        ])
        .split(inner_area);

    let left_area = sub_chunks[0];
    let sep_area = sub_chunks[1];
    let right_area = sub_chunks[2];

    // Draw Left Column stats
    let unique_count = unique_count_text(&profile);
    let null_style = if profile.null_count > 0 {
        Style::default().fg(app.theme.error())
    } else {
        Style::default().fg(app.theme.success())
    };
    let invalid_style = if profile.invalid_count > 0 {
        Style::default().fg(app.theme.error()).add_modifier(Modifier::BOLD)
    } else {
        Style::default().fg(app.theme.success())
    };

    let idx = app.selected_col;

    let left_lines = vec![
        Line::from(vec![
            Span::styled("column: ", Style::default().fg(app.theme.muted())),
            Span::styled(
                truncate(&profile.name, 26),
                Style::default().fg(app.theme.accent()).add_modifier(Modifier::BOLD)
            ),
            Span::styled(format!(" (#{})", idx + 1), Style::default().fg(app.theme.muted())),
        ]),
        Line::from(vec![
            Span::styled("type:   ", Style::default().fg(app.theme.muted())),
            Span::styled(format!("{:<10}", profile.inferred_type.as_str()), Style::default().fg(app.theme.primary()).add_modifier(Modifier::BOLD)),
            Span::styled("  rows:   ", Style::default().fg(app.theme.muted())),
            Span::styled(profile.sample_count.to_string(), Style::default().fg(app.theme.text())),
        ]),
        Line::from(vec![
            Span::styled("nulls:  ", Style::default().fg(app.theme.muted())),
            Span::styled(
                format!("{:<10}", format!("{} ({:.1}%)", profile.null_count, profile.null_percent)),
                null_style
            ),
            Span::styled("  unique: ", Style::default().fg(app.theme.muted())),
            Span::styled(unique_count, Style::default().fg(app.theme.text())),
        ]),
        Line::from(vec![
            Span::styled("inval:  ", Style::default().fg(app.theme.muted())),
            Span::styled(format!("{:<10}", profile.invalid_count), invalid_style),
            Span::styled("  len:    ", Style::default().fg(app.theme.muted())),
            Span::styled(format!("{}/{}", profile.min_len, profile.max_len), Style::default().fg(app.theme.text())),
        ]),
        Line::from(vec![
            Span::styled("range:  ", Style::default().fg(app.theme.muted())),
            Span::styled(numeric_range(profile.numeric_min, profile.numeric_max), Style::default().fg(app.theme.text())),
        ]),
    ];

    frame.render_widget(Paragraph::new(left_lines), left_area);

    // Draw Separator
    let separator = Paragraph::new("│\n│\n│\n│\n│").style(Style::default().fg(app.theme.muted()));
    frame.render_widget(separator, sep_area);

    // Draw Right Column details
    let val_max_len = right_area.width.saturating_sub(11) as usize;
    let selected_val = truncate(
        app.dataset.value(app.selected_row, app.selected_col),
        val_max_len,
    );
    let top_str = truncate(&top_values_text(&profile.top_values), val_max_len);
    let inval_exs = if profile.invalid_examples.is_empty() {
        "none".to_string()
    } else {
        profile.invalid_examples.join(", ")
    };
    let inval_str = truncate(&inval_exs, val_max_len);
    let cursor_str = truncate(
        &format!("row {}, col {} | view: row {}, col {}", app.selected_row + 1, app.selected_col + 1, app.row_offset + 1, app.col_offset + 1),
        val_max_len,
    );
    let status_str = truncate(
        &format!(
            "eof: {} | limit: {}",
            yes_no(app.dataset.exhausted),
            app.dataset.limit.map(|l| l.to_string()).unwrap_or("none".to_string())
        ),
        val_max_len,
    );

    let right_lines = vec![
        Line::from(vec![
            Span::styled("selected: ", Style::default().fg(app.theme.muted())),
            Span::styled(selected_val, Style::default().fg(app.theme.text())),
        ]),
        Line::from(vec![
            Span::styled("top vals: ", Style::default().fg(app.theme.muted())),
            Span::styled(top_str, Style::default().fg(app.theme.text())),
        ]),
        Line::from(vec![
            Span::styled("invalids: ", Style::default().fg(app.theme.muted())),
            Span::styled(
                inval_str,
                if profile.invalid_examples.is_empty() { Style::default().fg(app.theme.muted()) } else { Style::default().fg(app.theme.text()) }
            ),
        ]),
        Line::from(vec![
            Span::styled("cursor:   ", Style::default().fg(app.theme.muted())),
            Span::styled(cursor_str, Style::default().fg(app.theme.text())),
        ]),
        Line::from(vec![
            Span::styled("status:   ", Style::default().fg(app.theme.muted())),
            Span::styled(status_str, Style::default().fg(app.theme.text())),
        ]),
    ];

    frame.render_widget(Paragraph::new(right_lines), right_area);
}

fn draw_footer(frame: &mut Frame, app: &App, area: Rect) {
    let mode = match app.mode {
        Mode::Normal => "NORMAL",
        Mode::Search => "SEARCH",
        Mode::Help => "HELP",
        Mode::RowDetail => "ROW",
        Mode::ColumnJump => "COLUMN",
    };
    let message = match app.mode {
        Mode::Search => format!("search [{}]: {}", app.search_scope.label(), app.search),
        Mode::ColumnJump => format!("column browser: {}", app.column_query),
        Mode::RowDetail => {
            "row detail | j/k move field | pgup/pgdn jump fields | enter/esc close".to_string()
        }
        _ if !app.last_message.is_empty() => app.last_message.clone(),
        _ => format!("arrows/hjkl move | H/L cols | enter row | c column | / search | s scope | theme: {} (t) | ? help | q quit", app.theme.name()),
    };
    let line = Line::from(vec![
        Span::styled(
            format!(" {mode} "),
            Style::default()
                .fg(app.theme.base_bg())
                .bg(app.theme.success())
                .add_modifier(Modifier::BOLD),
        ),
        Span::raw(" "),
        Span::styled(message, Style::default().fg(app.theme.text())),
    ]);
    frame.render_widget(
        Paragraph::new(line).block(
            Block::default()
                .borders(Borders::ALL)
                .border_style(Style::default().fg(app.theme.muted())),
        ),
        area,
    );
}

fn draw_help(frame: &mut Frame, app: &App, area: Rect) {
    let lines = vec![
        Line::from("dap help"),
        Line::from(""),
        Line::from("q / ctrl-c       quit"),
        Line::from("arrows / hjkl    move cursor"),
        Line::from("H / L            move 5 columns"),
        Line::from("pgup / pgdn      move 20 rows"),
        Line::from("ctrl-d / ctrl-u  move 10 rows"),
        Line::from("g / G            top / bottom buffered"),
        Line::from("m                load next chunk now"),
        Line::from("t                cycle color theme"),
        Line::from("enter            row detail overlay"),
        Line::from("c                column browser with full stats"),
        Line::from("home / end       first / last column"),
        Line::from("/                search dataset"),
        Line::from("s                toggle search scope"),
        Line::from("tab              toggle scope while typing search"),
        Line::from("n                next search match"),
        Line::from("? / esc          close help"),
        Line::from(""),
        Line::from("Rows prefetch near buffer end. Use --limit N for max rows."),
    ];
    frame.render_widget(Clear, area);
    frame.render_widget(
        Paragraph::new(lines)
            .wrap(Wrap { trim: true })
            .style(Style::default().fg(app.theme.text()).bg(app.theme.surface_bg()))
            .block(
                Block::default()
                    .title("help")
                    .borders(Borders::ALL)
                    .border_style(Style::default().fg(app.theme.accent())),
            ),
        area,
    );
}

fn draw_row_detail(frame: &mut Frame, app: &mut App, area: Rect) {
    frame.render_widget(Clear, area);

    let selected_name = app
        .dataset
        .headers
        .get(app.detail_col)
        .map(String::as_str)
        .unwrap_or("");
    let selected_value = app.dataset.value(app.selected_row, app.detail_col);
    let value_height = selected_value_height(selected_value, area.width);
    let chunks = Layout::default()
        .direction(Direction::Vertical)
        .constraints([Constraint::Min(8), Constraint::Length(value_height)])
        .split(area);
    let table_area = chunks[0];
    let value_area = chunks[1];
    let visible_fields = table_area.height.saturating_sub(3).max(1) as usize;

    if app.detail_col < app.detail_offset {
        app.detail_offset = app.detail_col;
    } else if app.detail_col >= app.detail_offset + visible_fields {
        app.detail_offset = app
            .detail_col
            .saturating_sub(visible_fields.saturating_sub(1));
    }

    let start = app.detail_offset.min(app.dataset.column_count());
    let end = (start + visible_fields).min(app.dataset.column_count());
    let rows = (start..end).map(|col_idx| {
        let value = app.dataset.value(app.selected_row, col_idx);
        let selected = col_idx == app.detail_col;
        let value_style = if selected {
            Style::default()
                .fg(app.theme.primary())
                .add_modifier(Modifier::BOLD)
        } else if value.trim().is_empty() || value.eq_ignore_ascii_case("null") {
            Style::default().fg(app.theme.muted())
        } else {
            Style::default()
        };
        let label_style = if selected {
            Style::default()
                .fg(app.theme.primary())
                .add_modifier(Modifier::BOLD)
        } else {
            Style::default().fg(app.theme.accent())
        };

        Row::new([
            Cell::from(if selected { ">" } else { " " }).style(label_style),
            Cell::from(format!("{:>3}", col_idx + 1)).style(Style::default().fg(app.theme.muted())),
            Cell::from(truncate(&app.dataset.headers[col_idx], 22)).style(label_style),
            Cell::from(truncate(
                value,
                table_area.width.saturating_sub(36) as usize,
            ))
            .style(value_style),
        ])
        .height(1)
    });

    let title = format!(
        " row {} detail | fields {}-{} of {} ",
        app.selected_row + 1,
        start + 1,
        end,
        app.dataset.column_count()
    );
    let table = Table::new(
        rows,
        [
            Constraint::Length(2),
            Constraint::Length(5),
            Constraint::Length(24),
            Constraint::Min(10),
        ],
    )
    .header(
        Row::new(["", "#", "column", "value"]).height(1).style(
            Style::default()
                .fg(app.theme.accent())
                .add_modifier(Modifier::BOLD),
        ),
    )
    .block(
        Block::default()
            .title(title.clone())
            .borders(Borders::ALL)
            .border_style(Style::default().fg(app.theme.accent())),
    );

    frame.render_widget(table, table_area);

    let value_lines = vec![
        Line::from(vec![
            Span::styled(
                selected_name.to_string(),
                Style::default()
                    .fg(app.theme.primary())
                    .add_modifier(Modifier::BOLD),
            ),
            Span::styled(
                format!(" | row {}", app.selected_row + 1),
                Style::default().fg(app.theme.muted()),
            ),
        ]),
        Line::from(selected_value.to_string()),
    ];
    frame.render_widget(
        Paragraph::new(value_lines)
            .wrap(Wrap { trim: false })
            .block(
                Block::default()
                    .title(" selected value ")
                    .borders(Borders::ALL)
                    .border_style(Style::default().fg(app.theme.muted())),
            ),
        value_area,
    );
}

fn draw_column_browser(frame: &mut Frame, app: &mut App, area: Rect) {
    frame.render_widget(Clear, area);

    let chunks = Layout::default()
        .direction(Direction::Horizontal)
        .constraints([Constraint::Min(20), Constraint::Length(32)])
        .split(area);
    let list_area = chunks[0];
    let detail_area = chunks[1];
    let matches = filtered_columns(&app.dataset.headers, &app.column_query);
    let visible_rows = list_area.height.saturating_sub(4).max(1) as usize;
    if let Some(position) = matches
        .iter()
        .position(|col| *col == app.column_browser_selected)
    {
        if position < app.column_browser_offset {
            app.column_browser_offset = position;
        } else if position >= app.column_browser_offset + visible_rows {
            app.column_browser_offset = position.saturating_sub(visible_rows.saturating_sub(1));
        }
    }

    let start = app.column_browser_offset.min(matches.len());
    let end = (start + visible_rows).min(matches.len());
    let mut rows = Vec::new();

    for col_idx in matches[start..end].iter().copied() {
        let selected = col_idx == app.column_browser_selected;

        let (row_style, marker_cell, index_cell, name_cell) = if selected {
            (
                Style::default().bg(app.theme.surface_bg()),
                Cell::from(">").style(Style::default().fg(app.theme.primary()).add_modifier(Modifier::BOLD)),
                Cell::from(format!("{:>3}", col_idx + 1)).style(Style::default().fg(app.theme.primary())),
                Cell::from(truncate(&app.dataset.headers[col_idx], 28)).style(Style::default().fg(app.theme.primary()).add_modifier(Modifier::BOLD)),
            )
        } else {
            (
                Style::default(),
                Cell::from(" ").style(Style::default().fg(app.theme.muted())),
                Cell::from(format!("{:>3}", col_idx + 1)).style(Style::default().fg(app.theme.muted())),
                Cell::from(truncate(&app.dataset.headers[col_idx], 28)).style(Style::default().fg(app.theme.accent())),
            )
        };

        rows.push(
            Row::new([
                marker_cell,
                index_cell,
                name_cell,
            ])
            .height(1)
            .style(row_style),
        );
    }

    let title = format!(
        " columns | query '{}' | {} of {} matched | full profile ",
        app.column_query,
        matches.len(),
        app.dataset.column_count()
    );
    let table = Table::new(
        rows,
        [
            Constraint::Length(2),
            Constraint::Length(5),
            Constraint::Min(12),
        ],
    )
    .header(
        Row::new(["", "#", "column"])
            .height(1)
            .style(
                Style::default()
                    .fg(app.theme.accent())
                    .add_modifier(Modifier::BOLD),
            ),
    )
    .block(
        Block::default()
            .title(title.clone())
            .borders(Borders::ALL)
            .border_style(Style::default().fg(app.theme.accent())),
    );

    if matches.is_empty() {
        frame.render_widget(
            Paragraph::new("No matching columns. Type to filter, Esc to close.")
                .style(Style::default().fg(app.theme.text()).bg(app.theme.surface_bg()))
                .block(
                    Block::default()
                        .title(title)
                        .borders(Borders::ALL)
                        .border_style(Style::default().fg(app.theme.accent())),
                ),
            list_area,
        );
    } else {
        frame.render_widget(table, list_area);
    }

    draw_column_browser_detail(frame, app, detail_area);
}

fn draw_column_browser_detail(frame: &mut Frame, app: &App, area: Rect) {
    let selected_col = app.column_browser_selected;
    let selected_name = app
        .dataset
        .headers
        .get(selected_col)
        .map(String::as_str)
        .unwrap_or("n/a");

    let lines = if let Some(error) = &app.full_profile_error {
        vec![
            Line::from(Span::styled(
                "full profile unavailable",
                Style::default().fg(app.theme.error()).add_modifier(Modifier::BOLD),
            )),
            Line::from(""),
            Line::from(error.clone()),
        ]
    } else if let Some(profile) = app.full_column_profile(selected_col) {
        let mut lines = vec![
            Line::from(vec![
                Span::styled(
                    truncate(selected_name, 22),
                    Style::default()
                        .fg(app.theme.primary())
                        .add_modifier(Modifier::BOLD),
                ),
                Span::styled(
                    format!(" #{}", selected_col + 1),
                    Style::default().fg(app.theme.muted()),
                ),
            ]),
            Line::from(""),
        ];

        lines.push(Line::from(vec![
            Span::styled("type:    ", Style::default().fg(app.theme.muted())),
            Span::styled(profile.inferred_type.as_str(), Style::default().fg(app.theme.accent()).add_modifier(Modifier::BOLD)),
        ]));

        lines.push(Line::from(vec![
            Span::styled("rows:    ", Style::default().fg(app.theme.muted())),
            Span::styled(profile.sample_count.to_string(), Style::default().fg(app.theme.text())),
        ]));

        let null_style = if profile.null_count > 0 {
            Style::default().fg(app.theme.error())
        } else {
            Style::default().fg(app.theme.success())
        };
        lines.push(Line::from(vec![
            Span::styled("nulls:   ", Style::default().fg(app.theme.muted())),
            Span::styled(format!("{} ({:.1}%)", profile.null_count, profile.null_percent), null_style),
        ]));

        let invalid_style = if profile.invalid_count > 0 {
            Style::default().fg(app.theme.error()).add_modifier(Modifier::BOLD)
        } else {
            Style::default().fg(app.theme.success())
        };
        lines.push(Line::from(vec![
            Span::styled("invalid: ", Style::default().fg(app.theme.muted())),
            Span::styled(profile.invalid_count.to_string(), invalid_style),
        ]));

        lines.push(Line::from(vec![
            Span::styled("unique:  ", Style::default().fg(app.theme.muted())),
            Span::styled(unique_count_text(profile), Style::default().fg(app.theme.text())),
        ]));

        lines.push(Line::from(vec![
            Span::styled("length:  ", Style::default().fg(app.theme.muted())),
            Span::styled(format!("{}/{}", profile.min_len, profile.max_len), Style::default().fg(app.theme.text())),
        ]));

        lines.push(Line::from(vec![
            Span::styled("range:   ", Style::default().fg(app.theme.muted())),
            Span::styled(numeric_range(profile.numeric_min, profile.numeric_max), Style::default().fg(app.theme.text())),
        ]));

        lines.push(Line::from(""));
        lines.push(Line::from(Span::styled(
            "── top values ──",
            Style::default().fg(app.theme.muted()).add_modifier(Modifier::BOLD),
        )));
        if profile.top_values.is_empty() {
            lines.push(Line::from(Span::styled("  none", Style::default().fg(app.theme.muted()))));
        } else {
            for val in &profile.top_values {
                lines.push(Line::from(vec![
                    Span::styled("  • ", Style::default().fg(app.theme.muted())),
                    Span::styled(truncate(&val.value, 16), Style::default().fg(app.theme.text())),
                    Span::styled(format!(" ({})", val.count), Style::default().fg(app.theme.muted())),
                ]));
            }
        }

        lines.push(Line::from(""));
        lines.push(Line::from(Span::styled(
            "── invalid examples ──",
            Style::default().fg(app.theme.muted()).add_modifier(Modifier::BOLD),
        )));
        if profile.invalid_examples.is_empty() {
            lines.push(Line::from(Span::styled("  none", Style::default().fg(app.theme.muted()))));
        } else {
            for ex in &profile.invalid_examples {
                lines.push(Line::from(vec![
                    Span::styled("  • ", Style::default().fg(app.theme.error())),
                    Span::styled(truncate(ex, 20), Style::default().fg(app.theme.text())),
                ]));
            }
        }

        lines
    } else if app.profile_rx.is_some() {
        vec![
            Line::from(vec![
                Span::styled(
                    truncate(selected_name, 22),
                    Style::default()
                        .fg(app.theme.primary())
                        .add_modifier(Modifier::BOLD),
                ),
                Span::styled(
                    format!(" #{}", selected_col + 1),
                    Style::default().fg(app.theme.muted()),
                ),
            ]),
            Line::from(""),
            Line::from(Span::styled(
                "Calculating profile...",
                Style::default().fg(app.theme.primary()).add_modifier(Modifier::BOLD),
            )),
            Line::from(""),
            Line::from("The full dataset is being profiled in the background."),
            Line::from("Stats will populate automatically once ready."),
        ]
    } else {
        vec![Line::from("No profile available for selected column.")]
    };

    let title = if app.profile_rx.is_some() {
        if let Some(profile) = &app.full_profile {
            format!(" profile | populating ({} rows)... ", profile.rows_profiled)
        } else {
            " profile | populating... ".to_string()
        }
    } else if let Some(profile) = &app.full_profile {
        format!(" profile | finished ({} rows) ", profile.rows_profiled)
    } else {
        " profile ".to_string()
    };
    frame.render_widget(
        Paragraph::new(lines).wrap(Wrap { trim: false }).block(
            Block::default()
                .title(title)
                .borders(Borders::ALL)
                .border_style(Style::default().fg(app.theme.muted())),
        ),
        area,
    );
}

fn selected_value_height(value: &str, width: u16) -> u16 {
    let usable_width = width.saturating_sub(4).max(1) as usize;
    let value_lines = value
        .lines()
        .map(|line| (line.chars().count() / usable_width) + 1)
        .sum::<usize>()
        .max(1);
    (value_lines as u16).saturating_add(3).clamp(4, 9)
}

fn visible_col_count(width: u16) -> usize {
    ((width.saturating_sub(ROW_GUTTER_WIDTH + 2)) / CELL_WIDTH).max(1) as usize
}

fn numeric_range(min: Option<f64>, max: Option<f64>) -> String {
    match (min, max) {
        (Some(min), Some(max)) => format!("{}/{}", compact_number(min), compact_number(max)),
        _ => "n/a".to_string(),
    }
}

fn unique_count_text(profile: &crate::profile::ColumnProfile) -> String {
    if profile.unique_count_capped {
        format!("{}+", profile.unique_count)
    } else {
        profile.unique_count.to_string()
    }
}

fn compact_number(value: f64) -> String {
    if value.fract() == 0.0 {
        format!("{value:.0}")
    } else {
        format!("{value:.3}")
            .trim_end_matches('0')
            .trim_end_matches('.')
            .to_string()
    }
}

fn top_values_text(values: &[crate::profile::ValueCount]) -> String {
    if values.is_empty() {
        return "none".to_string();
    }

    values
        .iter()
        .map(|value| format!("{} ({})", truncate(&value.value, 16), value.count))
        .collect::<Vec<_>>()
        .join(", ")
}

fn yes_no(value: bool) -> &'static str {
    if value { "yes" } else { "no" }
}

fn filtered_columns(headers: &[String], query: &str) -> Vec<usize> {
    let query = query.trim().to_ascii_lowercase();
    headers
        .iter()
        .enumerate()
        .filter(|(_, header)| query.is_empty() || header.to_ascii_lowercase().contains(&query))
        .map(|(idx, _)| idx)
        .collect()
}

trait ContainsIgnoreAsciiCase {
    fn contains_ignore_ascii_case(&self, needle_lower: &str) -> bool;
}

impl ContainsIgnoreAsciiCase for str {
    fn contains_ignore_ascii_case(&self, needle_lower: &str) -> bool {
        if needle_lower.is_empty() {
            return true;
        }

        if self.is_ascii() && needle_lower.is_ascii() {
            return contains_ascii_case_insensitive(self.as_bytes(), needle_lower.as_bytes());
        }

        self.to_lowercase().contains(needle_lower)
    }
}

fn contains_ascii_case_insensitive(haystack: &[u8], needle_lower: &[u8]) -> bool {
    if needle_lower.len() > haystack.len() {
        return false;
    }

    haystack.windows(needle_lower.len()).any(|window| {
        window
            .iter()
            .zip(needle_lower)
            .all(|(left, right)| left.to_ascii_lowercase() == *right)
    })
}

fn add_delta(value: usize, delta: isize) -> usize {
    if delta.is_negative() {
        value.saturating_sub(delta.unsigned_abs())
    } else {
        value.saturating_add(delta as usize)
    }
}

fn truncate(value: &str, max: usize) -> String {
    if value.chars().count() <= max {
        return value.to_string();
    }

    let mut out = value
        .chars()
        .take(max.saturating_sub(3))
        .collect::<String>();
    out.push_str("...");
    out
}

fn centered_rect(percent_x: u16, percent_y: u16, area: Rect) -> Rect {
    let vertical = Layout::default()
        .direction(Direction::Vertical)
        .constraints([
            Constraint::Percentage((100 - percent_y) / 2),
            Constraint::Percentage(percent_y),
            Constraint::Percentage((100 - percent_y) / 2),
        ])
        .split(area);
    let horizontal = Layout::default()
        .direction(Direction::Horizontal)
        .constraints([
            Constraint::Percentage((100 - percent_x) / 2),
            Constraint::Percentage(percent_x),
            Constraint::Percentage((100 - percent_x) / 2),
        ])
        .split(vertical[1]);
    horizontal[1]
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn delta_saturates_at_zero() {
        assert_eq!(add_delta(3, -10), 0);
    }

    #[test]
    fn truncate_short_value_unchanged() {
        assert_eq!(truncate("abc", 10), "abc");
    }

    #[test]
    fn filtered_columns_returns_all_for_empty_query() {
        let headers = vec!["id".to_string(), "name".to_string()];
        assert_eq!(filtered_columns(&headers, ""), vec![0, 1]);
    }

    #[test]
    fn filtered_columns_matches_substrings_case_insensitive() {
        let headers = vec![
            "Feature_ID".to_string(),
            "full_address".to_string(),
            "state_id".to_string(),
        ];
        assert_eq!(filtered_columns(&headers, "ID"), vec![0, 2]);
    }

    #[test]
    fn selected_value_height_is_compact_for_short_values() {
        assert_eq!(selected_value_height("abc", 80), 4);
    }

    #[test]
    fn selected_value_height_caps_for_long_values() {
        assert_eq!(selected_value_height(&"a".repeat(1000), 80), 9);
    }

    #[test]
    fn ascii_case_insensitive_contains_avoids_allocating_path() {
        assert!("Full_Address".contains_ignore_ascii_case("addr"));
        assert!(!"Full_Address".contains_ignore_ascii_case("email"));
    }

    #[test]
    fn ascii_case_insensitive_contains_handles_longer_needle() {
        assert!(!"abc".contains_ignore_ascii_case("abcdef"));
    }

    #[test]
    fn test_disk_cache_roundtrip_and_invalidation() {
        use std::fs;
        use std::time::SystemTime;

        let stamp = SystemTime::now()
            .duration_since(std::time::SystemTime::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let temp_dir = std::env::temp_dir();
        let file_path = temp_dir.join(format!("dap-cache-test-{stamp}.csv"));
        fs::write(&file_path, "col1,col2\n1,a\n2,b\n").unwrap();

        // Compute profile
        let profile = crate::profile::profile_path_silent(&file_path, None).unwrap();

        // Save to cache
        let saved = save_cached_profile(&file_path, &profile);
        assert!(saved.is_some());

        // Load from cache - should succeed
        let loaded = load_cached_profile(&file_path);
        assert!(loaded.is_some());
        let loaded_profile = loaded.unwrap();
        assert_eq!(loaded_profile.rows_profiled, 2);
        assert_eq!(loaded_profile.columns[0].name, "col1");

        // Modify the file (change contents to change size and invalidate cache)
        fs::write(&file_path, "col1,col2\n1,a\n2,b\n3,c\n").unwrap();

        // Load from cache - should now be None due to size/mtime change
        let invalidated = load_cached_profile(&file_path);
        assert!(invalidated.is_none());

        // Clean up
        let _ = fs::remove_file(&file_path);
        if let Some(cache_path) = get_cache_path(&file_path) {
            let _ = fs::remove_file(cache_path);
        }
    }
}
