use std::env;
use std::fs;
use std::path::{Path, PathBuf};

use rho_core::{RhoResult, arg_value, from_yaml, has_flag, normalize_actor_id, yaml_quote};
use serde::Deserialize;

fn usage() -> ! {
    eprintln!(
        "usage:\n  rho dataset bind <name> --real <path> [--uuid <uuid>] [--owner <owner>] [--bindings-root <path>]\n  rho dataset list [--root <repo>] [--bindings-root <path>] [--repo-only|--bindings-only]\n  rho dataset remove <name|uuid> (--repo|--binding) [--root <repo>] [--bindings-root <path>] [--yes]"
    );
    std::process::exit(2);
}

pub fn bind(args: &[String]) -> RhoResult<()> {
    let Some(name) = args.first().filter(|value| !value.starts_with('-')) else {
        usage();
    };
    let dataset_slug = dataset_path_slug(name)?;
    let source = arg_value(args, "--real")
        .or_else(|| arg_value(args, "--local"))
        .ok_or("missing --real <path>")?;
    let source_path = PathBuf::from(&source).canonicalize().map_err(|_| {
        format!(
            "private dataset path not found: {}",
            PathBuf::from(&source).display()
        )
    })?;
    let owner_arg = arg_value(args, "--owner")
        .or_else(active_handle)
        .unwrap_or_else(|| "user1".to_string());
    let owner_id = normalize_actor_id(&owner_arg)?;
    let uuid = arg_value(args, "--uuid")
        .or_else(|| {
            arg_value(args, "--root")
                .map(PathBuf::from)
                .and_then(|root| repo_dataset_uuid(&root, name).ok().flatten())
        })
        .unwrap_or_default();
    let bindings_root = bindings_root(args)?;
    let binding_dir = bindings_root.join(&dataset_slug);
    let binding_path = binding_dir.join("bindings.yaml");
    fs::create_dir_all(&binding_dir)?;
    fs::write(
        &binding_path,
        format!(
            concat!(
                "version: 1\n",
                "dataset_binding:\n",
                "  name: {}\n",
                "  uuid: {}\n",
                "  owner: {}\n",
                "  variants:\n",
                "    real:\n",
                "      source:\n",
                "        kind: \"local_path\"\n",
                "        path: {}\n",
                "      materialization:\n",
                "        mode: \"existing\"\n",
            ),
            yaml_quote(name),
            yaml_quote(&uuid),
            yaml_quote(&owner_id),
            yaml_quote(&source_path.display().to_string()),
        ),
    )?;
    println!("bound dataset: {name}");
    if !uuid.is_empty() {
        println!("uuid: {uuid}");
    }
    println!("real source: {}", source_path.display());
    println!("binding: {}", binding_path.display());
    Ok(())
}

pub fn list(args: &[String]) -> RhoResult<()> {
    let repo_only = has_flag(args, "--repo-only");
    let bindings_only = has_flag(args, "--bindings-only");
    if repo_only && bindings_only {
        return Err("choose only one of --repo-only or --bindings-only".into());
    }
    if !bindings_only {
        let root = arg_value(args, "--root")
            .map(PathBuf::from)
            .unwrap_or_else(|| PathBuf::from("."));
        println!("repo datasets:");
        for dataset in repo_datasets(&root)? {
            println!(
                "  {}  uuid={}  variants={}  path={}",
                dataset.name,
                empty_dash(&dataset.uuid),
                dataset.variants.join(","),
                dataset.path.display()
            );
        }
    }
    if !repo_only {
        println!("local bindings:");
        for binding in local_bindings(&bindings_root(args)?)? {
            println!(
                "  {}  uuid={}  variants={}  path={}",
                binding.name,
                empty_dash(&binding.uuid),
                binding.variants.join(","),
                binding.path.display()
            );
        }
    }
    Ok(())
}

