lodestone-proto 0.0.31

use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};

pub const MAX_SOURCE_PROJECTION_PATTERNS: usize = 128;
pub const MAX_SOURCE_PROJECTION_PATTERN_BYTES: usize = 512;

/// Endpoint-declared source representation needed after shard execution.
///
/// Parsed JSON remains available inside execution because result features,
/// diversification, fragments, and compatibility adapters may consume it.
/// Exact stored bytes are carried only for native consumers that promise a
/// byte-preserving full-source projection. Output filtering is applied after
/// those source-dependent controls have completed.
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SearchSourceProjection {
    Parsed,
    #[default]
    ParsedAndExact,
    /// Byte-exact source plus one mapping-native logical footprint per hit.
    /// This is an internal receiver/coordinator projection used only by the
    /// public committed-PIT logical occupancy command.
    ParsedExactAndLogicalOccupancy,
    Omit,
    Filtered {
        includes: Vec<String>,
        excludes: Vec<String>,
    },
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum SearchSourceProjectionError {
    TooManyPatterns,
    InvalidPattern,
    SourceTooDeep,
}

impl SearchSourceProjection {
    pub fn filtered(mut includes: Vec<String>, mut excludes: Vec<String>) -> Result<Self, SearchSourceProjectionError> {
        if includes.len().saturating_add(excludes.len()) > MAX_SOURCE_PROJECTION_PATTERNS {
            return Err(SearchSourceProjectionError::TooManyPatterns);
        }
        if includes.iter().chain(&excludes).any(|pattern| {
            pattern.is_empty()
                || pattern.len() > MAX_SOURCE_PROJECTION_PATTERN_BYTES
                || pattern.split('.').any(|segment| segment.is_empty())
                || pattern.chars().any(char::is_control)
        }) {
            return Err(SearchSourceProjectionError::InvalidPattern);
        }
        includes.sort();
        includes.dedup();
        excludes.sort();
        excludes.dedup();
        Ok(Self::Filtered { includes, excludes })
    }

    pub fn validate(&self) -> Result<(), SearchSourceProjectionError> {
        match self {
            Self::Parsed | Self::ParsedAndExact | Self::ParsedExactAndLogicalOccupancy | Self::Omit => Ok(()),
            Self::Filtered { includes, excludes } => {
                let canonical = Self::filtered(includes.clone(), excludes.clone())?;
                if canonical == *self {
                    Ok(())
                } else {
                    Err(SearchSourceProjectionError::InvalidPattern)
                }
            }
        }
    }

    #[must_use]
    pub const fn needs_exact_source(&self) -> bool {
        matches!(self, Self::ParsedAndExact | Self::ParsedExactAndLogicalOccupancy)
    }

    #[must_use]
    pub const fn needs_logical_occupancy(&self) -> bool {
        matches!(self, Self::ParsedExactAndLogicalOccupancy)
    }

    pub fn project(&self, source: &Value) -> Result<Option<Value>, SearchSourceProjectionError> {
        match self {
            Self::Parsed | Self::ParsedAndExact | Self::ParsedExactAndLogicalOccupancy => Ok(Some(source.clone())),
            Self::Omit => Ok(None),
            Self::Filtered { includes, excludes } => {
                let filtered = filter_value(source, &[], includes, excludes, false, 0)?;
                Ok(Some(filtered.unwrap_or_else(|| Value::Object(Map::new()))))
            }
        }
    }
}

fn filter_value(
    value: &Value,
    path: &[&str],
    includes: &[String],
    excludes: &[String],
    inherited_selection: bool,
    depth: usize,
) -> Result<Option<Value>, SearchSourceProjectionError> {
    if depth > 128 {
        return Err(SearchSourceProjectionError::SourceTooDeep);
    }
    if excludes.iter().any(|pattern| pattern_matches(pattern, path)) {
        return Ok(None);
    }
    let selected_here =
        inherited_selection || includes.is_empty() || includes.iter().any(|pattern| pattern_matches(pattern, path));
    match value {
        Value::Object(object) => {
            let mut filtered = Map::new();
            for (name, child) in object {
                let mut child_path = path.to_vec();
                child_path.push(name);
                if let Some(child) = filter_value(child, &child_path, includes, excludes, selected_here, depth + 1)? {
                    filtered.insert(name.clone(), child);
                }
            }
            if selected_here || !filtered.is_empty() {
                Ok(Some(Value::Object(filtered)))
            } else {
                Ok(None)
            }
        }
        Value::Array(values) => {
            let mut filtered = Vec::with_capacity(values.len());
            for child in values {
                if let Some(child) = filter_value(child, path, includes, excludes, selected_here, depth + 1)? {
                    filtered.push(child);
                }
            }
            if selected_here || !filtered.is_empty() {
                Ok(Some(Value::Array(filtered)))
            } else {
                Ok(None)
            }
        }
        _ if selected_here || includes.iter().any(|pattern| ancestor_selected(pattern, path)) => {
            Ok(Some(value.clone()))
        }
        _ => Ok(None),
    }
}

fn ancestor_selected(pattern: &str, path: &[&str]) -> bool {
    (1..path.len()).any(|length| pattern_matches(pattern, &path[..length]))
}

fn pattern_matches(pattern: &str, path: &[&str]) -> bool {
    let segments = pattern.split('.').collect::<Vec<_>>();
    match_segments(&segments, path)
}

fn match_segments(pattern: &[&str], path: &[&str]) -> bool {
    match (pattern.first(), path.first()) {
        (None, None) => true,
        (Some(&"**"), _) => {
            match_segments(&pattern[1..], path) || (!path.is_empty() && match_segments(pattern, &path[1..]))
        }
        (Some(segment), Some(value)) if wildcard_segment_matches(segment, value) => {
            match_segments(&pattern[1..], &path[1..])
        }
        _ => false,
    }
}

fn wildcard_segment_matches(pattern: &str, value: &str) -> bool {
    if pattern == "*" {
        return true;
    }
    let parts = pattern.split('*').collect::<Vec<_>>();
    if parts.len() == 1 {
        return pattern == value;
    }
    let mut remainder = value;
    for (position, part) in parts.iter().enumerate() {
        if part.is_empty() {
            continue;
        }
        if position == 0 && !pattern.starts_with('*') {
            let Some(next) = remainder.strip_prefix(part) else {
                return false;
            };
            remainder = next;
            continue;
        }
        let Some(offset) = remainder.find(part) else {
            return false;
        };
        remainder = &remainder[offset + part.len()..];
    }
    pattern.ends_with('*') || remainder.is_empty()
}

#[cfg(test)]
mod tests {
    use serde_json::json;

    use super::*;

    #[test]
    fn default_preserves_existing_native_exact_source_contract() {
        assert!(SearchSourceProjection::default().needs_exact_source());
        assert!(!SearchSourceProjection::Parsed.needs_exact_source());
        assert!(!SearchSourceProjection::ParsedAndExact.needs_logical_occupancy());
        assert!(SearchSourceProjection::ParsedExactAndLogicalOccupancy.needs_logical_occupancy());
    }

    #[test]
    fn filtered_projection_applies_includes_then_excludes_across_arrays() -> Result<(), SearchSourceProjectionError> {
        let projection = SearchSourceProjection::filtered(
            vec!["service.*".to_owned(), "events.**".to_owned()],
            vec!["service.secret".to_owned(), "events.token".to_owned()],
        )?;
        let projected = projection.project(&json!({
            "service": {"name": "api", "secret": "hidden"},
            "events": [{"message": "one", "token": "a"}, {"message": "two", "token": "b"}],
            "ignored": true
        }))?;
        assert_eq!(
            projected,
            Some(json!({
                "service": {"name": "api"},
                "events": [{"message": "one"}, {"message": "two"}]
            }))
        );
        Ok(())
    }

    #[test]
    fn filtered_projection_is_canonical_and_empty_selection_is_an_object() -> Result<(), SearchSourceProjectionError> {
        let projection = SearchSourceProjection::filtered(
            vec!["missing".to_owned(), "missing".to_owned()],
            vec!["z".to_owned(), "a".to_owned()],
        )?;
        assert_eq!(
            projection,
            SearchSourceProjection::Filtered {
                includes: vec!["missing".to_owned()],
                excludes: vec!["a".to_owned(), "z".to_owned()]
            }
        );
        assert_eq!(projection.project(&json!({"present": 1}))?, Some(json!({})));
        Ok(())
    }

    #[test]
    fn filtered_projection_rejects_unbounded_or_ambiguous_patterns() {
        assert_eq!(
            SearchSourceProjection::filtered(vec![String::new()], Vec::new()),
            Err(SearchSourceProjectionError::InvalidPattern)
        );
        assert_eq!(
            SearchSourceProjection::filtered(vec!["a..b".to_owned()], Vec::new()),
            Err(SearchSourceProjectionError::InvalidPattern)
        );
        assert_eq!(
            SearchSourceProjection::filtered(
                (0..=MAX_SOURCE_PROJECTION_PATTERNS)
                    .map(|index| format!("field{index}"))
                    .collect(),
                Vec::new()
            ),
            Err(SearchSourceProjectionError::TooManyPatterns)
        );
    }

    #[test]
    fn execution_projection_is_not_a_second_serialized_request_authority() -> Result<(), serde_json::Error> {
        let request = crate::SearchRequest {
            execution_source_projection: SearchSourceProjection::Parsed,
            ..crate::SearchRequest::default()
        };
        let bytes = serde_json::to_vec(&request)?;
        assert!(
            !bytes
                .windows(b"execution_source_projection".len())
                .any(|window| { window == b"execution_source_projection" })
        );
        let decoded: crate::SearchRequest = serde_json::from_slice(&bytes)?;
        assert_eq!(
            decoded.execution_source_projection,
            SearchSourceProjection::ParsedAndExact
        );
        Ok(())
    }
}