el-retired/engram/engrams/engram-projection/src/engine.rs

/// Projection engine — maps activated nodes through a ProjectionSchema.
///
/// The engine is stateless: it takes a schema and a result set, and returns
/// the projected view. No mutation of the underlying graph occurs.
use base64::{engine::general_purpose::STANDARD as B64, Engine as _};
use engram_core::types::{ActivatedNode, NodeType};
use serde_json::{json, Value};
use std::collections::HashMap;
use uuid::Uuid;

use crate::error::ProjectionResult;
use crate::schema::{
    FieldMapping, FieldSource, NodeFilter, ProjectedRow, ProjectionResult as PResult,
    ProjectionSchema, ProjectionType,
};

/// Stateless projection executor.
pub struct ProjectionEngine;

impl ProjectionEngine {
    /// Apply a projection schema to a set of activated nodes.
    ///
    /// Returns a `ProjectionResult` containing the shaped output.
    /// The activation result set is not modified.
    pub fn project(
        schema: &ProjectionSchema,
        activated: &[ActivatedNode],
    ) -> ProjectionResult<PResult> {
        // Step 1: filter to in-scope nodes
        let in_scope: Vec<&ActivatedNode> = activated
            .iter()
            .filter(|a| matches_filter(a, &schema.node_filter))
            .collect();

        let nodes_in_scope = in_scope.len();

        match schema.projection_type {
            ProjectionType::Relational | ProjectionType::Document => {
                let rows = in_scope
                    .iter()
                    .map(|a| project_row(a, &schema.field_mappings))
                    .collect::<ProjectionResult<Vec<_>>>()?;

                Ok(PResult {
                    schema_name: schema.name.clone(),
                    nodes_in_scope,
                    rows,
                    key_value: None,
                    wide_column: None,
                })
            }

            ProjectionType::KeyValue => {
                let mut kv: HashMap<String, Value> = HashMap::new();
                for a in &in_scope {
                    let key = a.node.id.to_string();
                    let val = String::from_utf8_lossy(&a.node.content).to_string();
                    kv.insert(key, Value::String(val));
                }
                Ok(PResult {
                    schema_name: schema.name.clone(),
                    nodes_in_scope,
                    rows: vec![],
                    key_value: Some(kv),
                    wide_column: None,
                })
            }

            ProjectionType::WideColumn => {
                let mut wc: HashMap<String, HashMap<String, Value>> = HashMap::new();
                for a in &in_scope {
                    let id = a.node.id.to_string();
                    let mut cols = HashMap::new();
                    for mapping in &schema.field_mappings {
                        let val = extract_field(a, &mapping.source)
                            .unwrap_or_else(|| mapping.default.clone().unwrap_or(Value::Null));
                        cols.insert(mapping.field_name.clone(), val);
                    }
                    wc.insert(id, cols);
                }
                Ok(PResult {
                    schema_name: schema.name.clone(),
                    nodes_in_scope,
                    rows: vec![],
                    key_value: None,
                    wide_column: Some(wc),
                })
            }
        }
    }
}

// ── Node filter evaluation ────────────────────────────────────────────────────

fn matches_filter(a: &ActivatedNode, filter: &NodeFilter) -> bool {
    match filter {
        NodeFilter::All => true,

        NodeFilter::ByType(types) => {
            let node_type_str = node_type_str(&a.node.node_type);
            types.iter().any(|t| t == &node_type_str)
        }

        NodeFilter::ByTier(tiers) => tiers.iter().any(|t| t == &a.node.tier),

        NodeFilter::ByTag(tags) => {
            // Tags are searched in content (treated as UTF-8) as a simple substring match.
            // This is intentionally lenient — callers may embed tag metadata in content.
            let content_str = String::from_utf8_lossy(&a.node.content);
            tags.iter().any(|tag| content_str.contains(tag.as_str()))
        }

        NodeFilter::ByActivationThreshold(threshold) => a.activation_strength >= *threshold,

        NodeFilter::BySalience(threshold) => a.node.salience >= *threshold,

        NodeFilter::Combined(filters) => filters.iter().all(|f| matches_filter(a, f)),

        NodeFilter::Any(filters) => filters.iter().any(|f| matches_filter(a, f)),
    }
}

