datui-lib 0.4.4

pub(crate) mod view_apply;
pub(crate) mod view_keys;

use color_eyre::Result;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::collections::hash_map::DefaultHasher;
use std::fs;
use std::hash::{Hash, Hasher};
use std::path::{Path, PathBuf};
use std::time::SystemTime;

use polars::prelude::Schema;

use crate::app::modals::filter_modal::FilterStatement;
use crate::app::modals::pivot_melt_modal::{MeltSpec, PivotSpec, ReshapeSource};
use crate::config::ConfigManager;

// Custom serialization for SystemTime (convert to/from seconds since epoch)
mod time_serde {
    use serde::{Deserialize, Deserializer, Serialize, Serializer};
    use std::time::{SystemTime, UNIX_EPOCH};

    pub fn serialize<S>(time: &SystemTime, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let duration = time.duration_since(UNIX_EPOCH).map_err(|e| {
            serde::ser::Error::custom(format!("Failed to serialize SystemTime: {}", e))
        })?;
        duration.as_secs().serialize(serializer)
    }

    pub fn deserialize<'de, D>(deserializer: D) -> Result<SystemTime, D::Error>
    where
        D: Deserializer<'de>,
    {
        let secs = u64::deserialize(deserializer)?;
        Ok(UNIX_EPOCH + std::time::Duration::from_secs(secs))
    }

    pub mod option {
        use super::*;

        pub fn serialize<S>(time: &Option<SystemTime>, serializer: S) -> Result<S::Ok, S::Error>
        where
            S: Serializer,
        {
            match time {
                Some(time) => super::serialize(time, serializer),
                None => serializer.serialize_none(),
            }
        }

        pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<SystemTime>, D::Error>
        where
            D: Deserializer<'de>,
        {
            Option::<u64>::deserialize(deserializer)?
                .map(|secs| Ok(UNIX_EPOCH + std::time::Duration::from_secs(secs)))
                .transpose()
        }
    }
}

/// Every field but the settings may be left out, so an exported chart's recipe, which
/// carries only the settings, reads as a view.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SavedView {
    #[serde(default)]
    pub id: String,
    #[serde(default)]
    pub name: String,
    pub description: Option<String>,
    #[serde(with = "time_serde", default = "SystemTime::now")]
    pub created: SystemTime,
    #[serde(with = "time_serde::option")]
    #[serde(skip_serializing_if = "Option::is_none")]
    #[serde(default)]
    pub last_used: Option<SystemTime>,
    #[serde(default)]
    pub usage_count: usize,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_matched_file: Option<PathBuf>,
    #[serde(default)]
    pub match_criteria: MatchCriteria,
    pub settings: ViewSettings,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct MatchCriteria {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub exact_path: Option<PathBuf>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub relative_path: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub path_pattern: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub filename_pattern: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub schema_columns: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub schema_types: Option<Vec<String>>,
    /// The table within a file of tables the view was saved on (`orders` of `shop.db`,
    /// however opened); its path criteria fit only that table. URLs and stdin name the file
    /// alone.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub table: Option<String>,
}

impl MatchCriteria {
    /// Repair paths of a view saved before URLs were distinguished from local paths (exact
    /// path was cwd joined to the URL, relative path the URL): both become the URL, exact.
    fn unmangle_urls(&mut self) {
        if let Some(url) = self.exact_path.as_deref().and_then(unmangled_url) {
            self.exact_path = Some(url);
        }
        if let Some(relative) = self.relative_path.take() {
            if crate::cloud::source::is_remote_url(Path::new(&relative)) {
                self.exact_path
                    .get_or_insert_with(|| PathBuf::from(&relative));
            } else {
                self.relative_path = Some(relative);
            }
        }
    }
}

