1use crate::error::{CliError, CliResult};
13use serde::{Deserialize, Serialize};
14use std::path::Path;
15
16const EMBEDDED_INDEX: &str = include_str!("../connectors/registry.json");
19
20#[derive(Debug, Clone, Serialize, Deserialize)]
22pub struct ConnectorEntry {
23 pub name: String,
25 pub kind: String,
27 #[serde(default = "default_true")]
30 pub verified: bool,
31 #[serde(default)]
33 pub description: String,
34 #[serde(rename = "crate", default)]
36 pub krate: Option<String>,
37 #[serde(default)]
41 pub feature: Option<String>,
42 #[serde(default)]
44 pub keywords: Vec<String>,
45 #[serde(default)]
47 pub core_compat: Option<String>,
48 #[serde(default)]
54 pub tier: Option<String>,
55}
56
57fn default_true() -> bool {
58 true
59}
60
61impl ConnectorEntry {
62 pub fn crate_name(&self) -> String {
64 self.krate
65 .clone()
66 .unwrap_or_else(|| format!("faucet-{}-{}", self.kind, self.name))
67 }
68
69 pub fn feature_flag(&self) -> String {
71 self.feature
72 .clone()
73 .unwrap_or_else(|| format!("{}-{}", self.kind, self.name))
74 }
75
76 pub fn matches(&self, lowered_term: &str) -> bool {
79 self.name.to_lowercase().contains(lowered_term)
80 || self.description.to_lowercase().contains(lowered_term)
81 || self.crate_name().to_lowercase().contains(lowered_term)
82 || self
83 .keywords
84 .iter()
85 .any(|k| k.to_lowercase().contains(lowered_term))
86 }
87}
88
89#[derive(Debug, Clone, Serialize, Deserialize)]
91pub struct RegistryIndex {
92 #[serde(default = "default_version")]
93 pub version: u32,
94 pub connectors: Vec<ConnectorEntry>,
95}
96
97fn default_version() -> u32 {
98 1
99}
100
101impl RegistryIndex {
102 pub fn embedded() -> Self {
104 serde_json::from_str(EMBEDDED_INDEX)
105 .expect("embedded cli/connectors/registry.json is valid")
106 }
107
108 pub fn load(path: Option<&Path>) -> CliResult<Self> {
110 match path {
111 None => Ok(Self::embedded()),
112 Some(p) => {
113 let text = std::fs::read_to_string(p)?;
114 serde_json::from_str(&text).map_err(|e| {
115 CliError::Config(format!("invalid connector index `{}`: {e}", p.display()))
116 })
117 }
118 }
119 }
120
121 pub fn search(&self, term: &str) -> Vec<&ConnectorEntry> {
123 let t = term.to_lowercase();
124 let mut hits: Vec<&ConnectorEntry> =
125 self.connectors.iter().filter(|c| c.matches(&t)).collect();
126 hits.sort_by(|a, b| {
127 (a.kind.as_str(), a.name.as_str()).cmp(&(b.kind.as_str(), b.name.as_str()))
128 });
129 hits
130 }
131
132 pub fn find(&self, name: &str, kind: Option<&str>) -> Vec<&ConnectorEntry> {
134 self.connectors
135 .iter()
136 .filter(|c| c.name == name && kind.map(|k| k == c.kind).unwrap_or(true))
137 .collect()
138 }
139}
140
141#[derive(Debug, Clone, PartialEq, Eq)]
143pub enum InstallRecipe {
144 AlreadyAvailable { feature: String },
146 CargoInstall { feature: String },
148 CustomBinary { krate: String, feature: String },
150}
151
152pub fn install_recipe(entry: &ConnectorEntry, compiled_in: bool) -> InstallRecipe {
155 let feature = entry.feature_flag();
156 if !entry.verified {
157 return InstallRecipe::CustomBinary {
158 krate: entry.crate_name(),
159 feature,
160 };
161 }
162 if compiled_in {
163 InstallRecipe::AlreadyAvailable { feature }
164 } else {
165 InstallRecipe::CargoInstall { feature }
166 }
167}
168
169#[cfg(test)]
170mod tests {
171 use super::*;
172
173 #[test]
174 fn embedded_index_parses_and_is_non_empty() {
175 let idx = RegistryIndex::embedded();
176 assert_eq!(idx.version, 1);
177 assert!(idx.connectors.len() > 30, "expected the built-in catalog");
178 let kafka = idx
180 .find("kafka", Some("source"))
181 .into_iter()
182 .next()
183 .unwrap();
184 assert_eq!(kafka.crate_name(), "faucet-source-kafka");
185 assert_eq!(kafka.feature_flag(), "source-kafka");
186 assert!(kafka.verified);
187 }
188
189 #[test]
190 fn search_is_case_insensitive_over_fields() {
191 let idx = RegistryIndex::embedded();
192 let hits = idx.search("KAFKA");
193 assert!(hits.iter().any(|c| c.name == "kafka" && c.kind == "source"));
194 assert!(hits.iter().any(|c| c.name == "kafka" && c.kind == "sink"));
195 assert!(!idx.search("cdc").is_empty());
197 assert!(idx.search("definitely-not-a-connector").is_empty());
199 }
200
201 #[test]
202 fn install_recipe_for_builtin_compiled_and_not() {
203 let idx = RegistryIndex::embedded();
204 let entry = idx
205 .find("bigquery", Some("sink"))
206 .into_iter()
207 .next()
208 .unwrap();
209 assert_eq!(
210 install_recipe(entry, true),
211 InstallRecipe::AlreadyAvailable {
212 feature: "sink-bigquery".into()
213 }
214 );
215 assert_eq!(
216 install_recipe(entry, false),
217 InstallRecipe::CargoInstall {
218 feature: "sink-bigquery".into()
219 }
220 );
221 }
222
223 #[test]
224 fn install_recipe_for_community_is_custom_binary() {
225 let entry = ConnectorEntry {
226 name: "acme".into(),
227 kind: "source".into(),
228 verified: false,
229 description: "community".into(),
230 krate: None,
231 feature: None,
232 keywords: vec![],
233 core_compat: None,
234 tier: None,
235 };
236 assert_eq!(
237 install_recipe(&entry, false),
238 InstallRecipe::CustomBinary {
239 krate: "faucet-source-acme".into(),
240 feature: "source-acme".into()
241 }
242 );
243 }
244
245 #[test]
246 fn load_from_path_parses_custom_index() {
247 let dir = tempfile::tempdir().unwrap();
248 let p = dir.path().join("idx.json");
249 std::fs::write(
250 &p,
251 r#"{"version":1,"connectors":[{"name":"acme","kind":"source","verified":false,"description":"Acme","crate":"acme-faucet"}]}"#,
252 )
253 .unwrap();
254 let idx = RegistryIndex::load(Some(&p)).unwrap();
255 let e = &idx.connectors[0];
256 assert_eq!(e.crate_name(), "acme-faucet"); assert_eq!(e.feature_flag(), "source-acme"); assert!(!e.verified);
259 }
260
261 #[test]
265 fn index_covers_every_compiled_builtin() {
266 let idx = RegistryIndex::embedded();
267 for k in crate::registry::source_kinds() {
268 assert!(
269 idx.find(k, Some("source")).iter().any(|c| c.verified),
270 "built-in source `{k}` is missing a verified entry in cli/connectors/registry.json"
271 );
272 }
273 for k in crate::registry::sink_kinds() {
274 assert!(
275 idx.find(k, Some("sink")).iter().any(|c| c.verified),
276 "built-in sink `{k}` is missing a verified entry in cli/connectors/registry.json"
277 );
278 }
279 }
280
281 #[test]
285 fn builtin_tiers_match_conformance() {
286 let idx = RegistryIndex::embedded();
287 for k in crate::registry::source_kinds() {
288 let entry = idx
289 .find(k, Some("source"))
290 .into_iter()
291 .find(|c| c.verified)
292 .unwrap_or_else(|| panic!("built-in source `{k}` missing a verified entry"));
293 let want = crate::conformance::tier_for(k, true).as_str();
294 assert_eq!(
295 entry.tier.as_deref(),
296 Some(want),
297 "source `{k}` registry tier {:?} != computed `{want}`",
298 entry.tier
299 );
300 }
301 for k in crate::registry::sink_kinds() {
302 let entry = idx
303 .find(k, Some("sink"))
304 .into_iter()
305 .find(|c| c.verified)
306 .unwrap_or_else(|| panic!("built-in sink `{k}` missing a verified entry"));
307 let want = crate::conformance::tier_for(k, false).as_str();
308 assert_eq!(
309 entry.tier.as_deref(),
310 Some(want),
311 "sink `{k}` registry tier {:?} != computed `{want}`",
312 entry.tier
313 );
314 }
315 }
316
317 #[test]
318 fn load_bad_index_errors() {
319 let dir = tempfile::tempdir().unwrap();
320 let p = dir.path().join("bad.json");
321 std::fs::write(&p, "{ not json").unwrap();
322 assert!(matches!(
323 RegistryIndex::load(Some(&p)),
324 Err(CliError::Config(_))
325 ));
326 }
327}