// ── Row projection ────────────────────────────────────────────────────────────

fn project_row(a: &ActivatedNode, mappings: &[FieldMapping]) -> ProjectionResult<ProjectedRow> {
    let mut fields = HashMap::new();
    for mapping in mappings {
        let val = extract_field(a, &mapping.source)
            .unwrap_or_else(|| mapping.default.clone().unwrap_or(Value::Null));
        fields.insert(mapping.field_name.clone(), val);
    }
    Ok(ProjectedRow {
        node_id: a.node.id,
        fields,
    })
}

// ── Field extraction ──────────────────────────────────────────────────────────

fn extract_field(a: &ActivatedNode, source: &FieldSource) -> Option<Value> {
    match source {
        FieldSource::NodeId => Some(Value::String(a.node.id.to_string())),

        FieldSource::NodeType => Some(Value::String(node_type_str(&a.node.node_type).to_string())),

        FieldSource::Tier => Some(Value::String(tier_str(&a.node.tier).to_string())),

        FieldSource::Salience => Some(json!(a.node.salience)),

        FieldSource::Importance => Some(json!(a.node.importance)),

        FieldSource::ActivationStrength => Some(json!(a.activation_strength)),

        FieldSource::Hops => Some(json!(a.hops)),

        FieldSource::CreatedAt => Some(json!(a.node.created_at)),

        FieldSource::LastActivated => Some(json!(a.node.last_activated)),

        FieldSource::ActivationCount => Some(json!(a.node.activation_count)),

        FieldSource::ContentRaw => {
            Some(Value::String(String::from_utf8_lossy(&a.node.content).to_string()))
        }

        FieldSource::ContentBase64 => Some(Value::String(B64.encode(&a.node.content))),

        FieldSource::ContentJsonPath(path) => {
            // Parse content as JSON, then traverse the dot-path
            let content_str = std::str::from_utf8(&a.node.content).ok()?;
            let doc: Value = serde_json::from_str(content_str).ok()?;
            traverse_json_path(&doc, path).cloned()
        }

        FieldSource::Literal(v) => Some(v.clone()),
    }
}

/// Traverse a dot-separated JSON path.
/// E.g., "user.name" on `{"user": {"name": "Alice"}}` returns `"Alice"`.
fn traverse_json_path<'a>(doc: &'a Value, path: &str) -> Option<&'a Value> {
    let mut current = doc;
    for segment in path.split('.') {
        current = match current {
            Value::Object(map) => map.get(segment)?,
            Value::Array(arr) => {
                let idx: usize = segment.parse().ok()?;
                arr.get(idx)?
            }
            _ => return None,
        };
    }
    Some(current)
}

// ── String helpers ────────────────────────────────────────────────────────────

fn node_type_str(t: &NodeType) -> String {
    match t {
        NodeType::Memory => "Memory".to_string(),
        NodeType::Concept => "Concept".to_string(),
        NodeType::Event => "Event".to_string(),
        NodeType::Entity => "Entity".to_string(),
        NodeType::Process => "Process".to_string(),
        NodeType::InternalState => "InternalState".to_string(),
        NodeType::Custom(s) => s.clone(),
    }
}

fn tier_str(t: &engram_core::types::MemoryTier) -> &'static str {
    use engram_core::types::MemoryTier;
    match t {
        MemoryTier::Working => "Working",
        MemoryTier::Episodic => "Episodic",
        MemoryTier::Semantic => "Semantic",
        MemoryTier::Procedural => "Procedural",
    }
}