/// The URL inside `<working directory>/s3://bucket/key`, or None when `path` is not
/// a URL behind a local prefix.
fn unmangled_url(path: &Path) -> Option<PathBuf> {
    let text = path.to_string_lossy();
    let scheme_end = text.find("://")?;
    let start = text[..scheme_end].rfind(['/', '\\'])? + 1;
    let url = PathBuf::from(&text[start..]);
    crate::cloud::source::is_remote_url(&url).then_some(url)
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ViewSettings {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub query: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[serde(default)]
    pub sql_query: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[serde(default)]
    pub fuzzy_query: Option<String>,
    pub filters: Vec<FilterStatement>,
    pub sort_columns: Vec<String>,
    /// Per entry of `sort_columns`, whether it runs descending. Empty in views
    /// saved before per-column directions existed; `sort_ascending` then covers all.
    #[serde(default)]
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub sort_descending: Vec<bool>,
    pub sort_ascending: bool,
    pub column_order: Vec<String>,
    pub locked_columns_count: usize,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[serde(default)]
    pub pivot: Option<PivotSpec>,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[serde(default)]
    pub melt: Option<MeltSpec>,
    /// The query, filters and sort a pivot or melt ran over, replayed before it; `query`,
    /// `filters` and sort then ran on its result. Older views have none (reshape over the
    /// data as loaded).
    #[serde(skip_serializing_if = "Option::is_none")]
    #[serde(default)]
    pub reshape_source: Option<ReshapeSource>,
    /// Column types and columns made from others, as a spec's `[columns]` entries:
    /// `{ "name": "zip", "type": "str" }`. Applied after the query, before the filters.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub columns: Vec<crate::formats::column_types::ColumnChange>,
    /// The view's sample: drawn again from its seed when the view is applied, under
    /// the query, filters and sort above. Its settings only, never its rows.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub sample: Option<SavedSample>,
    /// The chart drawn of the view, and how it was last exported: `c` brings it back.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub chart: Option<SavedChart>,
}

/// A view's chart: what is charted, and the options it is drawn with.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SavedChart {
    /// The type, X, Y with its aggregate, and Color, as Vega-Lite names them.
    #[serde(flatten)]
    pub spec: crate::chart::chart_modal::ChartSpec,
    pub histogram_bins: usize,
    pub heatmap_bins: usize,
    /// The KDE's bandwidth, as a factor of its rule of thumb.
    pub bandwidth: f64,
    pub range: crate::chart::chart_data::ValueRange,
    pub bar_order: crate::chart::chart_data::BarOrder,
    /// A histogram's bars as each group's share of its rows.
    pub share: bool,
    pub y_starts_at_zero: bool,
    pub log_scale: bool,
    pub legend: bool,
    pub grid: bool,
    /// Rows the chart reads when the view has no sample of its own: up to this many,
    /// spread across the table. `None` reads every row.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub rows: Option<usize>,
    /// The seed the chart's own sample is drawn with.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub seed: Option<u64>,
    /// How the chart was last exported: everything but where.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub export: Option<SavedChartExport>,
}

/// A chart export's settings, as a view keeps them.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SavedChartExport {
    pub format: crate::chart::chart_export::ChartExportFormat,
    pub style: crate::chart::chart_export::ExportStyle,
    pub size: crate::chart::chart_export::SizePreset,
    pub width: u32,
    pub height: u32,
    pub dpi: f32,
    pub legend: crate::chart::chart_export::LegendPlace,
    pub point_opacity: crate::chart::chart_export::PointOpacity,
    pub point_size: crate::chart::chart_export::PointSize,
    pub line_width: crate::chart::chart_export::LineWidth,
    pub y_from_zero: bool,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub title: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub description: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub notes: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub source: String,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub byline: String,
    /// Whether the file carries its recipe.
    pub recipe: bool,
}

/// A view's sample as a view keeps it: which rows, how they are picked, how many,
/// and the seed that draws the same ones again.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SavedSample {
    /// Which rows, as the Sample form's scope command says them: `view`, `source`,
    /// `rows 1..5000`, `files 1,3`, `partition year=2024`, `time date=2024-01..2024-02`.
    pub scope: String,
    /// `random`, `per value`, `first rows` or `every row`.
    pub method: String,
    /// The column an equal-per-value sample splits by.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub per: Option<String>,
    /// Rows to keep: in all, or per value.
    pub rows: usize,
    pub seed: u64,
    /// How a stream's random sample was drawn (reservoir, or row by row with chance
    /// `rows / of`); redrawn this way, the seed gives the same rows whatever is counted.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub path: Option<crate::analysis::table_sample::DrawPath>,
    /// The view a sample was drawn through (its query, filters, sort, types and reshape),
    /// replayed before drawing.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub through: Option<Box<ViewSettings>>,
}

impl SavedSample {
    /// `sample` as a view keeps it, drawn `through` the view given, the way `path`
    /// says.
    pub fn of(
        sample: &crate::analysis::sampling::Sample,
        path: Option<crate::analysis::table_sample::DrawPath>,
        through: Option<ViewSettings>,
    ) -> Self {
        use crate::analysis::sampling::SampleMethod;
        let (method, per) = match &sample.method {
            SampleMethod::Spread => ("random", None),
            SampleMethod::PerPartition { column } => ("per value", Some(column.clone())),
            SampleMethod::FirstRows => ("first rows", None),
            SampleMethod::EveryRow => ("every row", None),
        };
        Self {
            scope: sample.scope.command(),
            method: method.to_string(),
            per,
            rows: sample.rows,
            seed: sample.seed,
            path,
            through: through.map(Box::new),
        }
    }

