genson-core 0.8.0

use crate::genson_rs::{build_json_schema, build_json_schema_into, get_builder, BuildConfig};
use crate::{debug, profile, profile_verbose};
use rayon::prelude::*;
use serde::de::Error as DeError;
use serde::Deserialize;
use serde_json::{json, Value};
use std::borrow::Cow;
use std::collections::HashSet;
use std::panic::{self, AssertUnwindSafe};
use std::time::{SystemTime, UNIX_EPOCH};

use crate::genson_rs::SchemaBuilder;

pub(crate) mod core;
pub use core::*;
mod map_inference;
use map_inference::*;

/// Maximum length of JSON string to include in error messages before truncating
const MAX_JSON_ERROR_LENGTH: usize = 100;
/// Threshold for switching to parallel processing. Below this, use sequential.
const PARALLEL_THRESHOLD: usize = 10;

/// Get current RSS memory usage in bytes
pub(crate) fn get_rss_bytes() -> Option<usize> {
    let status = std::fs::read_to_string("/proc/self/status").ok()?;
    for line in status.lines() {
        if line.starts_with("VmRSS:") {
            let parts: Vec<&str> = line.split_whitespace().collect();
            if parts.len() >= 2 {
                let kb: usize = parts[1].parse().ok()?;
                return Some(kb * 1024); // Convert KB to bytes
            }
        }
    }
    None
}

/// Format bytes to human-readable string
pub(crate) fn format_bytes(bytes: usize) -> String {
    const MB: usize = 1024 * 1024;
    const GB: usize = 1024 * 1024 * 1024;

    if bytes >= GB {
        format!("{:.2} GB", bytes as f64 / GB as f64)
    } else if bytes >= MB {
        format!("{:.2} MB", bytes as f64 / MB as f64)
    } else {
        format!("{:.2} KB", bytes as f64 / 1024.0)
    }
}

fn current_time_hms() -> String {
    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();

    let total_seconds = now.as_secs() % 86_400; // seconds in a day
    let hours = total_seconds / 3600;
    let minutes = (total_seconds % 3600) / 60;
    let seconds = (total_seconds % 60) as f64 + now.subsec_millis() as f64 / 1000.0; // add fractional part

    format!("{:02}:{:02}:{:04.1}", hours, minutes, seconds)
}

fn validate_json(s: &str) -> Result<(), serde_json::Error> {
    let mut de = serde_json::Deserializer::from_str(s);
    serde::de::IgnoredAny::deserialize(&mut de)?; // lightweight: ignores the parsed value
    de.end()
}

fn validate_ndjson(s: &str) -> Result<(), serde_json::Error> {
    for line in s.lines() {
        let trimmed = line.trim();
        if trimmed.is_empty() {
            continue;
        }
        validate_json(trimmed)?; // propagate serde_json::Error
    }
    Ok(())
}

/// Recursively reorder union type arrays in a JSON Schema by canonical precedence.
///
/// Special case: preserves the common `["null", T]` pattern without reordering.
pub fn reorder_unions(schema: &mut Value) {
    match schema {
        Value::Object(obj) => {
            if let Some(Value::Array(types)) = obj.get_mut("type") {
                // sort by canonical precedence, but keep ["null", T] pattern intact
                if !(types.len() == 2 && types.iter().any(|t| t == "null")) {
                    types.sort_by_key(type_rank);
                }
            }
            // recurse into properties/items/etc.
            for v in obj.values_mut() {
                reorder_unions(v);
            }
        }
        Value::Array(arr) => {
            // Parallelize across array elements (if large enough)
            if arr.len() >= PARALLEL_THRESHOLD {
                arr.par_iter_mut().for_each(reorder_unions);
            } else {
                for v in arr {
                    reorder_unions(v);
                }
            }
        }
        _ => {}
    }
}

/// Assign a numeric precedence rank to a JSON Schema type.
///
/// Used by `reorder_unions` to sort union members deterministically.
/// - Null always first
/// - Containers before scalars (to enforce widening)
/// - Scalars ordered by narrowness
/// - Unknown types last
pub fn type_rank(val: &Value) -> usize {
    match val {
        Value::String(s) => type_string_rank(s),
        Value::Object(obj) => {
            if let Some(Value::String(t)) = obj.get("type") {
                type_string_rank(t)
            } else {
                100 // object with no "type" field
            }
        }
        _ => 100, // non-string/non-object
    }
}

