//! Importers that normalize public MCP datasets into the local corpus.
//!
//! Three offline source shapes are supported: an MCPCorpus JSON export, an
//! official-registry server export, and a curated repo list (JSON array or a
//! newline-delimited list of repo URLs). Each importer is deterministic and
//! pure: bytes in, normalized entries out, no network.

use serde_json::Value;
use sha2::{Digest, Sha256};

use super::model::{
    merge_entries, EcosystemCorpus, EcosystemEntry, Maintenance, ProjectRole, Provenance,
    SourceRecord, TransportKind, CORPUS_SCHEMA_VERSION,
};

/// Which dataset shape an input file holds.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SourceKind {
    /// An MCPCorpus-style JSON array of server/client records.
    McpCorpus,
    /// An official MCP registry server export (`{"servers": [...]}` or array).
    Registry,
    /// A curated repo list: JSON array of objects/strings, or newline list.
    Repos,
}

impl SourceKind {
    /// The short label stamped into provenance and the source record.
    pub fn label(self) -> &'static str {
        match self {
            SourceKind::McpCorpus => "mcpcorpus",
            SourceKind::Registry => "registry",
            SourceKind::Repos => "repos",
        }
    }
}

/// One source file to import: its kind, a reference label, and raw bytes.
pub struct ImportInput {
    /// The dataset shape.
    pub kind: SourceKind,
    /// A label for provenance (a path, usually).
    pub source_ref: String,
    /// The raw file bytes.
    pub bytes: Vec<u8>,
}

/// Build a corpus from a set of source inputs.
///
/// Entries are normalized, merged (dedup with enrichment), sorted, and hashed.
/// A parse error in one source is fatal: a half-imported corpus is worse than a
/// clear failure. Per-entry problems (a record missing a name and URL) are
/// recorded in `skipped`, not fatal.
pub fn build_corpus(inputs: &[ImportInput]) -> Result<EcosystemCorpus, ImportError> {
    let mut raw_entries: Vec<EcosystemEntry> = Vec::new();
    let mut sources: Vec<SourceRecord> = Vec::new();
    let mut skipped: Vec<String> = Vec::new();

    for input in inputs {
        let text = std::str::from_utf8(&input.bytes)
            .map_err(|_| ImportError::NotUtf8(input.source_ref.clone()))?;
        let (entries, dropped) = match input.kind {
            SourceKind::McpCorpus => parse_mcpcorpus(text, &input.source_ref)?,
            SourceKind::Registry => parse_registry(text, &input.source_ref)?,
            SourceKind::Repos => parse_repos(text, &input.source_ref)?,
        };
        sources.push(SourceRecord {
            kind: input.kind.label().to_string(),
            source_ref: input.source_ref.clone(),
            content_sha256: hex_sha256(&input.bytes),
            entry_count: entries.len(),
        });
        raw_entries.extend(entries);
        skipped.extend(dropped);
    }

    let (entries, merge_skipped) = merge_entries(raw_entries);
    skipped.extend(merge_skipped);
    let content_sha256 = entries_digest(&entries);
    Ok(EcosystemCorpus {
        schema_version: CORPUS_SCHEMA_VERSION.to_string(),
        sources,
        entries,
        skipped,
        content_sha256,
    })
}

/// The successfully parsed entries plus a skip reason for each dropped record.
type Parsed = (Vec<EcosystemEntry>, Vec<String>);

/// Run a per-record mapper over an array, collecting entries and skip reasons.
fn map_records(
    records: &[Value],
    source_ref: &str,
    map: impl Fn(&Value) -> Option<EcosystemEntry>,
) -> Parsed {
    let mut entries = Vec::new();
    let mut skipped = Vec::new();
    for (idx, record) in records.iter().enumerate() {
        match map(record) {
            Some(entry) => entries.push(entry),
            None => skipped.push(format!("{source_ref}[{idx}]: no usable name or source url")),
        }
    }
    (entries, skipped)
}

/// Parse an MCPCorpus-style JSON array.
fn parse_mcpcorpus(text: &str, source_ref: &str) -> Result<Parsed, ImportError> {
    let value: Value =
        serde_json::from_str(text).map_err(|e| ImportError::Parse(source_ref.into(), e))?;
    let records = as_record_array(&value, source_ref)?;
    let provenance = Provenance {
        source_kind: SourceKind::McpCorpus.label().to_string(),
        source_ref: source_ref.to_string(),
    };
    Ok(map_records(records, source_ref, |r| {
        mcpcorpus_entry(r, &provenance)
    }))
}