pub fn remove(args: &[String]) -> RhoResult<()> {
    let Some(reference) = args.first().filter(|value| !value.starts_with('-')) else {
        usage();
    };
    let remove_repo = has_flag(args, "--repo");
    let remove_binding = has_flag(args, "--binding");
    if remove_repo == remove_binding {
        return Err("choose exactly one of --repo or --binding".into());
    }
    if !has_flag(args, "--yes") {
        return Err("refusing to remove without --yes".into());
    }
    if remove_repo {
        let root = arg_value(args, "--root")
            .map(PathBuf::from)
            .unwrap_or_else(|| PathBuf::from("."));
        let dataset = find_repo_dataset(&root, reference)?
            .ok_or_else(|| format!("repo dataset not found: {reference}"))?;
        fs::remove_dir_all(&dataset.dir)?;
        println!("removed repo dataset: {}", dataset.dir.display());
    } else {
        let binding = find_local_binding(&bindings_root(args)?, reference)?
            .ok_or_else(|| format!("local binding not found: {reference}"))?;
        fs::remove_dir_all(&binding.dir)?;
        println!("removed local binding: {}", binding.dir.display());
    }
    Ok(())
}

fn bindings_root(args: &[String]) -> RhoResult<PathBuf> {
    if let Some(root) = arg_value(args, "--bindings-root") {
        return Ok(PathBuf::from(root));
    }
    let handle = active_handle().ok_or("missing --profile or --bindings-root")?;
    Ok(profile_root(&handle).join("datasets"))
}

fn profile_root(handle: &str) -> PathBuf {
    env::var("RHO_PROJECTS_ROOT")
        .map(PathBuf::from)
        .unwrap_or_else(|_| {
            env::var("HOME")
                .map(PathBuf::from)
                .unwrap_or_else(|_| PathBuf::from("."))
                .join("rho")
        })
        .join(handle)
}

fn active_handle() -> Option<String> {
    env::var("RHO_ENV_HANDLE")
        .ok()
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
}

fn repo_dataset_uuid(root: &Path, reference: &str) -> RhoResult<Option<String>> {
    Ok(find_repo_dataset(root, reference)?.map(|dataset| dataset.uuid))
}

fn repo_datasets(root: &Path) -> RhoResult<Vec<ListEntry>> {
    let datasets = root.join("datasets");
    let mut entries = Vec::new();
    if !datasets.is_dir() {
        return Ok(entries);
    }
    for entry in fs::read_dir(&datasets)? {
        let dir = entry?.path();
        let manifest_path = dir.join("dataset.yaml");
        if !manifest_path.is_file() {
            continue;
        }
        let manifest = read_repo_dataset(&manifest_path)?;
        entries.push(ListEntry {
            name: manifest.dataset.name,
            uuid: manifest.dataset.uuid,
            variants: manifest.dataset.variants.names(),
            path: manifest_path,
        });
    }
    entries.sort_by(|a, b| a.name.cmp(&b.name));
    Ok(entries)
}

fn find_repo_dataset(root: &Path, reference: &str) -> RhoResult<Option<RepoDatasetMatch>> {
    let datasets = root.join("datasets");
    if !datasets.is_dir() {
        return Ok(None);
    }
    let mut matches = Vec::new();
    for entry in fs::read_dir(&datasets)? {
        let dir = entry?.path();
        let manifest_path = dir.join("dataset.yaml");
        if !manifest_path.is_file() {
            continue;
        }
        let manifest = read_repo_dataset(&manifest_path)?;
        if manifest.dataset.name == reference || manifest.dataset.uuid == reference {
            matches.push(RepoDatasetMatch {
                uuid: manifest.dataset.uuid,
                dir,
            });
        }
    }
    match matches.len() {
        0 => Ok(None),
        1 => Ok(matches.pop()),
        _ => Err(format!("repo dataset reference is ambiguous: {reference}").into()),
    }
}

fn read_repo_dataset(path: &Path) -> RhoResult<RepoDatasetManifest> {
    let text = fs::read_to_string(path)?;
    from_yaml(&text)
}