/// Internal helper: rank by type string
fn type_string_rank(s: &str) -> usize {
    match s {
        // Null always first
        "null" => 0,

        // Containers before scalars: widening takes precedence
        "map" => 1,
        "array" => 2,
        "object" | "record" => 3,

        // Scalars (ordered by 'narrowness')
        "boolean" => 10,
        "integer" | "int" | "long" => 11,
        "number" | "float" | "double" => 12,
        "enum" => 13,
        "string" => 14,
        "fixed" => 15,
        "bytes" => 16,

        // Fallback
        _ => 99,
    }
}

/// Prepare JSON bytes for schema building (validation + wrap_root transformation)
fn prepare_json_bytes<'a>(
    json_bytes: &'a [u8],
    index: usize,
    config: &SchemaInferenceConfig,
) -> Result<Cow<'a, [u8]>, String> {
    // Early return for empty input
    let json_str = std::str::from_utf8(json_bytes)
        .map_err(|e| format!("Invalid UTF-8 at index {}: {}", index + 1, e))?;

    if json_str.trim().is_empty() {
        return Ok(Cow::Borrowed(&[]));
    }

    // Choose validation strategy based on delimiter
    let validation_result = if let Some(delim) = config.delimiter {
        if delim == b'\n' {
            validate_ndjson(json_str)
        } else {
            Err(serde_json::Error::custom(format!(
                "Unsupported delimiter: {:?}",
                delim
            )))
        }
    } else {
        validate_json(json_str)
    };

    if let Err(parse_error) = validation_result {
        let truncated_json = if json_str.len() > MAX_JSON_ERROR_LENGTH {
            format!(
                "{}... [truncated {} chars]",
                &json_str[..MAX_JSON_ERROR_LENGTH],
                json_str.len() - MAX_JSON_ERROR_LENGTH
            )
        } else {
            json_str.to_string()
        };

        return Err(format!(
            "Invalid JSON input at index {}: {} - JSON: {}",
            index + 1,
            parse_error,
            truncated_json
        ));
    }

    // Safe: JSON is valid, now hand off to genson-rs
    if let Some(ref field) = config.wrap_root {
        if config.delimiter == Some(b'\n') {
            // NDJSON: wrap each line separately
            let mut wrapped_bytes = Vec::new();
            for line in json_str.lines() {
                let trimmed = line.trim();
                if trimmed.is_empty() {
                    continue;
                }
                let inner_val: Value = serde_json::from_str(trimmed)
                    .map_err(|e| format!("Failed to parse NDJSON line before wrap_root: {}", e))?;

                if !wrapped_bytes.is_empty() {
                    wrapped_bytes.push(b'\n');
                }
                serde_json::to_writer(&mut wrapped_bytes, &json!({ field: inner_val }))
                    .map_err(|e| format!("Failed to serialize wrapped NDJSON: {}", e))?;
            }
            Ok(Cow::Owned(wrapped_bytes))
        } else {
            // Single JSON doc
            let inner_val: Value = serde_json::from_str(json_str)
                .map_err(|e| format!("Failed to parse JSON before wrap_root: {}", e))?;
            let wrapped_bytes = serde_json::to_vec(&json!({ field: inner_val }))
                .map_err(|e| format!("Failed to serialize wrapped JSON: {}", e))?;
            Ok(Cow::Owned(wrapped_bytes))
        }
    } else {
        // No wrapping needed - just borrow the original bytes
        Ok(Cow::Borrowed(json_bytes))
    }
}

