zenkey_fleet/
registry.rs

1//! Registry-slice sets (RFC 08 §6): one type over both sources.
2//!
3//! A slice is a slice regardless of where it was read — a producer's served
4//! `introspect` reply off the live bus, or a local `registry/*.toml` file.
5//! [`SliceSet`] carries them uniformly (with an optional on-disk cache so
6//! repeated invocations and shell completion answer instantly), and exposes
7//! the subject-refinement lookups every renderer needs.
8
9use std::path::{Path, PathBuf};
10use std::time::Duration;
11
12use anyhow::{Result, anyhow};
13use zenkey::{RegistrySlice, parse_slice};
14use zenoh::Session;
15
16/// A set of registry slices, indexed by producer/service base name.
17#[derive(Debug, Clone, Default)]
18pub struct SliceSet {
19    slices: Vec<RegistrySlice>,
20    /// The raw TOML per slice, kept for the disk cache (slices do not
21    /// re-serialize; the served text is the artifact).
22    raw: Vec<String>,
23}
24
25impl SliceSet {
26    /// Load from local `registry/*.toml` dirs — the offline source. What a
27    /// checked-out application *declares*. (`types.toml` is the type table,
28    /// not a slice — skipped.)
29    pub fn from_dirs(dirs: &[PathBuf]) -> Result<SliceSet> {
30        let mut set = SliceSet::default();
31        for dir in dirs {
32            let mut paths: Vec<_> = std::fs::read_dir(dir)
33                .map_err(|e| anyhow!("--registry {}: {e}", dir.display()))?
34                .filter_map(|e| e.ok().map(|e| e.path()))
35                .filter(|p| p.extension().is_some_and(|e| e == "toml"))
36                .filter(|p| p.file_name().is_none_or(|n| n != "types.toml"))
37                .collect();
38            paths.sort();
39            for path in paths {
40                let text = std::fs::read_to_string(&path)
41                    .map_err(|e| anyhow!("{}: {e}", path.display()))?;
42                let slice = parse_slice(&text).map_err(|e| {
43                    anyhow!(
44                        "{}: does not parse as a registry slice: {e}",
45                        path.display()
46                    )
47                })?;
48                set.push(slice, text);
49            }
50        }
51        Ok(set)
52    }
53
54    /// Discover every live producer's served slice from the bus
55    /// ([`crate::query::fleet_registry`]).
56    pub async fn from_bus(session: &Session, base: &str, timeout: Duration) -> Result<SliceSet> {
57        let pairs = crate::query::fleet_registry_raw(session, base, timeout).await?;
58        let mut set = SliceSet::default();
59        for (slice, raw) in pairs {
60            set.push(slice, raw);
61        }
62        Ok(set)
63    }
64
65    fn push(&mut self, slice: RegistrySlice, raw: String) {
66        // One slice per base name; last one wins (a fleet mid-rollout serves
67        // several versions — the newest reply is as good a pick as any, and
68        // `doctor` is where disagreement is *reported*).
69        if let Some(i) = self.slices.iter().position(|s| s.name == slice.name) {
70            self.slices[i] = slice;
71            self.raw[i] = raw;
72        } else {
73            self.slices.push(slice);
74            self.raw.push(raw);
75        }
76    }
77
78    pub fn slices(&self) -> &[RegistrySlice] {
79        &self.slices
80    }
81
82    pub fn get(&self, name: &str) -> Option<&RegistrySlice> {
83        self.slices.iter().find(|s| s.name == name)
84    }
85
86    /// The slice declaring a service origin (`@catalog`) — service keys have
87    /// no producer chunk, so refinement resolves through this.
88    pub fn by_service_origin(&self, origin: &str) -> Option<&RegistrySlice> {
89        self.slices
90            .iter()
91            .find(|s| s.service_origin.as_deref() == Some(origin))
92    }
93
94    /// Refine a subject tail against one producer's slice: the matching
95    /// subject declaration plus its named variable bindings.
96    pub fn refine<'s>(
97        &'s self,
98        producer: &str,
99        class: &str,
100        tail: &[&str],
101    ) -> Option<(&'s zenkey::slice::SubjectDecl, Vec<(String, String)>)> {
102        let slice = self.get(producer)?;
103        // Precedence-ordered via the shared matcher (issue #7): collect the
104        // class's patterns and let best_match pick — same order the codegen
105        // compiles.
106        let candidates: Vec<(usize, zenkey::pattern::SubjectPattern)> = slice
107            .subjects
108            .iter()
109            .enumerate()
110            .filter(|(_, s)| s.class == class)
111            .filter_map(|(i, s)| {
112                zenkey::pattern::SubjectPattern::parse(&s.path)
113                    .ok()
114                    .map(|p| (i, p))
115            })
116            .collect();
117        let patterns: Vec<zenkey::pattern::SubjectPattern> =
118            candidates.iter().map(|(_, p)| p.clone()).collect();
119        let (winner, binds) = zenkey::pattern::best_match(&patterns, tail)?;
120        let (subject_idx, _) = candidates[winner];
121        Some((
122            &slice.subjects[subject_idx],
123            binds.into_iter().map(|(n, v)| (n.to_string(), v)).collect(),
124        ))
125    }
126
127    /// Build from already-parsed slices (no raw TOML retained — such a set
128    /// is skipped by `write_cache`).
129    pub fn from_slices(slices: Vec<RegistrySlice>) -> SliceSet {
130        let raw = vec![String::new(); slices.len()];
131        SliceSet { slices, raw }
132    }
133
134    /// Write the raw slice TOMLs to a cache dir (one file per producer).
135    /// Repeated invocations and dynamic shell completion read this instead
136    /// of round-tripping the bus.
137    pub fn write_cache(&self, dir: &Path) -> Result<()> {
138        std::fs::create_dir_all(dir)?;
139        for (slice, raw) in self.slices.iter().zip(&self.raw) {
140            if raw.is_empty() {
141                continue; // from_slices sets: nothing faithful to persist
142            }
143            std::fs::write(dir.join(format!("{}.toml", slice.name)), raw)?;
144        }
145        Ok(())
146    }
147
148    /// Read a previously written cache dir. Same forgiving posture as
149    /// `from_dirs`, but a missing dir is an empty set, not an error.
150    pub fn read_cache(dir: &Path) -> SliceSet {
151        if !dir.is_dir() {
152            return SliceSet::default();
153        }
154        SliceSet::from_dirs(&[dir.to_path_buf()]).unwrap_or_default()
155    }
156}
157
158#[cfg(test)]
159impl SliceSet {
160    /// Test constructor from one slice TOML (crate-internal).
161    pub(crate) fn from_toml_for_tests(toml: &str) -> SliceSet {
162        let mut set = SliceSet::default();
163        set.push(parse_slice(toml).unwrap(), toml.to_string());
164        set
165    }
166}
167
168#[cfg(test)]
169mod tests {
170    use super::*;
171
172    const A: &str = r#"
173        [registry]
174        version = "1.0"
175        app = "t"
176        convention = 1
177        [producer]
178        name = "alpha"
179        [[subject]]
180        path = "flow/{q}"
181        class = "telemetry"
182        type = "Point"
183        [[subject]]
184        path = "flow/special"
185        class = "telemetry"
186        type = "Special"
187    "#;
188
189    #[test]
190    fn refine_uses_shared_precedence() {
191        let mut set = SliceSet::default();
192        set.push(parse_slice(A).unwrap(), A.to_string());
193        // Literal beats {var} — the shared best_match ordering.
194        let (s, binds) = set
195            .refine("alpha", "telemetry", &["flow", "special"])
196            .unwrap();
197        assert_eq!(s.type_name, "Special");
198        assert!(binds.is_empty());
199        let (s, binds) = set.refine("alpha", "telemetry", &["flow", "p95"]).unwrap();
200        assert_eq!(s.type_name, "Point");
201        assert_eq!(binds, vec![("q".to_string(), "p95".to_string())]);
202        assert!(set.refine("alpha", "state", &["flow", "p95"]).is_none());
203    }
204
205    #[test]
206    fn cache_round_trips_and_last_slice_wins() {
207        let mut set = SliceSet::default();
208        set.push(parse_slice(A).unwrap(), A.to_string());
209        // A newer slice for the same producer replaces, never duplicates.
210        set.push(parse_slice(A).unwrap(), A.to_string());
211        assert_eq!(set.slices().len(), 1);
212
213        let dir = std::env::temp_dir().join(format!("zenkey-fleet-cache-{}", std::process::id()));
214        let _ = std::fs::remove_dir_all(&dir);
215        set.write_cache(&dir).unwrap();
216        let back = SliceSet::read_cache(&dir);
217        assert_eq!(back.slices().len(), 1);
218        assert_eq!(back.get("alpha").unwrap().subjects.len(), 2);
219        let _ = std::fs::remove_dir_all(&dir);
220        // Missing dir: empty set, not an error.
221        assert!(
222            SliceSet::read_cache(Path::new("/nonexistent-zkf"))
223                .slices()
224                .is_empty()
225        );
226    }
227}
zenkey_fleet/registry.rs

zenkey_fleet/
registry.rs