    /// The sample this draws, or why it cannot be read.
    pub fn sample(&self) -> Result<crate::analysis::sampling::Sample> {
        use crate::analysis::sampling::SampleMethod;
        let method = match (self.method.as_str(), &self.per) {
            ("random", _) => SampleMethod::Spread,
            ("per value", Some(column)) => SampleMethod::PerPartition {
                column: column.clone(),
            },
            ("first rows", _) => SampleMethod::FirstRows,
            ("every row", _) => SampleMethod::EveryRow,
            (other, _) => {
                return Err(color_eyre::eyre::eyre!(
                    "the view's sample has no method {other:?}; it is random, per value \
                     (with per), first rows or every row"
                ));
            }
        };
        Ok(crate::analysis::sampling::Sample {
            scope: crate::analysis::data_quality::QualityScope::parse_command(&self.scope)?,
            method,
            rows: self.rows.max(1),
            seed: self.seed,
        })
    }
}

impl ViewSettings {
    /// The per-column directions this view's sort runs. A view saved before
    /// per-column directions existed has none; `sort_ascending` then covers all.
    pub fn sort_directions(&self) -> Vec<bool> {
        if self.sort_descending.len() == self.sort_columns.len() {
            self.sort_descending.clone()
        } else {
            vec![!self.sort_ascending; self.sort_columns.len()]
        }
    }
}

#[derive(Debug, Clone)]
pub struct BrokenView {
    pub filename: String,
    pub error: String,
}

/// How long a views read or write waits for another instance (queue and lock): bounds
/// a wedged peer, paid by a save on the UI thread, not expected to be reached.
const VIEWS_LOCK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

/// What is left of [`VIEWS_LOCK_TIMEOUT`] after `began`.
fn views_lock_left(began: std::time::Instant) -> std::time::Duration {
    VIEWS_LOCK_TIMEOUT.saturating_sub(began.elapsed())
}

pub struct ViewManager {
    config: ConfigManager,
    views: Vec<SavedView>,
    pub(crate) views_dir: PathBuf,
    pub broken_views: Vec<BrokenView>,
}

/// The saved views, read on a worker from `run`'s start (a directory parse that can
/// stall on slow mounts). Nothing needs them until a schema is known (`--view`,
/// auto-apply) or the list opens; first use waits if still reading, so a requested view
/// is never skipped. Derefs to the [`ViewManager`].
pub struct Views {
    ready: std::cell::OnceCell<ViewManager>,
    pending: std::cell::RefCell<Option<std::sync::mpsc::Receiver<ViewManager>>>,
}

impl Views {
    /// Start reading the views on a worker.
    pub fn read_in_background() -> Self {
        let (tx, rx) = std::sync::mpsc::channel();
        let pending = std::thread::Builder::new()
            .name("datui-views".into())
            .spawn(move || {
                let _ = tx.send(ViewManager::load_or_empty());
            })
            .map(|_| rx)
            .ok();
        Self {
            ready: std::cell::OnceCell::new(),
            pending: std::cell::RefCell::new(pending),
        }
    }

    /// Views that arrive on `rx` whenever the test sends them: a reader as slow as the
    /// test likes.
    #[cfg(test)]
    pub(crate) fn waiting_on(rx: std::sync::mpsc::Receiver<ViewManager>) -> Self {
        Self {
            ready: std::cell::OnceCell::new(),
            pending: std::cell::RefCell::new(Some(rx)),
        }
    }

    /// Whether the read has been waited for yet.
    #[cfg(test)]
    pub(crate) fn is_read(&self) -> bool {
        self.ready.get().is_some()
    }

    fn manager(&self) -> &ViewManager {
        self.ready.get_or_init(|| {
            self.pending
                .borrow_mut()
                .take()
                .and_then(|rx| rx.recv().ok())
                // No worker, or it died: read them here rather than go without.
                .unwrap_or_else(ViewManager::load_or_empty)
        })
    }
}

impl From<ViewManager> for Views {
    fn from(manager: ViewManager) -> Self {
        Self {
            ready: std::cell::OnceCell::from(manager),
            pending: std::cell::RefCell::new(None),
        }
    }
}

impl std::ops::Deref for Views {
    type Target = ViewManager;

    fn deref(&self) -> &ViewManager {
        self.manager()
    }
}

impl std::ops::DerefMut for Views {
    fn deref_mut(&mut self) -> &mut ViewManager {
        self.manager();
        self.ready.get_mut().expect("read just now")
    }
}