/// Process all JSON strings sequentially and build schemas
fn process_json_strings_sequential<S: AsRef<str>>(
    json_strings: &[S],
    config: &SchemaInferenceConfig,
    builder: &mut crate::genson_rs::SchemaBuilder,
) -> Result<usize, String> {
    let build_config = BuildConfig {
        delimiter: config.delimiter,
        ignore_outer_array: config.ignore_outer_array,
    };

    let mut processed_count = 0;

    // Process each JSON string
    for (i, json_str) in json_strings.iter().enumerate() {
        profile_verbose!(config, "PROCESSING JSON STRING {}", i);

        let prep_start = std::time::Instant::now();
        let prepared_json = prepare_json_bytes(json_str.as_ref().as_bytes(), i, config)?;
        let prep_elapsed = prep_start.elapsed();
        profile_verbose!(config, "  Preparation took: {:?}", prep_elapsed);

        if prepared_json.is_empty() {
            continue;
        }

        let mut bytes = prepared_json.into_owned(); // Only allocate if Cow::Owned

        // Build schema incrementally - this is where panics happen
        let build_start = std::time::Instant::now();
        let _schema = build_json_schema(builder, &mut bytes, &build_config);
        let build_elapsed = build_start.elapsed();
        profile_verbose!(config, "  Schema building took: {:?}", build_elapsed);

        processed_count += 1;
    }

    Ok(processed_count)
}

/// Apply force_field_types to a schema before merging.
/// This ensures structural consistency across schemas that will be unified.
fn apply_force_field_types(schema: &mut Value, config: &SchemaInferenceConfig) {
    match schema {
        Value::Object(obj) => {
            // Check if this object has properties
            if let Some(props) = obj.get_mut("properties") {
                if let Some(props_obj) = props.as_object_mut() {
                    for (field_name, field_schema) in props_obj.iter_mut() {
                        // Apply force_field_types
                        if let Some(forced) = config.force_field_types.get(field_name) {
                            if forced == "map" {
                                if let Some(field_obj) = field_schema.as_object_mut() {
                                    // Convert to map schema
                                    field_obj.shift_remove("properties");
                                    field_obj.shift_remove("required");
                                    field_obj.insert("type".to_string(), json!("object"));
                                    field_obj.insert(
                                        "additionalProperties".to_string(),
                                        json!({"type": "string"}),
                                    );
                                }
                            }
                        }
                        // Recurse
                        apply_force_field_types(field_schema, config);
                    }
                }
            }
            // Also recurse into items, additionalProperties, anyOf, etc.
            if let Some(items) = obj.get_mut("items") {
                apply_force_field_types(items, config);
            }
            if let Some(additional) = obj.get_mut("additionalProperties") {
                apply_force_field_types(additional, config);
            }
            if let Some(Value::Array(any_of)) = obj.get_mut("anyOf") {
                for item in any_of {
                    apply_force_field_types(item, config);
                }
            }
        }
        Value::Array(arr) => {
            for item in arr {
                apply_force_field_types(item, config);
            }
        }
        _ => {}
    }
}

