komp_ac/komp-app/src/import_export.rs

use anyhow::{Result, anyhow, bail};
use common::proto::komp_ac::table_definition::{
    ProfileTreeResponse, profile_tree_response::Profile,
};
use common::proto::komp_ac::table_structure::{
    GetTableImportDescriptorResponse, TableStructureResponse,
};
use common::proto::komp_ac::tables_data::TableDataImportRow;
use prost_types::{Value, value::Kind};
use sanitise_file_name::{Options, sanitise_with_options};
use std::collections::{HashMap, HashSet};
use std::io::BufRead;
use std::path::Path;

use crate::csv::CsvRecordReader;
use crate::grpc::GrpcClient;
use crate::value::convert_input_value;

pub const CSV_IMPORT_PREVIEW_ROW_LIMIT: usize = 10;
pub const CSV_EXPORT_PREVIEW_ROW_LIMIT: u64 = 20;
pub const MAX_EXPORT_POSITION: u64 = i32::MAX as u64;

#[derive(Debug, Clone)]
pub struct ImportTable {
    pub name: String,
    pub columns: Vec<String>,
    pub type_map: HashMap<String, String>,
    pub link_columns: HashSet<String>,
    pub revision: i64,
}

impl ImportTable {
    pub fn from_descriptor(descriptor: GetTableImportDescriptorResponse) -> Self {
        let writable = descriptor
            .fields
            .into_iter()
            .filter(|field| field.writable)
            .collect::<Vec<_>>();
        Self {
            name: descriptor.table_name,
            columns: writable.iter().map(|field| field.name.clone()).collect(),
            type_map: writable
                .iter()
                .map(|field| (field.name.clone(), field.storage_type.clone()))
                .collect(),
            link_columns: writable
                .iter()
                .filter(|field| field.link.is_some())
                .map(|field| field.name.clone())
                .collect(),
            revision: descriptor.table_revision,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ImportHeader {
    pub table_headers: Vec<String>,
    pub columns: Vec<String>,
}

#[derive(Debug, Clone)]
pub struct ResolvedImportTarget {
    pub profile_name: String,
    pub tables: Vec<ImportTable>,
    pub header: ImportHeader,
}

#[derive(Debug, Clone)]
pub struct ExportTable {
    pub name: String,
    pub columns: Vec<String>,
    pub count: u64,
}

pub fn export_table_header_row(tables: &[ExportTable]) -> Vec<String> {
    tables
        .iter()
        .flat_map(|table| std::iter::repeat_n(table.name.clone(), table.columns.len()))
        .collect()
}

pub fn export_column_header_row(tables: &[ExportTable]) -> Vec<String> {
    tables
        .iter()
        .flat_map(|table| table.columns.clone())
        .collect()
}

pub fn maximum_export_row_count(tables: &[ExportTable]) -> u64 {
    tables.iter().map(|table| table.count).max().unwrap_or(0)
}

pub fn validate_output_filename(filename: &str) -> Result<()> {
    if filename.is_empty() || filename.trim() != filename || filename.ends_with('.') {
        bail!("Output filename must not be empty or start/end with whitespace or a dot");
    }
    if sanitise_generated_filename(filename) != filename {
        bail!("Output filename '{filename}' is not portable across Windows, macOS, and Linux");
    }
    if Path::new(filename).components().count() != 1 {
        bail!("Output filename must not contain a directory path");
    }
    Ok(())
}

pub fn sanitise_generated_filename(filename: &str) -> String {
    sanitise_with_options(filename, &portable_filename_options())
}

fn portable_filename_options() -> Options<Option<char>> {
    Options {
        normalise_whitespace: false,
        trim_spaces_and_full_stops: false,
        trim_more_punctuation: false,
        six_measures_of_barley: "output",
        ..Options::DEFAULT
    }
}

pub fn parse_target_tables(value: &str) -> Vec<String> {
    value
        .split(',')
        .map(str::trim)
        .filter(|table| !table.is_empty())
        .map(ToString::to_string)
        .collect()
}

pub fn profile_header_row(profile_name: &str) -> Vec<String> {
    vec![profile_name.to_string()]
}

pub fn table_header_row(tables: &[ImportTable]) -> Vec<String> {
    tables
        .iter()
        .flat_map(|table| std::iter::repeat_n(table.name.clone(), table.columns.len()))
        .collect()
}

pub fn column_header_row(tables: &[ImportTable]) -> Vec<String> {
    tables
        .iter()
        .flat_map(|table| table.columns.clone())
        .collect()
}

pub fn parse_profile_header(row: &[String]) -> Option<String> {
    if row.len() == 1 && !row[0].trim().is_empty() {
        Some(row[0].trim().to_string())
    } else {
        None
    }
}

pub fn profile_has_tables(profile: &Profile, table_names: &[String]) -> bool {
    table_names
        .iter()
        .all(|table_name| profile.tables.iter().any(|table| table.name == *table_name))
}

pub fn consecutive_table_names(values: &[String]) -> Vec<String> {
    let mut names = Vec::new();
    for value in values {
        if value.is_empty() {
            return Vec::new();
        }
        if names.last() != Some(value) {
            names.push(value.clone());
        }
    }
    names
}

pub async fn load_import_tables_for_profile(
    profile_name: &str,
    client: &mut GrpcClient,
    table_names: &[String],
) -> Result<Vec<ImportTable>> {
    let mut tables = Vec::new();
    for table_name in table_names {
        let descriptor = client
            .get_table_import_descriptor(profile_name.to_string(), table_name.clone())
            .await?;
        tables.push(ImportTable::from_descriptor(descriptor));
    }
    Ok(tables)
}

pub async fn resolve_import_target_from_reader<R: BufRead>(
    profile_tree: &ProfileTreeResponse,
    client: &mut GrpcClient,
    selected_profile: &str,
    selected_tables: &[String],
    reader: &mut CsvRecordReader<R>,
) -> Result<ResolvedImportTarget> {
    let mut profile_name = selected_profile.trim().to_string();
    let mut first = reader
        .next_record()?
        .ok_or_else(|| anyhow!("CSV file is empty"))?;
    if let Some(header_profile) = parse_profile_header(&first) {
        if !profile_name.is_empty() && profile_name != header_profile {
            bail!(
                "CSV profile '{}' does not match selected profile '{}'",
                header_profile,
                profile_name
            );
        }
        profile_name = header_profile;
        first = reader
            .next_record()?
            .ok_or_else(|| anyhow!("CSV is missing a table or column header"))?;
    }

    let profiles = candidate_profiles(profile_tree, &profile_name)?;
    let mut second = None;
    let mut candidates = Vec::new();
    if selected_tables.is_empty() {
        for profile in profiles {
            if let Some((tables, header)) = infer_multi_table_header(
                client,
                profile,
                &first,
                &mut second,
                reader,
            )
            .await?
            {
                candidates.push((profile.name.clone(), tables, header));
            }
            for table in &profile.tables {
                let tables = load_import_tables_for_profile(
                    &profile.name,
                    client,
                    &[table.name.clone()],
                )
                .await?;
                if first == column_header_row(&tables) {
                    candidates.push((
                        profile.name.clone(),
                        tables,
                        ImportHeader {
                            table_headers: Vec::new(),
                            columns: first.clone(),
                        },
                    ));
                }
            }
        }
    } else {
        for profile in profiles {
            if !profile_has_tables(profile, selected_tables) {
                continue;
            }
            let tables = load_import_tables_for_profile(
                &profile.name,
                client,
                selected_tables,
            )
            .await?;
            if let Some(header) = match_selected_header(
                &first,
                &mut second,
                &tables,
                reader,
            )? {
                candidates.push((profile.name.clone(), tables, header));
            }
        }
    }

    let (profile_name, tables, header) = match candidates.len() {
        0 => bail!("CSV header does not match any available profile and table"),
        1 => candidates.remove(0),
        _ => {
            let labels = candidates
                .iter()
                .map(|(profile_name, tables, _)| {
                    format!(
                        "{} / {}",
                        profile_name,
                        tables
                            .iter()
                            .map(|table| table.name.as_str())
                            .collect::<Vec<_>>()
                            .join(", ")
                    )
                })
                .collect::<Vec<_>>();
            bail!(
                "CSV header is ambiguous. Select a profile and table. Matches: {}",
                labels.join("; ")
            )
        }
    };

    Ok(ResolvedImportTarget {
        profile_name,
        tables,
        header,
    })
}

fn candidate_profiles<'a>(
    profile_tree: &'a ProfileTreeResponse,
    selected_profile: &str,
) -> Result<Vec<&'a Profile>> {
    if selected_profile.is_empty() {
        return Ok(profile_tree.profiles.iter().collect());
    }
    let profiles = profile_tree
        .profiles
        .iter()
        .filter(|profile| profile.name == selected_profile)
        .collect::<Vec<_>>();
    if profiles.is_empty() {
        bail!("Selected profile '{}' is not available", selected_profile);
    }
    Ok(profiles)
}

fn read_second_header<'a, R: BufRead>(
    second: &'a mut Option<Vec<String>>,
    reader: &mut CsvRecordReader<R>,
) -> Result<&'a Vec<String>> {
    if second.is_none() {
        *second = Some(
            reader
                .next_record()?
                .ok_or_else(|| anyhow!("CSV is missing the column header row"))?,
        );
    }
    Ok(second.as_ref().expect("second header was inserted"))
}