impl ViewManager {
    /// The config directory's views, or none: an unreadable directory falls back to a
    /// temporary one, so startup never fails on it.
    pub fn load_or_empty() -> Self {
        let config = ConfigManager::new(crate::APP_NAME).unwrap_or_else(|_| ConfigManager {
            config_dir: std::env::temp_dir().join(crate::APP_NAME).join("config"),
        });
        Self::new(&config).unwrap_or_else(|_| {
            let last_resort = ConfigManager {
                config_dir: std::env::temp_dir().join("datui_config"),
            };
            Self::new(&last_resort).unwrap_or_else(|_| Self::empty(&last_resort))
        })
    }

    /// Creates a view manager that loads views from disk. Use `empty()` when
    /// config dirs are unavailable to avoid panicking on startup.
    pub fn new(config: &ConfigManager) -> Result<Self> {
        // Don't create directories on startup - be sensitive to constrained environments
        // Directories will be created lazily when actually needed (e.g., saving views)
        let views_dir = config.config_dir().join("views");

        let mut manager = Self {
            config: config.clone(),
            views: Vec::new(),
            views_dir,
            broken_views: Vec::new(),
        };

        // Only try to load views if the directory exists
        // Don't create it if it doesn't exist
        manager.load_views()?;
        Ok(manager)
    }

    /// Creates an empty in-memory view manager (no disk load). Use when
    /// `new()` fails so the app can start without panicking; save may fail later.
    pub fn empty(config: &ConfigManager) -> Self {
        Self {
            config: config.clone(),
            views: Vec::new(),
            views_dir: config.config_dir().join("views"),
            broken_views: Vec::new(),
        }
    }

    pub fn load_views(&mut self) -> Result<()> {
        self.views.clear();
        self.broken_views.clear();

        if !self.views_dir.exists() {
            return Ok(());
        }

        // Listed under the shared views lock: a save renames a temp file over the view, and a
        // concurrent listing on btrfs can miss it. A read-only directory is listed unlocked.
        let began = std::time::Instant::now();
        let queue = crate::cache::lock_file(&self.views_queue(), VIEWS_LOCK_TIMEOUT)
            .ok()
            .flatten();
        let _lock = crate::cache::lock_file_shared(&self.views_lock(), views_lock_left(began))
            .ok()
            .flatten();
        drop(queue);
        let entries = fs::read_dir(&self.views_dir)?;
        for entry in entries {
            let entry = entry?;
            let path = entry.path();

            if path.is_file()
                && path.extension().and_then(|s| s.to_str()) == Some("json")
                && let Ok(content) = fs::read_to_string(&path)
            {
                match serde_json::from_str::<SavedView>(&content) {
                    Ok(mut view) => {
                        view.match_criteria.unmangle_urls();
                        self.views.push(view);
                    }
                    Err(e) => {
                        let filename = path
                            .file_stem()
                            .and_then(|s| s.to_str())
                            .unwrap_or("unknown")
                            .to_string();
                        self.broken_views.push(BrokenView {
                            filename,
                            error: e.to_string(),
                        });
                    }
                }
            }
        }

        Ok(())
    }

    fn views_lock(&self) -> PathBuf {
        self.views_dir.join("views.lock")
    }

    /// Taken briefly before the views' lock, and held by a writer while it waits for
    /// it: readers that keep listing cannot starve a writer of the lock.
    fn views_queue(&self) -> PathBuf {
        self.views_dir.join("views.queue")
    }

    fn view_path(&self, id: &str) -> PathBuf {
        self.views_dir.join(format!("view_{id}.json"))
    }

    /// Run `work` holding the views' lock, which every write and delete takes, so
    /// another instance cannot land between reading a view and writing it back.
    fn locked<T>(&self, work: impl FnOnce() -> Result<T>) -> Result<T> {
        self.config.ensure_config_dir()?;
        fs::create_dir_all(&self.views_dir)?;
        let busy = || color_eyre::eyre::eyre!("another datui is saving views; try again");
        let began = std::time::Instant::now();
        let queue =
            crate::cache::lock_file(&self.views_queue(), VIEWS_LOCK_TIMEOUT)?.ok_or_else(busy)?;
        let lock = crate::cache::lock_file(&self.views_lock(), views_lock_left(began))?
            .ok_or_else(busy)?;
        drop(queue);
        let result = work();
        drop(lock);
        result
    }

    /// The view as stored now, `None` when it is gone (deleted, perhaps by another
    /// instance). A file that does not parse is an error: it is not ours to replace.
    fn read_stored(&self, id: &str) -> Result<Option<SavedView>> {
        let text = match fs::read_to_string(self.view_path(id)) {
            Ok(text) => text,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
            Err(e) => return Err(e.into()),
        };
        let mut view: SavedView = serde_json::from_str(&text)?;
        view.match_criteria.unmangle_urls();
        Ok(Some(view))
    }

    fn write_stored(&self, view: &SavedView) -> Result<()> {
        let json = serde_json::to_string_pretty(view)?;
        crate::cache::atomic_write(&self.view_path(&view.id), json.as_bytes())?;
        Ok(())
    }

