etdl_parser/

asyncapi.rs

use crate::ast::ExternalRef;
use crate::jsonptr;
use serde_json::Value;
use std::collections::BTreeMap;
use std::fs;
use std::path::{Path, PathBuf};

pub struct AsyncApiRegistry {
    documents: BTreeMap<String, AsyncApiDocument>,
}

struct AsyncApiDocument {
    _path: PathBuf,
    root: Value,
}

impl AsyncApiRegistry {
    pub fn new() -> Self {
        AsyncApiRegistry {
            documents: BTreeMap::new(),
        }
    }

    pub fn load(
        &mut self,
        alias: &str,
        location: &str,
        base_dir: &Path,
    ) -> Result<(), String> {
        let resolved_path = resolve_location(location, base_dir)?;
        let content =
            fs::read_to_string(&resolved_path).map_err(|e| {
                format!("cannot read AsyncAPI doc '{}': {}", resolved_path.display(), e)
            })?;

        let root: Value = if resolved_path.extension().map_or(false, |ext| ext == "json") {
            serde_json::from_str(&content).map_err(|e| {
                format!(
                    "invalid JSON in AsyncAPI doc '{}': {}",
                    resolved_path.display(),
                    e
                )
            })?
        } else {
            serde_yaml::from_str(&content).map_err(|e| {
                format!(
                    "invalid YAML in AsyncAPI doc '{}': {}",
                    resolved_path.display(),
                    e
                )
            })?
        };

        self.documents.insert(
            alias.to_string(),
            AsyncApiDocument {
                _path: resolved_path,
                root,
            },
        );
        Ok(())
    }

    pub fn resolve(&self, ext_ref: &ExternalRef) -> Result<&Value, String> {
        let doc = self.documents.get(&ext_ref.alias).ok_or_else(|| {
            format!(
                "import alias '{}' not found in loaded AsyncAPI documents",
                ext_ref.alias
            )
        })?;

        let pointer = &ext_ref.pointer;
        jsonptr::resolve_json_pointer(&doc.root, pointer).ok_or_else(|| {
            format!(
                "JSON Pointer '{}' does not resolve in AsyncAPI document '{}'",
                pointer, ext_ref.alias
            )
        })
    }

    pub fn resolve_ref(&self, alias: &str, pointer: &str) -> Result<&Value, String> {
        let doc = self.documents.get(alias).ok_or_else(|| {
            format!(
                "import alias '{}' not found in loaded AsyncAPI documents",
                alias
            )
        })?;

        jsonptr::resolve_json_pointer(&doc.root, pointer).ok_or_else(|| {
            format!(
                "JSON Pointer '{}' does not resolve in AsyncAPI document '{}'",
                pointer, alias
            )
        })
    }

    pub fn get_schema_for_path(
        &self,
        ext_ref: &ExternalRef,
        path_segments: &[crate::ecel::PathSegment],
    ) -> Result<Option<Value>, String> {
        let message_value = self.resolve(ext_ref)?;

        let payload_schema = message_value
            .get("payload")
            .or_else(|| message_value.get("schema"));

        let Some(schema) = payload_schema else {
            return Ok(None);
        };

        let resolved = resolve_schema_path(schema, path_segments);
        Ok(resolved)
    }
}

fn resolve_schema_path(
    schema: &Value,
    segments: &[crate::ecel::PathSegment],
) -> Option<Value> {
    if segments.is_empty() {
        return Some(schema.clone());
    }

    let first = &segments[0];

    match first {
        crate::ecel::PathSegment::Field(name) => {
            if name == "message" && segments.len() > 1 {
                return resolve_schema_path(schema, &segments[1..]);
            }

            let field_schema = resolve_field(schema, name)?;
            if segments.len() == 1 {
                Some(field_schema.clone())
            } else {
                resolve_schema_path(&field_schema, &segments[1..])
            }
        }
        crate::ecel::PathSegment::Wildcard => {
            let items_schema = resolve_array_items(schema)?;
            if segments.len() == 1 {
                Some(items_schema.clone())
            } else {
                resolve_schema_path(&items_schema, &segments[1..])
            }
        }
        crate::ecel::PathSegment::Index(_) => {
            let items_schema = resolve_array_items(schema)?;
            if segments.len() == 1 {
                Some(items_schema.clone())
            } else {
                resolve_schema_path(&items_schema, &segments[1..])
            }
        }
        crate::ecel::PathSegment::QuotedKey(name) => {
            let field_schema = resolve_field(schema, name)?;
            if segments.len() == 1 {
                Some(field_schema.clone())
            } else {
                resolve_schema_path(&field_schema, &segments[1..])
            }
        }
    }
}

fn resolve_field(schema: &Value, name: &str) -> Option<Value> {
    if let Some(properties) = schema.get("properties") {
        if let Some(field) = properties.get(name) {
            return Some(field.clone());
        }
    }

    if let Some(obj) = schema.as_object() {
        if let Some(field) = obj.get(name) {
            return Some(field.clone());
        }
    }

    None
}

fn resolve_array_items(schema: &Value) -> Option<Value> {
    if let Some(items) = schema.get("items") {
        return Some(items.clone());
    }

    if let Some(type_val) = schema.get("type") {
        if type_val.as_str() == Some("array") {
            if let Some(items) = schema.get("items") {
                return Some(items.clone());
            }
        }
    }

    None
}

fn resolve_location(location: &str, base_dir: &Path) -> Result<PathBuf, String> {
    if location.starts_with("http://") || location.starts_with("https://") {
        return Err(format!(
            "remote AsyncAPI imports not supported in this version: '{}'",
            location
        ));
    }

    let path = Path::new(location);
    if path.is_absolute() {
        Ok(path.to_path_buf())
    } else {
        Ok(base_dir.join(path))
    }
}