fn local_bindings(root: &Path) -> RhoResult<Vec<ListEntry>> {
    let mut entries = Vec::new();
    if !root.is_dir() {
        return Ok(entries);
    }
    for entry in fs::read_dir(root)? {
        let dir = entry?.path();
        let binding_path = dir.join("bindings.yaml");
        if !binding_path.is_file() {
            continue;
        }
        let binding = read_binding(&binding_path)?;
        entries.push(ListEntry {
            name: binding.dataset_binding.name,
            uuid: binding.dataset_binding.uuid.unwrap_or_default(),
            variants: binding.dataset_binding.variants.names(),
            path: binding_path,
        });
    }
    entries.sort_by(|a, b| a.name.cmp(&b.name));
    Ok(entries)
}

fn find_local_binding(root: &Path, reference: &str) -> RhoResult<Option<BindingMatch>> {
    let mut matches = Vec::new();
    for entry in local_bindings(root)? {
        if entry.name == reference || entry.uuid == reference {
            let dir = entry
                .path
                .parent()
                .ok_or("binding path has no parent")?
                .to_path_buf();
            matches.push(BindingMatch { dir });
        }
    }
    match matches.len() {
        0 => Ok(None),
        1 => Ok(matches.pop()),
        _ => Err(format!("local binding reference is ambiguous: {reference}").into()),
    }
}

fn read_binding(path: &Path) -> RhoResult<BindingManifest> {
    let text = fs::read_to_string(path)?;
    from_yaml(&text)
}

fn dataset_path_slug(value: &str) -> RhoResult<String> {
    if value.is_empty()
        || !value
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '.')
    {
        return Err(format!("dataset name is not path-safe: {value}").into());
    }
    Ok(value.to_string())
}

fn empty_dash(value: &str) -> &str {
    if value.is_empty() { "-" } else { value }
}

struct ListEntry {
    name: String,
    uuid: String,
    variants: Vec<&'static str>,
    path: PathBuf,
}

struct RepoDatasetMatch {
    uuid: String,
    dir: PathBuf,
}

struct BindingMatch {
    dir: PathBuf,
}

#[derive(Debug, Deserialize)]
struct RepoDatasetManifest {
    dataset: RepoDatasetRecord,
}

#[derive(Debug, Deserialize)]
struct RepoDatasetRecord {
    uuid: String,
    name: String,
    variants: RepoDatasetVariants,
}

#[derive(Debug, Deserialize)]
struct RepoDatasetVariants {
    #[serde(default)]
    public: Option<serde_yaml::Value>,
    #[serde(default)]
    mock: Option<serde_yaml::Value>,
    #[serde(default)]
    real: Option<serde_yaml::Value>,
}

impl RepoDatasetVariants {
    fn names(&self) -> Vec<&'static str> {
        let mut names = Vec::new();
        if self.public.is_some() {
            names.push("public");
        }
        if self.mock.is_some() {
            names.push("mock");
        }
        if self.real.is_some() {
            names.push("real");
        }
        names
    }
}

#[derive(Debug, Deserialize)]
struct BindingManifest {
    dataset_binding: BindingRecord,
}

#[derive(Debug, Deserialize)]
struct BindingRecord {
    name: String,
    #[serde(default)]
    uuid: Option<String>,
    variants: BindingVariants,
}

#[derive(Debug, Deserialize)]
struct BindingVariants {
    #[serde(default)]
    real: Option<serde_yaml::Value>,
}

impl BindingVariants {
    fn names(&self) -> Vec<&'static str> {
        if self.real.is_some() {
            vec!["real"]
        } else {
            Vec::new()
        }
    }
}

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

    #[test]
    fn profile_dataset_root_uses_profile_home() {
        let root = profile_root("madhavajay").join("datasets");
        assert!(root.ends_with("madhavajay/datasets"));
    }
}