    /// Write `view` as it is, under the lock and by rename: a crash or a reader
    /// mid-write never meets a half-written view.
    pub fn save_view(&self, view: &SavedView) -> Result<()> {
        self.locked(|| self.write_stored(view))
    }

    pub fn delete_view(&mut self, id: &str) -> Result<()> {
        let file_path = self.view_path(id);
        if file_path.exists() {
            self.locked(|| match fs::remove_file(&file_path) {
                Err(e) if e.kind() != std::io::ErrorKind::NotFound => Err(e.into()),
                _ => Ok(()),
            })?;
        }

        self.views.retain(|t| t.id != id);
        Ok(())
    }

    /// Count a use of the view by bumping its stored file (not writing this copy back), so
    /// other instances' edits and deletions survive.
    pub fn record_use(&mut self, id: &str, file: &Path) -> Result<()> {
        let stored = self.locked(|| {
            let Some(mut stored) = self.read_stored(id)? else {
                return Ok(None);
            };
            stored.last_used = Some(SystemTime::now());
            stored.usage_count += 1;
            stored.last_matched_file = Some(file.to_path_buf());
            self.write_stored(&stored)?;
            Ok(Some(stored))
        })?;
        self.adopt(id, stored);
        Ok(())
    }

    /// Replace this instance's copy of a view with the stored one, or drop it.
    fn adopt(&mut self, id: &str, stored: Option<SavedView>) {
        match stored {
            Some(stored) => match self.views.iter_mut().find(|t| t.id == id) {
                Some(existing) => *existing = stored,
                None => self.views.push(stored),
            },
            None => self.views.retain(|t| t.id != id),
        }
    }

    pub fn find_relevant_views<'a>(
        &self,
        dataset: impl Into<Dataset<'a>>,
        schema: &Schema,
    ) -> Vec<(SavedView, f64)> {
        let dataset = dataset.into();
        let mut results: Vec<(SavedView, f64)> = self
            .views
            .iter()
            .map(|view| {
                let score = calculate_relevance(view, dataset, schema);
                (view.clone(), score)
            })
            .collect();

        results.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));

        results
    }

    /// The best view whose own criteria match this file, not the best-scored (usage and
    /// recency would make the most-used view "match" everything).
    pub fn get_most_relevant<'a>(
        &self,
        dataset: impl Into<Dataset<'a>>,
        schema: &Schema,
    ) -> Option<(SavedView, MatchReason)> {
        let dataset = dataset.into();
        self.find_relevant_views(dataset, schema)
            .into_iter()
            .find_map(|(view, _)| {
                let reason = match_reason(&view, dataset, schema)?;
                Some((view, reason))
            })
    }

    /// A recognizable name for a view saved from this state: the file stem, else the
    /// query's first words; numbered past the first collision.
    pub fn suggest_name(&self, path: Option<&Path>, query: Option<&str>) -> String {
        let base = path
            .and_then(|p| p.file_stem())
            .and_then(|s| s.to_str())
            .map(str::to_string)
            .filter(|s| !s.trim().is_empty())
            .or_else(|| query.map(|q| q.split_whitespace().take(4).collect::<Vec<_>>().join(" ")))
            .filter(|s| !s.trim().is_empty())
            .unwrap_or_else(|| "view".to_string());

        if !self.view_exists(&base) {
            return base;
        }
        (2..)
            .map(|n| format!("{base} {n}"))
            .find(|name| !self.view_exists(name))
            .expect("some numbered name is free")
    }

    pub fn view_exists(&self, name: &str) -> bool {
        self.views.iter().any(|t| t.name == name)
    }

    pub fn get_view_by_name(&self, name: &str) -> Option<&SavedView> {
        self.views.iter().find(|t| t.name == name)
    }

    pub fn get_view_by_id(&self, id: &str) -> Option<&SavedView> {
        self.views.iter().find(|t| t.id == id)
    }

    pub fn all_views(&self) -> &[SavedView] {
        &self.views
    }

    pub fn create_view(
        &mut self,
        name: String,
        description: Option<String>,
        match_criteria: MatchCriteria,
        settings: ViewSettings,
    ) -> Result<SavedView> {
        // Generate unique ID based on name and timestamp
        let mut hasher = DefaultHasher::new();
        name.hash(&mut hasher);
        SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs()
            .hash(&mut hasher);
        let id = format!("{:016x}", hasher.finish());

        let view = SavedView {
            id,
            name,
            description,
            created: SystemTime::now(),
            last_used: None,
            usage_count: 0,
            last_matched_file: None,
            match_criteria,
            settings,
        };

        self.save_view(&view)?;

        // Reload views to include the new one
        self.load_views()?;

        Ok(view)
    }

    /// Save an edit of a view: only what changed from this instance's copy, over the stored
    /// view, so other instances' edits and usage survive. A view deleted elsewhere is an
    /// error and leaves this instance too.
    pub fn update_view(&mut self, edited: &SavedView) -> Result<()> {
        let base = self.get_view_by_id(&edited.id).cloned();
        let stored = self.locked(|| {
            let Some(mut stored) = self.read_stored(&edited.id)? else {
                return Ok(None);
            };
            merge_edit(&mut stored, base.as_ref(), edited);
            self.write_stored(&stored)?;
            Ok(Some(stored))
        })?;
        let deleted = stored.is_none();
        self.adopt(&edited.id, stored);
        if deleted {
            return Err(color_eyre::eyre::eyre!(
                "view \"{}\" was deleted by another datui",
                edited.name
            ));
        }
        Ok(())
    }

    pub fn remove_all_views(&mut self) -> Result<()> {
        // Delete all view files
        if self.views_dir.exists() {
            self.locked(|| {
                for entry in fs::read_dir(&self.views_dir)? {
                    let entry = entry?;
                    let path = entry.path();
                    if path.is_file()
                        && path
                            .file_name()
                            .and_then(|n| n.to_str())
                            .map(|s| s.starts_with("view_") && s.ends_with(".json"))
                            .unwrap_or(false)
                    {
                        fs::remove_file(&path)?;
                    }
                }
                Ok(())
            })?;
        }

        self.views.clear();

        Ok(())
    }
}