/// Map one MCPCorpus record onto an entry. Returns `None` for an empty record.
fn mcpcorpus_entry(record: &Value, provenance: &Provenance) -> Option<EcosystemEntry> {
    let name = str_field(record, &["name", "repo_name", "full_name"]).unwrap_or_default();
    let source_url =
        str_field(record, &["url", "repo", "repo_url", "html_url", "homepage"]).unwrap_or_default();
    if name.is_empty() && source_url.is_empty() {
        return None;
    }
    let role = match str_field(record, &["type", "role", "category_type"]).as_deref() {
        Some(t) if t.to_ascii_lowercase().contains("client") => ProjectRole::Client,
        Some(t) if t.to_ascii_lowercase().contains("server") => ProjectRole::Server,
        _ => ProjectRole::Server,
    };
    let transport = str_field(record, &["transport", "transports"])
        .map(|t| TransportKind::from_label(&t))
        .unwrap_or_default();
    Some(EcosystemEntry {
        name: if name.is_empty() {
            source_url.clone()
        } else {
            name
        },
        source_url,
        registry_id: None,
        role,
        transport,
        language: str_field(record, &["language", "lang"]).map(|s| s.to_ascii_lowercase()),
        auth_mode: str_field(record, &["auth", "auth_mode"]),
        tool_count: u32_field(record, &["tool_count", "tools", "num_tools"]),
        resource_count: u32_field(record, &["resource_count", "resources"]),
        prompt_count: u32_field(record, &["prompt_count", "prompts"]),
        license: str_field(record, &["license", "spdx", "license_id"]),
        maintenance: maintenance_from(record),
        security_caveats: str_array_field(record, &["security_caveats", "warnings"]),
        categories: str_array_field(record, &["categories", "tags", "topics"]),
        provenance: provenance.clone(),
    })
}

/// Parse an official-registry export: `{"servers": [...]}` or a bare array.
fn parse_registry(text: &str, source_ref: &str) -> Result<Parsed, ImportError> {
    let value: Value =
        serde_json::from_str(text).map_err(|e| ImportError::Parse(source_ref.into(), e))?;
    let records = match value.get("servers") {
        Some(servers) => as_record_array(servers, source_ref)?,
        None => as_record_array(&value, source_ref)?,
    };
    let provenance = Provenance {
        source_kind: SourceKind::Registry.label().to_string(),
        source_ref: source_ref.to_string(),
    };
    Ok(map_records(records, source_ref, |r| {
        registry_entry(r, &provenance)
    }))
}

/// Map one registry server record onto an entry.
fn registry_entry(record: &Value, provenance: &Provenance) -> Option<EcosystemEntry> {
    let name = str_field(record, &["name"]).unwrap_or_default();
    let source_url = record
        .get("repository")
        .and_then(|r| str_field(r, &["url", "source"]))
        .or_else(|| str_field(record, &["repository", "homepage"]))
        .unwrap_or_default();
    if name.is_empty() && source_url.is_empty() {
        return None;
    }
    // Registry transport lives under packages[].transport or remotes[].type.
    let transport = registry_transport(record);
    Some(EcosystemEntry {
        name: if name.is_empty() {
            source_url.clone()
        } else {
            name
        },
        source_url,
        registry_id: str_field(record, &["id", "registry_id"]),
        role: ProjectRole::Server,
        transport,
        language: None,
        auth_mode: None,
        tool_count: None,
        resource_count: None,
        prompt_count: None,
        license: str_field(record, &["license"]),
        maintenance: Maintenance::Unknown,
        security_caveats: Vec::new(),
        categories: str_array_field(record, &["categories", "tags"]),
        provenance: provenance.clone(),
    })
}

/// Resolve a registry record's transport from `remotes` or `packages`.
fn registry_transport(record: &Value) -> TransportKind {
    if let Some(remotes) = record.get("remotes").and_then(Value::as_array) {
        if let Some(first) = remotes.first() {
            if let Some(t) = str_field(first, &["type", "transport"]) {
                return TransportKind::from_label(&t);
            }
        }
    }
    if record.get("packages").and_then(Value::as_array).is_some() {
        return TransportKind::Stdio;
    }
    TransportKind::Unknown
}

/// Parse a curated repo list: JSON array (objects or strings) or newline list.
fn parse_repos(text: &str, source_ref: &str) -> Result<Parsed, ImportError> {
    let provenance = Provenance {
        source_kind: SourceKind::Repos.label().to_string(),
        source_ref: source_ref.to_string(),
    };
    let trimmed = text.trim_start();
    if trimmed.starts_with('[') {
        let value: Value =
            serde_json::from_str(text).map_err(|e| ImportError::Parse(source_ref.into(), e))?;
        let records = as_record_array(&value, source_ref)?;
        return Ok(map_records(records, source_ref, |r| {
            repo_record_entry(r, &provenance)
        }));
    }
    // Newline list: one repo URL (or `# comment`) per line.
    let entries = text
        .lines()
        .map(str::trim)
        .filter(|l| !l.is_empty() && !l.starts_with('#'))
        .map(|url| repo_url_entry(url, &provenance))
        .collect();
    Ok((entries, Vec::new()))
}