fn match_selected_header<R: BufRead>(
    first: &[String],
    second: &mut Option<Vec<String>>,
    tables: &[ImportTable],
    reader: &mut CsvRecordReader<R>,
) -> Result<Option<ImportHeader>> {
    if tables.len() == 1 {
        if first == table_header_row(tables) {
            let second = read_second_header(second, reader)?;
            return Ok(
                (second == &column_header_row(tables)).then(|| ImportHeader {
                    table_headers: first.to_vec(),
                    columns: second.clone(),
                }),
            );
        }
        return Ok((first == column_header_row(tables)).then(|| ImportHeader {
            table_headers: Vec::new(),
            columns: first.to_vec(),
        }));
    }
    if first != table_header_row(tables) {
        return Ok(None);
    }
    let second = read_second_header(second, reader)?;
    Ok(
        (second == &column_header_row(tables)).then(|| ImportHeader {
            table_headers: first.to_vec(),
            columns: second.clone(),
        }),
    )
}

async fn infer_multi_table_header<R: BufRead>(
    client: &mut GrpcClient,
    profile: &Profile,
    first: &[String],
    second: &mut Option<Vec<String>>,
    reader: &mut CsvRecordReader<R>,
) -> Result<Option<(Vec<ImportTable>, ImportHeader)>> {
    let table_names = consecutive_table_names(first);
    if table_names.is_empty() || !profile_has_tables(profile, &table_names) {
        return Ok(None);
    }
    let tables = load_import_tables_for_profile(&profile.name, client, &table_names).await?;
    if first != table_header_row(&tables) {
        return Ok(None);
    }
    let second = read_second_header(second, reader)?;
    if second != &column_header_row(&tables) {
        return Ok(None);
    }
    Ok(Some((
        tables,
        ImportHeader {
            table_headers: first.to_vec(),
            columns: second.clone(),
        },
    )))
}