/// Extract node_id from a projected row (used for display / keying).
pub fn row_id(row: &ProjectedRow) -> Uuid {
    row.node_id
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::schema::{FieldMapping, FieldSource, NodeFilter, ProjectionSchema, ProjectionType};
    use engram_core::types::{ActivatedNode, MemoryTier, Node, NodeType};

    fn make_node(content: &str, tier: MemoryTier, importance: f32) -> Node {
        Node::new(
            NodeType::Memory,
            vec![1.0, 0.0],
            content.as_bytes().to_vec(),
            tier,
            importance,
        )
    }

    fn make_activated(node: Node, strength: f32) -> ActivatedNode {
        ActivatedNode {
            node,
            activation_strength: strength,
            hops: 1,
        }
    }

    #[test]
    fn test_relational_projection_basic() {
        let node = make_node("hello world", MemoryTier::Semantic, 0.8);
        let activated = vec![make_activated(node, 0.9)];

        let schema = ProjectionSchema {
            name: "test".into(),
            description: None,
            projection_type: ProjectionType::Relational,
            node_filter: NodeFilter::All,
            field_mappings: vec![
                FieldMapping {
                    field_name: "content".into(),
                    source: FieldSource::ContentRaw,
                    default: None,
                },
                FieldMapping {
                    field_name: "tier".into(),
                    source: FieldSource::Tier,
                    default: None,
                },
                FieldMapping {
                    field_name: "strength".into(),
                    source: FieldSource::ActivationStrength,
                    default: None,
                },
            ],
        };

        let result = ProjectionEngine::project(&schema, &activated).unwrap();
        assert_eq!(result.nodes_in_scope, 1);
        assert_eq!(result.rows.len(), 1);
        assert_eq!(result.rows[0].fields["content"], Value::String("hello world".into()));
        assert_eq!(result.rows[0].fields["tier"], Value::String("Semantic".into()));
    }

    #[test]
    fn test_key_value_projection() {
        let node = make_node("test content", MemoryTier::Working, 0.5);
        let activated = vec![make_activated(node, 0.7)];

        let schema = ProjectionSchema {
            name: "kv".into(),
            description: None,
            projection_type: ProjectionType::KeyValue,
            node_filter: NodeFilter::All,
            field_mappings: vec![],
        };

        let result = ProjectionEngine::project(&schema, &activated).unwrap();
        assert_eq!(result.nodes_in_scope, 1);
        let kv = result.key_value.unwrap();
        assert_eq!(kv.len(), 1);
        let content = kv.values().next().unwrap();
        assert_eq!(content, &Value::String("test content".into()));
    }

    #[test]
    fn test_filter_by_activation_threshold() {
        let n1 = make_activated(make_node("high", MemoryTier::Semantic, 0.9), 0.8);
        let n2 = make_activated(make_node("low", MemoryTier::Episodic, 0.3), 0.1);

        let schema = ProjectionSchema {
            name: "filtered".into(),
            description: None,
            projection_type: ProjectionType::Relational,
            node_filter: NodeFilter::ByActivationThreshold(0.5),
            field_mappings: vec![FieldMapping {
                field_name: "content".into(),
                source: FieldSource::ContentRaw,
                default: None,
            }],
        };

        let result = ProjectionEngine::project(&schema, &[n1, n2]).unwrap();
        assert_eq!(result.nodes_in_scope, 1);
        assert_eq!(result.rows[0].fields["content"], Value::String("high".into()));
    }

    #[test]
    fn test_filter_by_tier() {
        let n1 = make_activated(make_node("semantic", MemoryTier::Semantic, 0.9), 0.5);
        let n2 = make_activated(make_node("working", MemoryTier::Working, 0.5), 0.5);

        let schema = ProjectionSchema {
            name: "tier_filter".into(),
            description: None,
            projection_type: ProjectionType::Relational,
            node_filter: NodeFilter::ByTier(vec![MemoryTier::Semantic]),
            field_mappings: vec![FieldMapping {
                field_name: "content".into(),
                source: FieldSource::ContentRaw,
                default: None,
            }],
        };

        let result = ProjectionEngine::project(&schema, &[n1, n2]).unwrap();
        assert_eq!(result.nodes_in_scope, 1);
        assert_eq!(result.rows[0].fields["content"], Value::String("semantic".into()));
    }

    #[test]
    fn test_json_path_extraction() {
        let content = r#"{"user": {"name": "Alice", "age": 30}}"#;
        let node = make_node(content, MemoryTier::Semantic, 0.8);
        let activated = vec![make_activated(node, 0.9)];

        let schema = ProjectionSchema {
            name: "json_path".into(),
            description: None,
            projection_type: ProjectionType::Relational,
            node_filter: NodeFilter::All,
            field_mappings: vec![
                FieldMapping {
                    field_name: "name".into(),
                    source: FieldSource::ContentJsonPath("user.name".into()),
                    default: None,
                },
                FieldMapping {
                    field_name: "age".into(),
                    source: FieldSource::ContentJsonPath("user.age".into()),
                    default: None,
                },
            ],
        };

        let result = ProjectionEngine::project(&schema, &activated).unwrap();
        assert_eq!(result.rows[0].fields["name"], Value::String("Alice".into()));
        assert_eq!(result.rows[0].fields["age"], json!(30));
    }

    #[test]
    fn test_wide_column_projection() {
        let n1 = make_activated(make_node("col_content", MemoryTier::Procedural, 0.6), 0.5);

        let schema = ProjectionSchema {
            name: "wide".into(),
            description: None,
            projection_type: ProjectionType::WideColumn,
            node_filter: NodeFilter::All,
            field_mappings: vec![
                FieldMapping {
                    field_name: "raw".into(),
                    source: FieldSource::ContentRaw,
                    default: None,
                },
                FieldMapping {
                    field_name: "tier".into(),
                    source: FieldSource::Tier,
                    default: None,
                },
            ],
        };

        let result = ProjectionEngine::project(&schema, &[n1]).unwrap();
        assert_eq!(result.nodes_in_scope, 1);
        let wc = result.wide_column.unwrap();
        assert_eq!(wc.len(), 1);
        let cols = wc.values().next().unwrap();
        assert_eq!(cols["raw"], Value::String("col_content".into()));
        assert_eq!(cols["tier"], Value::String("Procedural".into()));
    }

    #[test]
    fn test_combined_filter() {
        let n1 = make_activated(make_node("tag:important semantic", MemoryTier::Semantic, 0.9), 0.8);
        let n2 = make_activated(make_node("no tag", MemoryTier::Semantic, 0.9), 0.8);
        let n3 = make_activated(make_node("tag:important working", MemoryTier::Working, 0.3), 0.8);

        let filter = NodeFilter::Combined(vec![
            NodeFilter::ByTier(vec![MemoryTier::Semantic]),
            NodeFilter::ByTag(vec!["tag:important".into()]),
        ]);

        let schema = ProjectionSchema {
            name: "combined".into(),
            description: None,
            projection_type: ProjectionType::Relational,
            node_filter: filter,
            field_mappings: vec![FieldMapping {
                field_name: "content".into(),
                source: FieldSource::ContentRaw,
                default: None,
            }],
        };

        let result = ProjectionEngine::project(&schema, &[n1, n2, n3]).unwrap();
        assert_eq!(result.nodes_in_scope, 1);
        assert_eq!(
            result.rows[0].fields["content"],
            Value::String("tag:important semantic".into())
        );
    }

    #[test]
    fn test_literal_field() {
        let node = make_node("any", MemoryTier::Working, 0.5);
        let activated = vec![make_activated(node, 0.5)];

        let schema = ProjectionSchema {
            name: "literal".into(),
            description: None,
            projection_type: ProjectionType::Relational,
            node_filter: NodeFilter::All,
            field_mappings: vec![FieldMapping {
                field_name: "schema_version".into(),
                source: FieldSource::Literal(json!("v1")),
                default: None,
            }],
        };

        let result = ProjectionEngine::project(&schema, &activated).unwrap();
        assert_eq!(result.rows[0].fields["schema_version"], Value::String("v1".into()));
    }
}