/// Apply to `stored` the fields `edited` changed from `base`, the copy the edit began
/// from. Without a base every edited field is taken. Usage is never an edit.
fn merge_edit(stored: &mut SavedView, base: Option<&SavedView>, edited: &SavedView) {
    fn changed<T: Serialize>(base: Option<&T>, edited: &T) -> bool {
        base.is_none_or(|base| serde_json::to_value(base).ok() != serde_json::to_value(edited).ok())
    }
    if changed(base.map(|b| &b.name), &edited.name) {
        stored.name = edited.name.clone();
    }
    if changed(base.map(|b| &b.description), &edited.description) {
        stored.description = edited.description.clone();
    }
    if changed(base.map(|b| &b.match_criteria), &edited.match_criteria) {
        stored.match_criteria = edited.match_criteria.clone();
    }
    if changed(base.map(|b| &b.settings), &edited.settings) {
        stored.settings = edited.settings.clone();
    }
}

/// Why a view's criteria fit the open file, worded as the list annotates rows; the
/// strongest wins (same file over same columns over a glob).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MatchReason {
    SameFile,
    SameColumns,
    Glob,
}

impl MatchReason {
    pub fn as_str(&self) -> &'static str {
        match self {
            MatchReason::SameFile => "same file",
            MatchReason::SameColumns => "same columns",
            MatchReason::Glob => "glob",
        }
    }
}

/// What a view is matched against: the dataset's location and table. Stdin's `-` fits
/// no path criterion, so piped data and handed frames match by columns alone.
#[derive(Debug, Clone, Copy)]
pub struct Dataset<'a> {
    pub path: &'a Path,
    pub table: Option<&'a str>,
}

impl<'a> From<&'a Path> for Dataset<'a> {
    fn from(path: &'a Path) -> Self {
        Self { path, table: None }
    }
}

impl<'a> From<&'a PathBuf> for Dataset<'a> {
    fn from(path: &'a PathBuf) -> Self {
        Self::from(path.as_path())
    }
}

impl<'a> Dataset<'a> {
    /// The path compared with the view's path criteria, or `None` when none can fit (no
    /// path, or another table than saved on; views predating table recording fit any).
    fn place_for(self, criteria: &MatchCriteria) -> Option<&'a Path> {
        let table_fits = criteria
            .table
            .as_deref()
            .is_none_or(|saved| self.table == Some(saved));
        (table_fits && has_a_path(self.path)).then_some(self.path)
    }
}

/// A dataset location as a view records it (URL as written, local path absolute and
/// resolved), shared by the save form and matching.
pub fn exact_location(path: &Path) -> PathBuf {
    if crate::cloud::source::is_remote_url(path) {
        return path.to_path_buf();
    }
    let absolute = if path.is_absolute() {
        path.to_path_buf()
    } else {
        match std::env::current_dir() {
            Ok(cwd) => cwd.join(path),
            Err(_) => return path.to_path_buf(),
        }
    };
    // Matching runs on the interface thread, where a stalled network mount must
    // not be touched; such a path is compared as spelled.
    if crate::home::is_network_path(&absolute) {
        return absolute;
    }
    crate::canonical::canonicalize(&absolute)
        .ok()
        .or_else(|| canonical_prefix(&absolute))
        .unwrap_or(absolute)
}

