pjson_rs/stream/

priority.rs

//! Priority-based JSON streaming implementation
//!
//! This module implements the core Priority JSON Streaming protocol with:
//! - Skeleton-first approach
//! - JSON Path based patching
//! - Priority-based field ordering
//! - Incremental reconstruction

use crate::Result;
use crate::domain::value_objects::Priority;
use serde_json::{Map as JsonMap, Value as JsonValue};
use std::collections::VecDeque;

/// Custom serde for Priority in stream module
mod serde_priority {
    use crate::domain::value_objects::Priority;
    use serde::{Serialize, Serializer};

    pub fn serialize<S>(priority: &Priority, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        priority.value().serialize(serializer)
    }
}

/// JSON Path for addressing specific nodes in the JSON structure
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
pub struct JsonPath {
    segments: Vec<PathSegment>,
}

/// Single segment of a [`JsonPath`].
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
pub enum PathSegment {
    /// Root marker (`$`).
    Root,
    /// Object property key.
    Key(String),
    /// Array index.
    Index(usize),
    /// Wildcard matching any segment.
    Wildcard,
}

/// Patch operation for updating JSON structure
#[derive(Debug, Clone, serde::Serialize)]
pub struct JsonPatch {
    /// Path within the target JSON document.
    pub path: JsonPath,
    /// Operation to apply at `path`.
    pub operation: PatchOperation,
    /// Priority assigned to this patch.
    #[serde(with = "serde_priority")]
    pub priority: Priority,
}

/// Operation a [`JsonPatch`] performs at its target path.
#[derive(Debug, Clone, serde::Serialize)]
pub enum PatchOperation {
    /// Replace the value at the path with `value`.
    Set {
        /// New value to set at the path.
        value: JsonValue,
    },
    /// Append values to the array at the path.
    Append {
        /// Values appended to the target array.
        values: Vec<JsonValue>,
    },
    /// Replace the value at the path with `value` (semantically distinct from `Set`).
    Replace {
        /// Replacement value.
        value: JsonValue,
    },
    /// Remove the value at the path.
    Remove,
}

/// Streaming frame containing skeleton or patch data
#[derive(Debug, Clone, serde::Serialize)]
pub enum PriorityStreamFrame {
    /// Initial skeleton frame with placeholder values.
    Skeleton {
        /// Skeleton JSON value (nulls/empties for fields filled later).
        data: JsonValue,
        /// Priority of the skeleton frame.
        #[serde(with = "serde_priority")]
        priority: Priority,
        /// Whether the skeleton is final or further skeletons may follow.
        complete: bool,
    },
    /// Batch of patches sharing the same priority.
    Patch {
        /// Patches in this batch.
        patches: Vec<JsonPatch>,
        /// Priority shared by all patches in the batch.
        #[serde(with = "serde_priority")]
        priority: Priority,
    },
    /// Terminal frame indicating the stream is complete.
    Complete {
        /// Optional checksum of the reconstructed payload.
        checksum: Option<u64>,
    },
}

/// Priority-based JSON streamer
pub struct PriorityStreamer {
    config: StreamerConfig,
}

/// Configuration for [`PriorityStreamer`].
#[derive(Debug, Clone)]
pub struct StreamerConfig {
    /// Enable name-based heuristics that infer priorities from common field names.
    pub detect_semantics: bool,
    /// Maximum number of patches per [`PriorityStreamFrame::Patch`] batch.
    pub max_patch_size: usize,
    /// Patches with priority below this threshold are dropped.
    pub priority_threshold: Priority,
}

impl Default for StreamerConfig {
    fn default() -> Self {
        Self {
            detect_semantics: true,
            max_patch_size: 100,
            priority_threshold: Priority::LOW,
        }
    }
}

impl PriorityStreamer {
    /// Create new priority streamer
    pub fn new() -> Self {
        Self::with_config(StreamerConfig::default())
    }

    /// Create streamer with custom configuration
    pub fn with_config(config: StreamerConfig) -> Self {
        Self { config }
    }

    /// Analyze JSON and create streaming plan
    pub fn analyze(&self, json: &JsonValue) -> Result<StreamingPlan> {
        let mut plan = StreamingPlan::new();

        // Generate skeleton
        let skeleton = self.generate_skeleton(json);
        plan.frames.push_back(PriorityStreamFrame::Skeleton {
            data: skeleton,
            priority: Priority::CRITICAL,
            complete: false,
        });

        // Extract patches by priority
        let mut patches = Vec::new();
        self.extract_patches(json, &JsonPath::root(), &mut patches)?;

        // Group patches by priority
        patches.sort_by_key(|patch| std::cmp::Reverse(patch.priority));

        let mut current_priority = Priority::CRITICAL;
        let mut current_batch = Vec::new();

        for patch in patches {
            if patch.priority != current_priority && !current_batch.is_empty() {
                plan.frames.push_back(PriorityStreamFrame::Patch {
                    patches: current_batch,
                    priority: current_priority,
                });
                current_batch = Vec::new();
            }
            current_priority = patch.priority;
            current_batch.push(patch);

            if current_batch.len() >= self.config.max_patch_size {
                plan.frames.push_back(PriorityStreamFrame::Patch {
                    patches: current_batch,
                    priority: current_priority,
                });
                current_batch = Vec::new();
            }
        }

        // Add remaining patches
        if !current_batch.is_empty() {
            plan.frames.push_back(PriorityStreamFrame::Patch {
                patches: current_batch,
                priority: current_priority,
            });
        }

        // Add completion frame
        plan.frames
            .push_back(PriorityStreamFrame::Complete { checksum: None });

        Ok(plan)
    }

