krabka_protocol_codegen/
ir.rs

1use std::{fs, io, path::Path};
2
3use serde::{Deserialize, de};
4use thiserror::Error;
5
6/// One parsed Kafka message schema.
7#[derive(Debug, Deserialize)]
8#[serde(rename_all = "camelCase")]
9pub struct MessageSpec {
10    pub name: String,
11    #[serde(rename = "type")]
12    pub message_type: MessageType,
13    #[serde(default)]
14    pub api_key: Option<i16>,
15    pub valid_versions: VersionRange,
16    /// Kafka's `latestVersionUnstable`. When it is true, the highest version in
17    /// `valid_versions` is still in development. A Kafka broker advertises that
18    /// version only with `unstable.api.versions.enable`, and a Kafka client
19    /// never picks it. The key is optional in the schema, and Kafka reads a
20    /// missing key as false.
21    #[serde(default)]
22    pub latest_version_unstable: bool,
23    #[serde(default)]
24    pub flexible_versions: FlexibleVersions,
25    #[serde(default)]
26    pub fields: Vec<FieldSpec>,
27    #[serde(default)]
28    pub common_structs: Vec<CommonStruct>,
29    /// krabka-internal RPC with no upstream Apache Kafka equivalent, for
30    /// example the KIP-966 `GetReplicaLogInfo` controller↔broker RPC. The JVM
31    /// differential sweep excludes such messages, because the oracle has no
32    /// matching message class and krabka owns their wire shape outright.
33    #[serde(default)]
34    pub internal: bool,
35}
36
37#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq)]
38#[serde(rename_all = "lowercase")]
39pub enum MessageType {
40    Request,
41    Response,
42    Header,
43    Data,
44}
45
46#[derive(Debug, Deserialize)]
47#[serde(rename_all = "camelCase")]
48pub struct FieldSpec {
49    pub name: String,
50    #[serde(rename = "type")]
51    pub field_type: String,
52    pub versions: VersionRange,
53    #[serde(default)]
54    pub nullable_versions: Option<VersionRange>,
55    #[serde(default)]
56    pub tagged_versions: Option<VersionRange>,
57    #[serde(default)]
58    pub tag: Option<u32>,
59    #[serde(default)]
60    pub fields: Vec<Self>,
61    #[serde(default)]
62    pub default: Option<serde_json::Value>,
63    #[serde(default = "default_entity_type")]
64    pub entity_type: String,
65    #[serde(default)]
66    pub map_key: bool,
67    #[serde(default)]
68    pub about: String,
69    /// Per-field flexible-version override. `Some(FlexibleVersions::None)`
70    /// means this field always uses legacy (non-compact) encoding, even in a
71    /// flexible-version message. `None` means the field inherits from the
72    /// message-level `flexible_versions`, which is the common case.
73    #[serde(default)]
74    pub flexible_versions: Option<FlexibleVersions>,
75}
76
77fn default_entity_type() -> String {
78    String::new()
79}
80
81#[derive(Debug, Deserialize)]
82pub struct CommonStruct {
83    pub name: String,
84    pub versions: VersionRange,
85    pub fields: Vec<FieldSpec>,
86}
87
88impl MessageSpec {
89    /// The highest version that a client can send, which is Kafka's
90    /// `ApiMessageType.highestSupportedVersion(false)`.
91    ///
92    /// It is one less than `valid_versions.max` when the latest version is
93    /// unstable. The result is less than `valid_versions.min` when the only
94    /// version is unstable. Kafka reads that as "no enabled version".
95    #[must_use]
96    pub fn latest_stable_version(&self) -> i16 {
97        if self.latest_version_unstable {
98            self.valid_versions.max.saturating_sub(1)
99        } else {
100            self.valid_versions.max
101        }
102    }
103}
104
105/// `"0+"`, `"3+"`, `"0-2"`, `"none"`, `"4"` etc.
106#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
107pub struct VersionRange {
108    pub min: i16,
109    pub max: i16, // inclusive; i16::MAX represents `+` (open-ended)
110}
111
112impl<'de> Deserialize<'de> for VersionRange {
113    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
114        let s = String::deserialize(d)?;
115        // "none" means the version range is empty (e.g. removed messages).
116        if s == "none" {
117            return Ok(Self {
118                min: i16::MAX,
119                max: i16::MIN,
120            });
121        }
122        parse_version_range(&s).map_err(de::Error::custom)
123    }
124}
125
126fn parse_version_range(s: &str) -> Result<VersionRange, String> {
127    if let Some(rest) = s.strip_suffix('+') {
128        let min: i16 = rest
129            .parse()
130            .map_err(|e| format!("bad version `{s}`: {e}"))?;
131        return Ok(VersionRange { min, max: i16::MAX });
132    }
133    if let Some((lo, hi)) = s.split_once('-') {
134        let min: i16 = lo.parse().map_err(|e| format!("bad version `{s}`: {e}"))?;
135        let max: i16 = hi.parse().map_err(|e| format!("bad version `{s}`: {e}"))?;
136        return Ok(VersionRange { min, max });
137    }
138    let single: i16 = s.parse().map_err(|e| format!("bad version `{s}`: {e}"))?;
139    Ok(VersionRange {
140        min: single,
141        max: single,
142    })
143}
144
145impl VersionRange {
146    #[must_use]
147    pub fn contains(&self, v: i16) -> bool {
148        v >= self.min && v <= self.max
149    }
150    #[must_use]
151    pub fn is_empty(&self) -> bool {
152        self.min > self.max
153    }
154}
155
156#[derive(Debug, Clone, Copy, Default)]
157pub enum FlexibleVersions {
158    #[default]
159    None,
160    Range(VersionRange),
161}
162
163impl<'de> Deserialize<'de> for FlexibleVersions {
164    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
165        let s = String::deserialize(d)?;
166        if s == "none" {
167            return Ok(Self::None);
168        }
169        Ok(Self::Range(
170            parse_version_range(&s).map_err(de::Error::custom)?,
171        ))
172    }
173}
174
175impl FlexibleVersions {
176    #[must_use]
177    pub fn is_flexible(&self, v: i16) -> bool {
178        match self {
179            Self::None => false,
180            Self::Range(r) => r.contains(v),
181        }
182    }
183}
184
185#[derive(Debug, Error)]
186pub enum IrError {
187    #[error("I/O: {0}")]
188    Io(#[from] io::Error),
189    #[error("parse {file}: {source}")]
190    Parse {
191        file: String,
192        #[source]
193        source: serde_json::Error,
194    },
195}
196
197/// Read every `*.json` file in `dir`, strip `//` comments, and parse each file
198/// as a `MessageSpec`.
199/// # Errors
200/// Returns an error when the schema model is invalid or generated Rust cannot be formatted or written.
201pub fn load_dir(dir: &Path) -> Result<Vec<MessageSpec>, IrError> {
202    let mut out = Vec::new();
203    for entry in fs::read_dir(dir)? {
204        let entry = entry?;
205        let path = entry.path();
206        if path.extension().and_then(|s| s.to_str()) != Some("json") {
207            continue;
208        }
209        let raw = fs::read_to_string(&path)?;
210        let stripped = strip_line_comments(&raw);
211        let spec: MessageSpec = serde_json::from_str(&stripped).map_err(|e| IrError::Parse {
212            file: path.display().to_string(),
213            source: e,
214        })?;
215        out.push(spec);
216    }
217    out.sort_by(|a, b| a.name.cmp(&b.name));
218    Ok(out)
219}
220
221/// Strip JavaScript-style `//` line comments. This function ignores `//` that
222/// appears inside a double-quoted JSON string, so string values such as URLs
223/// survive intact.
224fn strip_line_comments(src: &str) -> String {
225    let mut out = String::with_capacity(src.len());
226    for line in src.lines() {
227        let bytes = line.as_bytes();
228        let mut in_str = false;
229        let mut escaped = false;
230        let mut end = bytes.len();
231        for (i, &b) in bytes.iter().enumerate() {
232            if in_str {
233                if escaped {
234                    escaped = false;
235                } else if b == b'\\' {
236                    escaped = true;
237                } else if b == b'"' {
238                    in_str = false;
239                }
240            } else if b == b'"' {
241                in_str = true;
242            } else if b == b'/' && bytes.get(i + 1) == Some(&b'/') {
243                end = i;
244                break;
245            }
246        }
247        // `end` sits on an ASCII `/` or at the end of the line, so it is
248        // always a character boundary.
249        out.push_str(line.get(..end).unwrap_or(line));
250        out.push('\n');
251    }
252    out
253}
254
255#[cfg(test)]
256mod tests {
257    use assert2::assert;
258
259    use super::*;
260
261    #[test]
262    fn version_range_parsing() {
263        for (input, want) in [
264            (
265                "0+",
266                VersionRange {
267                    min: 0,
268                    max: i16::MAX,
269                },
270            ),
271            (
272                "3+",
273                VersionRange {
274                    min: 3,
275                    max: i16::MAX,
276                },
277            ),
278            ("0-2", VersionRange { min: 0, max: 2 }),
279            ("4", VersionRange { min: 4, max: 4 }),
280        ] {
281            assert!(parse_version_range(input).unwrap() == want);
282        }
283        assert!(parse_version_range("none").is_err()); // handled at call site
284    }
285
286    #[test]
287    fn comment_strip() {
288        let src = "{\n// hi\n  \"x\": 1 // trailing\n}";
289        let out = strip_line_comments(src);
290        assert!(out == "{\n\n  \"x\": 1 \n}\n");
291    }
292
293    #[test]
294    fn comment_strip_preserves_double_slash_in_string() {
295        let src = "{ \"default\": \"http://example.com\" } // tail";
296        let out = strip_line_comments(src);
297        assert!(out.contains("http://example.com"));
298        assert!(!out.contains("tail"));
299    }
300}
krabka_protocol_codegen/ir.rs

krabka_protocol_codegen/
ir.rs