/// Map a JSON repo-list record (object or bare string) onto an entry.
fn repo_record_entry(record: &Value, provenance: &Provenance) -> Option<EcosystemEntry> {
    if let Some(url) = record.as_str() {
        return Some(repo_url_entry(url, provenance));
    }
    mcpcorpus_entry(record, provenance).map(|mut e| {
        e.provenance = provenance.clone();
        e
    })
}

/// Build a minimal entry from a bare repo URL, naming it after the repo path.
fn repo_url_entry(url: &str, provenance: &Provenance) -> EcosystemEntry {
    let name = url
        .trim_end_matches('/')
        .rsplit('/')
        .find(|s| !s.is_empty())
        .unwrap_or(url)
        .to_string();
    EcosystemEntry {
        name,
        source_url: url.to_string(),
        registry_id: None,
        role: ProjectRole::Unknown,
        transport: TransportKind::Unknown,
        language: None,
        auth_mode: None,
        tool_count: None,
        resource_count: None,
        prompt_count: None,
        license: None,
        maintenance: Maintenance::Unknown,
        security_caveats: Vec::new(),
        categories: Vec::new(),
        provenance: provenance.clone(),
    }
}

/// Derive a maintenance signal from whatever recency flag the record carries.
fn maintenance_from(record: &Value) -> Maintenance {
    if record
        .get("archived")
        .and_then(Value::as_bool)
        .unwrap_or(false)
    {
        return Maintenance::Archived;
    }
    match str_field(record, &["maintenance", "activity", "status"]).map(|s| s.to_ascii_lowercase())
    {
        Some(s) if s.contains("archiv") => Maintenance::Archived,
        Some(s) if s.contains("active") || s.contains("maintained") => Maintenance::Active,
        Some(s) if s.contains("dorm") || s.contains("stale") || s.contains("inactive") => {
            Maintenance::Dormant
        }
        _ => Maintenance::Unknown,
    }
}

/// Require a JSON value to be an array and return its elements.
fn as_record_array<'a>(value: &'a Value, source_ref: &str) -> Result<&'a Vec<Value>, ImportError> {
    value
        .as_array()
        .ok_or_else(|| ImportError::NotArray(source_ref.to_string()))
}

/// First present, non-empty string among candidate keys.
fn str_field(record: &Value, keys: &[&str]) -> Option<String> {
    for key in keys {
        if let Some(s) = record.get(key).and_then(Value::as_str) {
            let t = s.trim();
            if !t.is_empty() {
                return Some(t.to_string());
            }
        }
    }
    None
}

/// First present unsigned integer among candidate keys (numbers or arrays).
fn u32_field(record: &Value, keys: &[&str]) -> Option<u32> {
    for key in keys {
        match record.get(key) {
            Some(Value::Number(n)) => {
                if let Some(u) = n.as_u64() {
                    return Some(u as u32);
                }
            }
            Some(Value::Array(a)) => return Some(a.len() as u32),
            _ => {}
        }
    }
    None
}

/// Collected, trimmed strings from the first present array-of-strings key.
fn str_array_field(record: &Value, keys: &[&str]) -> Vec<String> {
    for key in keys {
        if let Some(arr) = record.get(key).and_then(Value::as_array) {
            return arr
                .iter()
                .filter_map(Value::as_str)
                .map(|s| s.trim().to_string())
                .filter(|s| !s.is_empty())
                .collect();
        }
    }
    Vec::new()
}

/// SHA-256 of raw bytes, hex-encoded with a `sha256:` prefix.
fn hex_sha256(bytes: &[u8]) -> String {
    let mut hasher = Sha256::new();
    hasher.update(bytes);
    format!("sha256:{:x}", hasher.finalize())
}

/// Stable content hash over the canonical JSON of the entry list.
pub(crate) fn entries_digest(entries: &[EcosystemEntry]) -> String {
    let bytes = serde_json::to_vec(entries).unwrap_or_default();
    hex_sha256(&bytes)
}

/// Errors raised while importing a source file.
#[derive(Debug, thiserror::Error)]
pub enum ImportError {
    /// A source file was not valid UTF-8.
    #[error("source {0} is not valid UTF-8")]
    NotUtf8(String),
    /// A source file did not parse as JSON.
    #[error("failed to parse {0} as JSON: {1}")]
    Parse(String, #[source] serde_json::Error),
    /// A source file was valid JSON but not the expected array shape.
    #[error("source {0} is not a JSON array of records")]
    NotArray(String),
}