    /// Generate skeleton structure with null/empty values
    fn generate_skeleton(&self, json: &JsonValue) -> JsonValue {
        match json {
            JsonValue::Object(map) => {
                let mut skeleton = JsonMap::new();
                for (key, value) in map {
                    skeleton.insert(
                        key.clone(),
                        match value {
                            JsonValue::Array(_) => JsonValue::Array(vec![]),
                            JsonValue::Object(_) => self.generate_skeleton(value),
                            JsonValue::String(_) => JsonValue::Null,
                            JsonValue::Number(_) => JsonValue::Number(0.into()),
                            JsonValue::Bool(_) => JsonValue::Bool(false),
                            JsonValue::Null => JsonValue::Null,
                        },
                    );
                }
                JsonValue::Object(skeleton)
            }
            JsonValue::Array(_) => JsonValue::Array(vec![]),
            _ => JsonValue::Null,
        }
    }

    /// Extract patches from JSON structure
    fn extract_patches(
        &self,
        json: &JsonValue,
        current_path: &JsonPath,
        patches: &mut Vec<JsonPatch>,
    ) -> Result<()> {
        match json {
            JsonValue::Object(map) => {
                for (key, value) in map {
                    let field_path = current_path.append_key(key);
                    let priority = self.calculate_field_priority(&field_path, key, value);

                    // Create patch for this field
                    patches.push(JsonPatch {
                        path: field_path.clone(),
                        operation: PatchOperation::Set {
                            value: value.clone(),
                        },
                        priority,
                    });

                    // Recursively process nested structures
                    self.extract_patches(value, &field_path, patches)?;
                }
            }
            JsonValue::Array(arr) => {
                // For arrays, create append operations in chunks
                if arr.len() > 10 {
                    // Chunk large arrays
                    for chunk in arr.chunks(self.config.max_patch_size) {
                        patches.push(JsonPatch {
                            path: current_path.clone(),
                            operation: PatchOperation::Append {
                                values: chunk.to_vec(),
                            },
                            priority: self.calculate_array_priority(current_path, chunk),
                        });
                    }
                } else if !arr.is_empty() {
                    patches.push(JsonPatch {
                        path: current_path.clone(),
                        operation: PatchOperation::Append {
                            values: arr.clone(),
                        },
                        priority: self.calculate_array_priority(current_path, arr),
                    });
                }
            }
            _ => {
                // Primitive values handled by parent object/array
            }
        }

        Ok(())
    }

    /// Calculate priority for a field based on path and content
    fn calculate_field_priority(&self, _path: &JsonPath, key: &str, value: &JsonValue) -> Priority {
        // Critical fields
        if matches!(key, "id" | "uuid" | "status" | "type" | "kind") {
            return Priority::CRITICAL;
        }

        // High priority fields
        if matches!(key, "name" | "title" | "label" | "email" | "username") {
            return Priority::HIGH;
        }

        // Low priority patterns
        if key.contains("analytics") || key.contains("stats") || key.contains("meta") {
            return Priority::LOW;
        }

        if matches!(key, "reviews" | "comments" | "logs" | "history") {
            return Priority::BACKGROUND;
        }

        // Content-based priority
        match value {
            JsonValue::Array(arr) if arr.len() > 100 => Priority::BACKGROUND,
            JsonValue::Object(obj) if obj.contains_key("timestamp") => Priority::MEDIUM,
            JsonValue::String(s) if s.len() > 1000 => Priority::LOW,
            _ => Priority::MEDIUM,
        }
    }

    /// Calculate priority for array elements
    fn calculate_array_priority(&self, path: &JsonPath, elements: &[JsonValue]) -> Priority {
        // Large arrays get background priority
        if elements.len() > 50 {
            return Priority::BACKGROUND;
        }

        // Arrays in certain paths get different priorities
        if let Some(last_key) = path.last_key() {
            if matches!(last_key.as_str(), "reviews" | "comments" | "logs") {
                return Priority::BACKGROUND;
            }
            if matches!(last_key.as_str(), "items" | "data" | "results") {
                return Priority::MEDIUM;
            }
        }

        Priority::MEDIUM
    }
}

