document_svg/document/
avro.rs

1//! Bounded Apache Avro JSON schema/protocol previews.
2//!
3//! Avro schemas describe records, enums, arrays, maps, unions and fixed values;
4//! protocols additionally describe typed messages. This adapter renders names,
5//! field paths and type shapes as inert metadata. Defaults, documentation,
6//! aliases, logical values, imports and protocol execution are never evaluated.
7
8use serde_json::Value;
9use std::path::Path;
10
11use crate::convert::{ConvertOptions, PageConsumer, read_limited_file};
12use crate::document::html::{HtmlBlock, render_blocks_to_pages};
13use crate::error::{Error, Result};
14use crate::table::{TableAlign, TableData};
15
16const MAX_AVRO_BYTES: u64 = 64 * 1024 * 1024;
17const MAX_AVRO_DEPTH: usize = 100;
18const MAX_AVRO_VALUES: usize = 300_000;
19const MAX_AVRO_ROWS: usize = 200_000;
20const MAX_AVRO_STRING_BYTES: usize = 2 * 1024 * 1024;
21const MAX_AVRO_DISPLAY_BYTES: usize = 512;
22
23pub(crate) fn looks_like_prefix(prefix: &[u8]) -> bool {
24    let text = String::from_utf8_lossy(prefix);
25    let trimmed = text.trim_start_matches('\u{feff}').trim_start();
26    if !trimmed.starts_with('{') && !trimmed.starts_with('[') {
27        return false;
28    }
29    if let Ok(value) = serde_json::from_str::<Value>(trimmed) {
30        return looks_like_value(&value);
31    }
32    text.contains("\"protocol\"") || (text.contains("\"fields\"") && text.contains("\"type\""))
33}
34
35fn looks_like_value(value: &Value) -> bool {
36    match value {
37        Value::Object(object) => {
38            (object.get("type").and_then(Value::as_str) == Some("record")
39                && object.get("fields").and_then(Value::as_array).is_some())
40                || (object.get("protocol").and_then(Value::as_str).is_some()
41                    && object.get("messages").and_then(Value::as_object).is_some())
42                || object
43                    .get("types")
44                    .and_then(Value::as_array)
45                    .is_some_and(|types| types.iter().any(looks_like_value))
46        }
47        Value::Array(values) => values.iter().any(looks_like_value),
48        _ => false,
49    }
50}
51
52struct AvroPageSink<'a> {
53    inner: &'a mut dyn PageConsumer,
54    warnings: &'a [String],
55}
56
57impl PageConsumer for AvroPageSink<'_> {
58    fn consume(&mut self, mut page: crate::ir::Page) -> Result<()> {
59        page.source_format = "avro".into();
60        if page.title.is_empty() {
61            page.title = "Apache Avro schema".into();
62        }
63        page.description =
64            "Avro schema and protocol metadata is rendered inertly; defaults, documentation, imports and message execution are not evaluated".into();
65        for warning in self.warnings {
66            page.warn(warning.clone());
67        }
68        self.inner.consume(page)
69    }
70}
71
72pub(crate) fn convert(
73    path: &Path,
74    options: &ConvertOptions,
75    sink: &mut dyn PageConsumer,
76) -> Result<Vec<String>> {
77    let bytes = read_limited_file(
78        path,
79        options.max_input_bytes.min(MAX_AVRO_BYTES),
80        "Avro JSON schema input",
81    )?;
82    let text = String::from_utf8(bytes)
83        .map_err(|error| Error::InvalidInput(format!("Avro JSON must be UTF-8 JSON: {error}")))?;
84    let (table, metadata, warnings) = parse(&text)?;
85    let blocks = vec![
86        HtmlBlock::Heading {
87            level: 1,
88            text: "Apache Avro schema".into(),
89        },
90        HtmlBlock::Paragraph { text: metadata },
91        HtmlBlock::Table(table),
92    ];
93    let mut page_sink = AvroPageSink {
94        inner: sink,
95        warnings: &warnings,
96    };
97    render_blocks_to_pages(&blocks, &mut page_sink, options)?;
98    Ok(warnings)
99}
100
101fn parse(text: &str) -> Result<(TableData, String, Vec<String>)> {
102    if text.len() as u64 > MAX_AVRO_BYTES {
103        return Err(Error::LimitExceeded(format!(
104            "Avro JSON exceeds {MAX_AVRO_BYTES} bytes"
105        )));
106    }
107    preflight_depth(text)?;
108    let value: Value = serde_json::from_str(text)
109        .map_err(|error| Error::InvalidInput(format!("invalid Avro JSON: {error}")))?;
110    let mut values = 0usize;
111    count_values(&value, 0, &mut values)?;
112    let mut rows = Vec::new();
113    let mut warnings = Vec::new();
114    let mut named = 0usize;
115    let mut fields = 0usize;
116    let mut messages = 0usize;
117    let metadata = match &value {
118        Value::Object(object) if object.get("protocol").is_some() => {
119            let protocol = required_string(object, "protocol")?;
120            let namespace = object
121                .get("namespace")
122                .and_then(Value::as_str)
123                .unwrap_or("—");
124            let types = object
125                .get("types")
126                .and_then(Value::as_array)
127                .map_or(0, Vec::len);
128            if let Some(types) = object.get("types") {
129                let types = types.as_array().ok_or_else(|| {
130                    Error::InvalidInput("Avro protocol types must be an array".into())
131                })?;
132                for schema in types {
133                    walk_schema(schema, "/types", 0, &mut rows, &mut named, &mut fields)?;
134                }
135            }
136            if let Some(message_map) = object.get("messages") {
137                let message_map = message_map.as_object().ok_or_else(|| {
138                    Error::InvalidInput("Avro protocol messages must be an object".into())
139                })?;
140                for (name, message) in message_map {
141                    messages = messages.saturating_add(1);
142                    let message = message.as_object().ok_or_else(|| {
143                        Error::InvalidInput(format!("Avro message {name} must be an object"))
144                    })?;
145                    let request = message
146                        .get("request")
147                        .and_then(Value::as_array)
148                        .map_or(0, Vec::len);
149                    let response = type_label(message.get("response"));
150                    let errors = message
151                        .get("errors")
152                        .and_then(Value::as_array)
153                        .map_or(0, Vec::len);
154                    push_row(
155                        &mut rows,
156                        vec![
157                            format!("/messages/{name}"),
158                            "message".into(),
159                            format!("request {request}; response {response}; errors {errors}"),
160                        ],
161                    )?;
162                }
163            }
164            let metadata = format!(
165                "Protocol: {}\nNamespace: {}\nNamed schemas: {}\nFields: {}\nMessages: {}\nDeclared types: {}",
166                truncate(protocol),
167                truncate(namespace),
168                named,
169                fields,
170                messages,
171                types
172            );
173            metadata
174        }
175        Value::Object(object) => {
176            walk_schema(&value, "/", 0, &mut rows, &mut named, &mut fields)?;
177            let name = object
178                .get("name")
179                .and_then(Value::as_str)
180                .unwrap_or("(anonymous)");
181            format!(
182                "Root schema: {}\nNamed schemas: {}\nFields: {}",
183                truncate(name),
184                named,
185                fields
186            )
187        }
188        Value::Array(values) => {
189            for (index, schema) in values.iter().enumerate() {
190                walk_schema(
191                    schema,
192                    &format!("/{index}"),
193                    0,
194                    &mut rows,
195                    &mut named,
196                    &mut fields,
197                )?;
198            }
199            format!(
200                "Root union members: {}\nNamed schemas: {}\nFields: {}",
201                values.len(),
202                named,
203                fields
204            )
205        }
206        _ => {
207            return Err(Error::InvalidInput(
208                "Avro JSON root must be a schema object, protocol or union array".into(),
209            ));
210        }
211    };
212    if rows.is_empty() {
213        rows.push(vec!["/".into(), "schema".into(), "(no fields)".into()]);
214    }
215    warnings.push("Avro defaults, docs, aliases, logical-type semantics, field values, imports and protocol messages remain inert; no code generation, deserialization, RPC or external resource access occurs".into());
216    warnings.push("Avro JSON is bounded by input size, nesting depth, value count, schema-row count and string length; named references are shown as type text and are not resolved".into());
217    Ok((
218        TableData {
219            headers: vec!["Path".into(), "Kind".into(), "Type / detail".into()],
220            rows,
221            alignments: vec![TableAlign::Left; 3],
222            raw_source: String::new(),
223        },
224        metadata,
225        warnings,
226    ))
227}
228
229fn walk_schema(
230    value: &Value,
231    path: &str,
232    depth: usize,
233    rows: &mut Vec<Vec<String>>,
234    named: &mut usize,
235    fields: &mut usize,
236) -> Result<()> {
237    if depth > MAX_AVRO_DEPTH {
238        return Err(Error::LimitExceeded(format!(
239            "Avro schema nesting exceeds {MAX_AVRO_DEPTH} levels"
240        )));
241    }
242    match value {
243        Value::String(type_name) => {
244            push_row(
245                rows,
246                vec![truncate(path), "reference".into(), truncate(type_name)],
247            )?;
248        }
249        Value::Array(members) => {
250            push_row(
251                rows,
252                vec![
253                    truncate(path),
254                    "union".into(),
255                    format!("{} members", members.len()),
256                ],
257            )?;
258            for (index, member) in members.iter().enumerate() {
259                walk_schema(
260                    member,
261                    &format!("{path}/{index}"),
262                    depth + 1,
263                    rows,
264                    named,
265                    fields,
266                )?;
267            }
268        }
269        Value::Object(object) => {
270            let type_value = object
271                .get("type")
272                .ok_or_else(|| Error::InvalidInput(format!("Avro schema {path} requires type")))?;
273            let kind = type_label(Some(type_value));
274            match type_value {
275                Value::Object(_) | Value::Array(_) => {
276                    walk_schema(
277                        type_value,
278                        &format!("{path}/type"),
279                        depth + 1,
280                        rows,
281                        named,
282                        fields,
283                    )?;
284                }
285                _ => {}
286            }
287            match kind.as_str() {
288                "record" | "error" | "request" => {
289                    let name = required_string(object, "name")?;
290                    *named = named.saturating_add(1);
291                    push_row(
292                        rows,
293                        vec![truncate(path), kind.clone(), format!("name {name}")],
294                    )?;
295                    let field_values =
296                        object
297                            .get("fields")
298                            .and_then(Value::as_array)
299                            .ok_or_else(|| {
300                                Error::InvalidInput(format!(
301                                    "Avro {kind} {name} requires fields array"
302                                ))
303                            })?;
304                    for field in field_values {
305                        let field_object = field.as_object().ok_or_else(|| {
306                            Error::InvalidInput(format!("Avro field in {name} must be an object"))
307                        })?;
308                        let field_name = required_string(field_object, "name")?;
309                        let field_type = field_object.get("type").ok_or_else(|| {
310                            Error::InvalidInput(format!("Avro field {field_name} requires type"))
311                        })?;
312                        *fields = fields.saturating_add(1);
313                        let field_path =
314                            format!("{}/fields/{field_name}", path.trim_end_matches('/'));
315                        push_row(
316                            rows,
317                            vec![
318                                truncate(&field_path),
319                                "field".into(),
320                                type_label(Some(field_type)),
321                            ],
322                        )?;
323                        walk_schema(
324                            field_type,
325                            &format!("{field_path}/type"),
326                            depth + 1,
327                            rows,
328                            named,
329                            fields,
330                        )?;
331                    }
332                }
333                "enum" => {
334                    let name = required_string(object, "name")?;
335                    let symbols =
336                        object
337                            .get("symbols")
338                            .and_then(Value::as_array)
339                            .ok_or_else(|| {
340                                Error::InvalidInput(format!(
341                                    "Avro enum {name} requires symbols array"
342                                ))
343                            })?;
344                    *named = named.saturating_add(1);
345                    push_row(
346                        rows,
347                        vec![
348                            truncate(path),
349                            "enum".into(),
350                            format!("name {name}; {} symbols", symbols.len()),
351                        ],
352                    )?;
353                }
354                "fixed" => {
355                    let name = required_string(object, "name")?;
356                    let size = object.get("size").and_then(Value::as_u64).ok_or_else(|| {
357                        Error::InvalidInput(format!("Avro fixed {name} requires unsigned size"))
358                    })?;
359                    *named = named.saturating_add(1);
360                    push_row(
361                        rows,
362                        vec![
363                            truncate(path),
364                            "fixed".into(),
365                            format!("name {name}; {size} bytes"),
366                        ],
367                    )?;
368                }
369                "array" => {
370                    let items = object.get("items").ok_or_else(|| {
371                        Error::InvalidInput(format!("Avro array {path} requires items"))
372                    })?;
373                    push_row(
374                        rows,
375                        vec![truncate(path), "array".into(), type_label(Some(items))],
376                    )?;
377                    walk_schema(
378                        items,
379                        &format!("{path}/items"),
380                        depth + 1,
381                        rows,
382                        named,
383                        fields,
384                    )?;
385                }
386                "map" => {
387                    let values = object.get("values").ok_or_else(|| {
388                        Error::InvalidInput(format!("Avro map {path} requires values"))
389                    })?;
390                    push_row(
391                        rows,
392                        vec![truncate(path), "map".into(), type_label(Some(values))],
393                    )?;
394                    walk_schema(
395                        values,
396                        &format!("{path}/values"),
397                        depth + 1,
398                        rows,
399                        named,
400                        fields,
401                    )?;
402                }
403                _ => {
404                    push_row(rows, vec![truncate(path), "schema".into(), kind])?;
405                }
406            }
407        }
408        _ => {
409            return Err(Error::InvalidInput(format!(
410                "Avro schema {path} must be a string, object or union array"
411            )));
412        }
413    }
414    Ok(())
415}
416
417fn push_row(rows: &mut Vec<Vec<String>>, row: Vec<String>) -> Result<()> {
418    if rows.len() >= MAX_AVRO_ROWS {
419        return Err(Error::LimitExceeded(format!(
420            "Avro schema rows exceed {MAX_AVRO_ROWS}"
421        )));
422    }
423    rows.push(row);
424    Ok(())
425}
426
427fn required_string<'a>(object: &'a serde_json::Map<String, Value>, key: &str) -> Result<&'a str> {
428    let value = object
429        .get(key)
430        .and_then(Value::as_str)
431        .ok_or_else(|| Error::InvalidInput(format!("Avro schema requires string {key}")))?;
432    if value.is_empty() {
433        return Err(Error::InvalidInput(format!(
434            "Avro schema {key} must not be empty"
435        )));
436    }
437    Ok(value)
438}
439
440fn type_label(value: Option<&Value>) -> String {
441    match value {
442        Some(Value::String(value)) => truncate(value),
443        Some(Value::Array(values)) => format!("union({})", values.len()),
444        Some(Value::Object(object)) => object
445            .get("type")
446            .and_then(Value::as_str)
447            .map_or_else(|| "object".into(), truncate),
448        Some(_) => "invalid".into(),
449        None => "—".into(),
450    }
451}
452
453fn truncate(value: &str) -> String {
454    if value.len() <= MAX_AVRO_DISPLAY_BYTES {
455        return value.to_owned();
456    }
457    let mut end = MAX_AVRO_DISPLAY_BYTES;
458    while !value.is_char_boundary(end) {
459        end -= 1;
460    }
461    format!("{}…", &value[..end])
462}
463
464fn count_values(value: &Value, depth: usize, count: &mut usize) -> Result<()> {
465    if depth > MAX_AVRO_DEPTH {
466        return Err(Error::LimitExceeded(format!(
467            "Avro JSON nesting exceeds {MAX_AVRO_DEPTH} levels"
468        )));
469    }
470    *count = count.saturating_add(1);
471    if *count > MAX_AVRO_VALUES {
472        return Err(Error::LimitExceeded(format!(
473            "Avro JSON contains more than {MAX_AVRO_VALUES} values"
474        )));
475    }
476    match value {
477        Value::Array(values) => {
478            for item in values {
479                count_values(item, depth + 1, count)?;
480            }
481        }
482        Value::Object(map) => {
483            for item in map.values() {
484                count_values(item, depth + 1, count)?;
485            }
486        }
487        Value::String(value) if value.len() > MAX_AVRO_STRING_BYTES => {
488            return Err(Error::LimitExceeded(format!(
489                "Avro JSON string exceeds {MAX_AVRO_STRING_BYTES} bytes"
490            )));
491        }
492        _ => {}
493    }
494    Ok(())
495}
496
497fn preflight_depth(text: &str) -> Result<()> {
498    let mut depth = 0usize;
499    let mut quoted = false;
500    let mut escaped = false;
501    for byte in text.bytes() {
502        if quoted {
503            if escaped {
504                escaped = false;
505            } else if byte == b'\\' {
506                escaped = true;
507            } else if byte == b'"' {
508                quoted = false;
509            }
510            continue;
511        }
512        match byte {
513            b'"' => quoted = true,
514            b'{' | b'[' => {
515                depth += 1;
516                if depth > MAX_AVRO_DEPTH {
517                    return Err(Error::LimitExceeded(format!(
518                        "Avro JSON nesting exceeds {MAX_AVRO_DEPTH} levels"
519                    )));
520                }
521            }
522            b'}' | b']' => depth = depth.saturating_sub(1),
523            _ => {}
524        }
525    }
526    Ok(())
527}
528
529#[cfg(test)]
530mod tests {
531    use super::*;
532
533    #[test]
534    fn recognizes_avro_schema_and_protocol() {
535        assert!(looks_like_prefix(
536            br#"{"type":"record","name":"User","fields":[{"name":"id","type":"long"}]}"#
537        ));
538        assert!(looks_like_prefix(br#"{"protocol":"Ping","messages":{}}"#));
539        assert!(!looks_like_prefix(br#"{"type":"object","properties":{}}"#));
540    }
541
542    #[test]
543    fn summarizes_nested_schema_without_defaults_or_docs() {
544        let (table, metadata, warnings) = parse(
545            r#"{"type":"record","name":"User","namespace":"example","doc":"secret docs","fields":[{"name":"id","type":"long","default":7},{"name":"profile","type":{"type":"record","name":"Profile","fields":[{"name":"email","type":"string"}]}},{"name":"tags","type":{"type":"array","items":"string"}}]}"#,
546        )
547        .unwrap();
548        assert!(metadata.contains("Root schema: User"));
549        assert!(table.rows.iter().any(|row| row[0].contains("profile")));
550        assert!(
551            !table
552                .rows
553                .iter()
554                .flatten()
555                .any(|value| value.contains("secret") || value.contains("7"))
556        );
557        assert!(warnings.iter().any(|warning| warning.contains("defaults")));
558    }
559}
document_svg/document/avro.rs

document_svg/document/
avro.rs