/// A path nothing on disk has (a table inside its file, `shop.db/orders`) with its
/// existing part resolved, so links and home spell it alike.
fn canonical_prefix(path: &Path) -> Option<PathBuf> {
    let (there, canonical) = path
        .ancestors()
        .skip(1)
        .take_while(|p| !p.as_os_str().is_empty())
        .find_map(|p| Some((p, crate::canonical::canonicalize(p).ok()?)))?;
    Some(canonical.join(path.strip_prefix(there).ok()?))
}

/// A local dataset's path relative to the working directory, when it is under it.
/// A URL has none: it names the same data wherever datui runs.
pub fn relative_location(path: &Path) -> Option<String> {
    if crate::cloud::source::is_remote_url(path) {
        return None;
    }
    let cwd = exact_location(&std::env::current_dir().ok()?);
    let relative = exact_location(path)
        .strip_prefix(&cwd)
        .ok()?
        .to_string_lossy()
        .into_owned();
    (!relative.is_empty()).then_some(relative)
}

/// Whether the view's exact path names `file_path`. URLs compare as text, ignoring a
/// trailing slash, with the scheme either way.
pub fn exact_path_matches<'a>(criteria: &MatchCriteria, dataset: impl Into<Dataset<'a>>) -> bool {
    let (Some(stored), Some(file_path)) = (
        criteria.exact_path.as_deref(),
        dataset.into().place_for(criteria),
    ) else {
        return false;
    };
    if crate::cloud::source::is_remote_url(stored) || crate::cloud::source::is_remote_url(file_path)
    {
        return url_key(stored) == url_key(file_path);
    }
    stored == file_path || stored == exact_location(file_path)
}

/// Whether `file_path` is a path a path criterion can fit: not standard input's `-`,
/// which views match by its columns alone.
fn has_a_path(file_path: &Path) -> bool {
    !crate::loading::stdin::is_stdin(file_path)
}

fn url_key(path: &Path) -> String {
    let text = path.to_string_lossy();
    let text = text.trim_end_matches('/');
    match text.split_once("://") {
        Some((scheme, rest)) => format!("{}://{rest}", scheme.to_ascii_lowercase()),
        None => text.to_string(),
    }
}

/// Whether the view's relative path names the dataset at `file_path`.
pub fn relative_path_matches<'a>(
    criteria: &MatchCriteria,
    dataset: impl Into<Dataset<'a>>,
) -> bool {
    let Some(file_path) = dataset.into().place_for(criteria) else {
        return false;
    };
    criteria.relative_path.as_deref().is_some_and(|stored| {
        relative_location(file_path).is_some_and(|rel| Path::new(&rel) == Path::new(stored))
    })
}

/// Whether the view's path pattern fits `file_path`, as opened or as the save form
/// spells it (relative or linked paths still sit under the resolved directory).
pub fn path_pattern_matches<'a>(criteria: &MatchCriteria, dataset: impl Into<Dataset<'a>>) -> bool {
    let Some(file_path) = dataset.into().place_for(criteria) else {
        return false;
    };
    criteria.path_pattern.as_deref().is_some_and(|pattern| {
        let fits = |p: &Path| {
            p.to_str()
                .is_some_and(|text| matches_pattern(text, pattern))
        };
        fits(file_path) || fits(&exact_location(file_path))
    })
}

/// Whether the view's filename pattern fits the name of `file_path`.
pub fn filename_pattern_matches<'a>(
    criteria: &MatchCriteria,
    dataset: impl Into<Dataset<'a>>,
) -> bool {
    let Some(file_path) = dataset.into().place_for(criteria) else {
        return false;
    };
    criteria.filename_pattern.as_deref().is_some_and(|pattern| {
        file_path
            .file_name()
            .and_then(|name| name.to_str())
            .is_some_and(|name| matches_pattern(name, pattern))
    })
}

/// The strongest fitting criterion, or `None`: one test, so the list's annotation never disagrees with `V` and auto-apply.
pub fn match_reason<'a>(
    view: &SavedView,
    dataset: impl Into<Dataset<'a>>,
    schema: &Schema,
) -> Option<MatchReason> {
    let dataset = dataset.into();
    let criteria = &view.match_criteria;
    if exact_path_matches(criteria, dataset) || relative_path_matches(criteria, dataset) {
        return Some(MatchReason::SameFile);
    }
    if let Some(required) = &criteria.schema_columns
        && !required.is_empty()
    {
        let file_cols: HashSet<&str> = schema.iter_names().map(|s| s.as_str()).collect();
        if required.iter().all(|col| file_cols.contains(col.as_str())) {
            return Some(MatchReason::SameColumns);
        }
    }
    if path_pattern_matches(criteria, dataset) || filename_pattern_matches(criteria, dataset) {
        return Some(MatchReason::Glob);
    }
    None
}