pub fn row_to_table_data(
    table: &ImportTable,
    header: &ImportHeader,
    row: &[String],
) -> Result<TableDataImportRow> {
    let mut data = HashMap::new();
    for (index, target_column) in header.columns.iter().enumerate() {
        if header
            .table_headers
            .get(index)
            .filter(|name| !name.is_empty())
            .is_some_and(|name| name != &table.name)
        {
            continue;
        }
        if !table.columns.contains(target_column) {
            continue;
        }
        let Some(data_type) = table.type_map.get(target_column) else {
            continue;
        };
        let raw = row.get(index).map(String::as_str).unwrap_or("");
        let value = if table.link_columns.contains(target_column) && !raw.is_empty() {
            Value {
                kind: Some(Kind::StringValue(raw.to_string())),
            }
        } else {
            convert_input_value(raw, data_type, target_column).map_err(anyhow::Error::msg)?
        };
        data.insert(target_column.clone(), value);
    }
    Ok(TableDataImportRow { data, link_display_columns: Vec::new() })
}

pub fn exportable_columns(schema: &TableStructureResponse) -> Vec<String> {
    schema
        .columns
        .iter()
        .filter(|column| {
            !column.is_primary_key
                && column.name != "id"
                && column.name != "deleted"
                && column.name != "created_at"
                && column.name != "row_revision"
        })
        .map(|column| column.name.clone())
        .collect()
}