/// Plan for streaming JSON with priority ordering
#[derive(Debug)]
pub struct StreamingPlan {
    /// Ordered queue of frames produced by analysis.
    pub frames: VecDeque<PriorityStreamFrame>,
}

impl Default for StreamingPlan {
    fn default() -> Self {
        Self::new()
    }
}

impl StreamingPlan {
    /// Create an empty plan with no frames.
    pub fn new() -> Self {
        Self {
            frames: VecDeque::new(),
        }
    }

    /// Get next frame to send
    pub fn next_frame(&mut self) -> Option<PriorityStreamFrame> {
        self.frames.pop_front()
    }

    /// Check if streaming is complete
    pub fn is_complete(&self) -> bool {
        self.frames.is_empty()
    }

    /// Get remaining frame count
    pub fn remaining_frames(&self) -> usize {
        self.frames.len()
    }

    /// Get iterator over frames
    pub fn frames(&self) -> impl Iterator<Item = &PriorityStreamFrame> {
        self.frames.iter()
    }
}

impl JsonPath {
    /// Create root path
    pub fn root() -> Self {
        let segments = vec![PathSegment::Root];
        Self { segments }
    }

    /// Append key segment
    pub fn append_key(&self, key: &str) -> Self {
        let mut segments = self.segments.clone();
        segments.push(PathSegment::Key(key.to_string()));
        Self { segments }
    }

    /// Append index segment
    pub fn append_index(&self, index: usize) -> Self {
        let mut segments = self.segments.clone();
        segments.push(PathSegment::Index(index));
        Self { segments }
    }

    /// Get the last key in the path
    pub fn last_key(&self) -> Option<String> {
        self.segments.iter().rev().find_map(|segment| {
            if let PathSegment::Key(key) = segment {
                Some(key.clone())
            } else {
                None
            }
        })
    }

    /// Get segments (read-only)
    pub fn segments(&self) -> &[PathSegment] {
        &self.segments
    }

    /// Get number of segments
    pub fn len(&self) -> usize {
        self.segments.len()
    }

    /// Check if path is empty
    pub fn is_empty(&self) -> bool {
        self.segments.is_empty()
    }

    /// Create JsonPath from segments (for testing)
    pub fn from_segments(segments: Vec<PathSegment>) -> Self {
        Self { segments }
    }

    /// Convert to JSON Pointer string format
    pub fn to_json_pointer(&self) -> String {
        let mut pointer = String::new();
        for segment in &self.segments {
            match segment {
                PathSegment::Root => {}
                PathSegment::Key(key) => {
                    pointer.push('/');
                    pointer.push_str(key);
                }
                PathSegment::Index(idx) => {
                    pointer.push('/');
                    pointer.push_str(&idx.to_string());
                }
                PathSegment::Wildcard => {
                    pointer.push_str("/*");
                }
            }
        }
        if pointer.is_empty() {
            "/".to_string()
        } else {
            pointer
        }
    }
}

impl Default for PriorityStreamer {
    fn default() -> Self {
        Self::new()
    }
}

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

    #[test]
    fn test_json_path_creation() {
        let path = JsonPath::root();
        assert_eq!(path.to_json_pointer(), "/");

        let path = path.append_key("users").append_index(0).append_key("name");
        assert_eq!(path.to_json_pointer(), "/users/0/name");
    }

    #[test]
    fn test_priority_comparison() {
        assert!(Priority::CRITICAL > Priority::HIGH);
        assert!(Priority::HIGH > Priority::MEDIUM);
        assert!(Priority::MEDIUM > Priority::LOW);
        assert!(Priority::LOW > Priority::BACKGROUND);
    }

    #[test]
    fn test_skeleton_generation() {
        let streamer = PriorityStreamer::new();
        let json = json!({
            "name": "John",
            "age": 30,
            "active": true,
            "posts": ["post1", "post2"]
        });

        let skeleton = streamer.generate_skeleton(&json);
        let expected = json!({
            "name": null,
            "age": 0,
            "active": false,
            "posts": []
        });

        assert_eq!(skeleton, expected);
    }

    #[test]
    fn test_field_priority_calculation() {
        let streamer = PriorityStreamer::new();
        let path = JsonPath::root();

        assert_eq!(
            streamer.calculate_field_priority(&path, "id", &json!(123)),
            Priority::CRITICAL
        );

        assert_eq!(
            streamer.calculate_field_priority(&path, "name", &json!("John")),
            Priority::HIGH
        );

        assert_eq!(
            streamer.calculate_field_priority(&path, "reviews", &json!([])),
            Priority::BACKGROUND
        );
    }

    #[test]
    fn test_streaming_plan_creation() {
        let streamer = PriorityStreamer::new();
        let json = json!({
            "id": 1,
            "name": "John",
            "bio": "Software developer",
            "reviews": ["Good", "Excellent"]
        });

        let plan = streamer.analyze(&json).unwrap();
        assert!(!plan.is_complete());
        assert!(plan.remaining_frames() > 0);
    }
}