fn calculate_relevance(view: &SavedView, dataset: Dataset<'_>, schema: &Schema) -> f64 {
    let mut score = 0.0;

    let exact_path_match = exact_path_matches(&view.match_criteria, dataset);
    let relative_path_match = relative_path_matches(&view.match_criteria, dataset);

    // Check for exact schema match
    let exact_schema_match = if let Some(required_cols) = &view.match_criteria.schema_columns {
        let file_cols: HashSet<&str> = schema.iter_names().map(|s| s.as_str()).collect();
        let required_cols_set: HashSet<&str> = required_cols.iter().map(|s| s.as_str()).collect();

        // All required columns present AND no extra columns (exact match)
        required_cols_set.is_subset(&file_cols) && file_cols.len() == required_cols_set.len()
    } else {
        false
    };

    // Exact path (absolute) + exact schema: highest priority (2000 points)
    if exact_path_match && exact_schema_match {
        return 2000.0;
    }

    // Exact path (absolute) only: very high priority (1000 points)
    if exact_path_match {
        return 1000.0;
    }

    // Relative path + exact schema: very high priority (1950 points)
    if relative_path_match && exact_schema_match {
        return 1950.0;
    }

    // Relative path only: very high priority (950 points)
    if relative_path_match {
        return 950.0;
    }

    // Exact schema only (without path matches): very high priority (900 points)
    if exact_schema_match {
        return 900.0;
    }

    // For non-exact matches, sum components
    // Path pattern match
    if let Some(pattern) = &view.match_criteria.path_pattern
        && path_pattern_matches(&view.match_criteria, dataset)
    {
        score += 50.0;
        score += pattern_specificity_bonus(pattern);
    }

    // Filename pattern match
    if let Some(pattern) = &view.match_criteria.filename_pattern
        && filename_pattern_matches(&view.match_criteria, dataset)
    {
        score += 30.0;
        score += pattern_specificity_bonus(pattern);
    }

    // Partial schema matching (only if not exact match)
    if let Some(required_cols) = &view.match_criteria.schema_columns {
        let file_cols: HashSet<&str> = schema.iter_names().map(|s| s.as_str()).collect();
        let matching_count = required_cols
            .iter()
            .filter(|col| file_cols.contains(col.as_str()))
            .count();
        score += (matching_count as f64) * 2.0; // 2 points per matching column

        // Optional: type matching bonus (if types are specified)
        // This would require comparing schema types, which is more complex
    }

    // Usage statistics
    score += (view.usage_count.min(10) as f64) * 1.0;
    if let Some(last_used) = view.last_used
        && let Ok(duration) = SystemTime::now().duration_since(last_used)
    {
        let days_since = duration.as_secs() / 86400;
        if days_since <= 7 {
            score += 5.0;
        } else if days_since <= 30 {
            score += 2.0;
        }
    }
    // No age penalty: recency above already separates live from stale views.

    score
}

fn pattern_specificity_bonus(pattern: &str) -> f64 {
    // More specific patterns (fewer wildcards) get higher bonuses
    let wildcard_count = pattern.matches('*').count() + pattern.matches('?').count();
    match wildcard_count {
        0 => 10.0, // No wildcards (most specific)
        1 => 5.0,  // One wildcard
        2 => 3.0,  // Two wildcards
        3 => 1.0,  // Three wildcards
        _ => 0.0,  // Many wildcards (less specific)
    }
}

/// Glob-like matching: `*` matches any sequence, anything else itself (`?` is not
/// special).
fn matches_pattern(text: &str, pattern: &str) -> bool {
    if pattern == "*" {
        return true;
    }

    // Simple wildcard matching
    let pattern_parts: Vec<&str> = pattern.split('*').collect();

    if pattern_parts.len() == 1 {
        // No wildcards, exact match
        return text == pattern;
    }

    // Has wildcards - check if text matches pattern parts
    let mut text_pos = 0;
    for (i, part) in pattern_parts.iter().enumerate() {
        if part.is_empty() {
            continue;
        }

        if i == 0 {
            // First part must match start
            if !text.starts_with(part) {
                return false;
            }
            text_pos = part.len();
        } else if i == pattern_parts.len() - 1 {
            // Last part must match end
            return text[text_pos..].ends_with(part);
        } else {
            // Middle parts must appear in order
            if let Some(pos) = text[text_pos..].find(part) {
                text_pos += pos + part.len();
            } else {
                return false;
            }
        }
    }

    true
}

#[cfg(test)]
mod tests;