/// xxh64 of the schema's compact JSON, streamed so the string is never materialised.
fn hash_schema(schema: &Value) -> u64 {
    struct HashWriter(xxhash_rust::xxh64::Xxh64);
    impl std::io::Write for HashWriter {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.update(buf);
            Ok(buf.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }
    let mut writer = HashWriter(xxhash_rust::xxh64::Xxh64::new(0));
    serde_json::to_writer(&mut writer, schema).expect("hashing a Value cannot fail");
    writer.0.digest()
}

/// Process all JSON strings in parallel while maintaining order
fn process_json_strings_parallel<S: AsRef<str> + Sync>(
    json_strings: &[S],
    config: &SchemaInferenceConfig,
    builder: &mut SchemaBuilder,
) -> Result<usize, String> {
    profile!(
        config,
        "Starting parallel preparation and building ({})",
        current_time_hms()
    );

    // Process in chunks to limit peak memory
    let chunk_size = config.max_builders.unwrap_or(json_strings.len());

    if config.profile {
        if let Some(rss) = get_rss_bytes() {
            anstream::eprintln!("📊 RSS before parallel processing: {}", format_bytes(rss));
        }
    }

    let mut processed_count = 0;
    let mut seen_hashes = HashSet::new();

    for (chunk_idx, chunk) in json_strings.chunks(chunk_size).enumerate() {
        profile!(
            config,
            "Processing chunk {} ({} strings)",
            chunk_idx,
            chunk.len()
        );

        if config.profile {
            if let Some(rss) = get_rss_bytes() {
                anstream::eprintln!("📊 RSS before chunk {}: {}", chunk_idx, format_bytes(rss));
            }
        }

        let t_par = std::time::Instant::now();
        let chunk_builders: Vec<(usize, Option<(Value, u64)>)> = chunk
            .par_iter()
            .enumerate()
            .map(
                |(i, json_str)| -> Result<(usize, Option<(Value, u64)>), String> {
                    profile_verbose!(config, "Thread processing JSON STRING {}", i);

                    let prep_start = std::time::Instant::now();
                    let prepared = prepare_json_bytes(json_str.as_ref().as_bytes(), i, config)?;
                    let prep_elapsed = prep_start.elapsed();
                    profile_verbose!(
                        config,
                        "  String {} preparation took: {:?}",
                        i,
                        prep_elapsed
                    );

                    if prepared.is_empty() {
                        return Ok((i, None));
                    }

                    let mut chunk_builder = get_builder(config.schema_uri.as_deref());
                    let mut bytes = prepared.into_owned();
                    let chunk_build_config = BuildConfig {
                        delimiter: config.delimiter,
                        ignore_outer_array: config.ignore_outer_array,
                    };

                    let build_start = std::time::Instant::now();
                    build_json_schema_into(&mut chunk_builder, &mut bytes, &chunk_build_config);
                    let build_elapsed = build_start.elapsed();
                    profile_verbose!(
                        config,
                        "  String {} schema building took: {:?}",
                        i,
                        build_elapsed
                    );

                    // Convert on the worker thread so the builder is dropped here
                    // instead of piling up until the serial merge below.
                    let mut schema = chunk_builder.to_schema();
                    drop(chunk_builder);
                    // The walk only ever rewrites forced fields
                    if !config.force_field_types.is_empty() {
                        apply_force_field_types(&mut schema, config);
                    }
                    let hash = hash_schema(&schema);
                    Ok((i, Some((schema, hash))))
                },
            )
            .collect::<Result<Vec<_>, String>>()?;

        if config.profile {
            if let Some(rss) = get_rss_bytes() {
                anstream::eprintln!("📊 RSS after collecting chunk: {}", format_bytes(rss));
            }
        }

        let par_el = t_par.elapsed();
        let t_mrg = std::time::Instant::now();
        // Extract and merge schemas from this chunk
        let mut merged: Vec<Value> = Vec::with_capacity(chunk_builders.len());
        for (_i, item) in chunk_builders {
            let Some((schema, hash)) = item else {
                continue;
            };
            if !seen_hashes.insert(hash) {
                continue;
            }
            processed_count += 1;
            merged.push(schema);
        }
        builder.add_schemas_mut(&mut merged);
        // Freeing the merged schemas is a large share of the serial merge, so spread it
        // over the rayon pool.
        merged.into_par_iter().for_each(drop);

        profile!(
            config,
            "chunk {} parallel {:?} serial-merge {:?}",
            chunk_idx,
            par_el,
            t_mrg.elapsed()
        );
        if config.profile {
            if let Some(rss) = get_rss_bytes() {
                anstream::eprintln!("📊 RSS after merging chunk: {}", format_bytes(rss));
            }
        }
    }

    profile!(config, "All chunks processed ({})", current_time_hms());

    Ok(processed_count)
}

/// Recursively convert fields matching force_field_types BEFORE unification runs.
/// This ensures structural consistency so that unification can succeed.
pub(crate) fn preprocess_force_field_types(schema: &mut Value, config: &SchemaInferenceConfig) {
    match schema {
        Value::Object(obj) => {
            // Process properties
            if let Some(props) = obj.get_mut("properties") {
                if let Some(props_obj) = props.as_object_mut() {
                    for (field_name, field_schema) in props_obj.iter_mut() {
                        // Check if this field should be forced to a type
                        if let Some(forced) = config.force_field_types.get(field_name) {
                            if forced == "map" {
                                convert_to_map(field_schema);
                            }
                        }
                        // Recurse into the field schema
                        preprocess_force_field_types(field_schema, config);
                    }
                }
            }
            // Recurse into items (for arrays)
            if let Some(items) = obj.get_mut("items") {
                preprocess_force_field_types(items, config);
            }
            // Recurse into additionalProperties (for maps)
            if let Some(additional) = obj.get_mut("additionalProperties") {
                preprocess_force_field_types(additional, config);
            }
            // Recurse into anyOf branches
            if let Some(Value::Array(any_of)) = obj.get_mut("anyOf") {
                for item in any_of {
                    preprocess_force_field_types(item, config);
                }
            }
        }
        Value::Array(arr) => {
            // Handle union types like ["null", {...}]
            for item in arr {
                preprocess_force_field_types(item, config);
            }
        }
        _ => {}
    }
}

/// Convert any schema to a Map<string, string> schema
fn convert_to_map(schema: &mut Value) {
    // Handle union types first: ["null", {...}] or [Record, "string"]
    if let Value::Array(arr) = schema {
        // Check if it's a nullable union
        let has_null = arr.iter().any(|v| {
            v == &Value::String("null".into())
                || v.get("type") == Some(&Value::String("null".into()))
        });
        if has_null {
            // Create nullable map
            *schema = serde_json::json!({
                "type": ["null", "object"],
                "additionalProperties": {"type": "string"}
            });
        } else {
            // Non-nullable map
            *schema = serde_json::json!({
                "type": "object",
                "additionalProperties": {"type": "string"}
            });
        }
        return;
    }

    if let Value::Object(obj) = schema {
        // Check if already a map
        if obj.contains_key("additionalProperties") {
            return;
        }

        // Check for nullable type: {"type": ["null", "object"]}
        let is_nullable = obj
            .get("type")
            .and_then(|t| t.as_array())
            .map(|arr| arr.contains(&Value::String("null".into())))
            .unwrap_or(false);

        // Convert to map
        obj.shift_remove("properties");
        obj.shift_remove("required");
        obj.shift_remove("anyOf");

        if is_nullable {
            obj.insert("type".to_string(), serde_json::json!(["null", "object"]));
        } else {
            obj.insert("type".to_string(), serde_json::json!("object"));
        }
        obj.insert(
            "additionalProperties".to_string(),
            serde_json::json!({"type": "string"}),
        );
    }
}

/// Infer JSON schema from a collection of JSON strings
pub fn infer_json_schema_from_strings<S: AsRef<str> + Sync>(
    json_strings: &[S],
    config: SchemaInferenceConfig,
) -> Result<SchemaInferenceResult, String> {
    profile!(
        config,
        "Processing {} strings ({})",
        json_strings.len(),
        current_time_hms()
    );
    debug!(config, "Schema inference config: {:#?}", config);
    if json_strings.is_empty() {
        return Err("No JSON strings provided".to_string());
    }

    // Wrap the entire genson-rs interaction in panic handling
    let result = panic::catch_unwind(AssertUnwindSafe(
        || -> Result<SchemaInferenceResult, String> {
            // Create schema builder
            let mut builder = get_builder(config.schema_uri.as_deref());

            profile!(config, "Starting preparation loop ({})", current_time_hms());

            let use_parallel = std::env::var("GENSON_PARALLEL")
                .map(|v| v == "1" || v.to_lowercase() == "true")
                .unwrap_or_else(|_| json_strings.len() >= PARALLEL_THRESHOLD);

            let processed_count = if use_parallel {
                process_json_strings_parallel(json_strings, &config, &mut builder)?
            } else {
                process_json_strings_sequential(json_strings, &config, &mut builder)?
            };

            // Get final schema
            let mut final_schema = builder.to_schema();
            profile!(
                config,
                "Applying force field types ({})",
                current_time_hms()
            );
            preprocess_force_field_types(&mut final_schema, &config);
            profile!(config, "Rewriting objects ({})", current_time_hms());
            rewrite_objects(&mut final_schema, None, &config, true);
            profile!(config, "Reordering unions ({})", current_time_hms());
            reorder_unions(&mut final_schema);

            #[cfg(feature = "avro")]
            if config.avro {
                let avro_schema = SchemaInferenceResult {
                    schema: final_schema.clone(),
                    processed_count,
                }
                .to_avro_schema(
                    "genson", // namespace
                    Some(""),
                    Some(""), // base_uri
                    false,    // don't split top-level
                );
                return Ok(SchemaInferenceResult {
                    schema: avro_schema,
                    processed_count,
                });
            }

            Ok(SchemaInferenceResult {
                schema: final_schema,
                processed_count,
            })
        },
    ));

    // Handle the result of panic::catch_unwind
    match result {
        Ok(Ok(schema_result)) => Ok(schema_result),
        Ok(Err(e)) => Err(e),
        Err(_panic) => Err("JSON schema inference failed due to invalid JSON input".to_string()),
    }
}

#[cfg(test)]
mod tests {
    include!("tests/schema.rs");
}