use crate::error::{CliError, CliResult};
use serde::{Deserialize, Serialize};
use std::path::Path;
const EMBEDDED_INDEX: &str = include_str!("../connectors/registry.json");
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConnectorEntry {
pub name: String,
pub kind: String,
#[serde(default = "default_true")]
pub verified: bool,
#[serde(default)]
pub description: String,
#[serde(rename = "crate", default)]
pub krate: Option<String>,
#[serde(default)]
pub feature: Option<String>,
#[serde(default)]
pub keywords: Vec<String>,
#[serde(default)]
pub core_compat: Option<String>,
}
fn default_true() -> bool {
true
}
impl ConnectorEntry {
pub fn crate_name(&self) -> String {
self.krate
.clone()
.unwrap_or_else(|| format!("faucet-{}-{}", self.kind, self.name))
}
pub fn feature_flag(&self) -> String {
self.feature
.clone()
.unwrap_or_else(|| format!("{}-{}", self.kind, self.name))
}
pub fn matches(&self, lowered_term: &str) -> bool {
self.name.to_lowercase().contains(lowered_term)
|| self.description.to_lowercase().contains(lowered_term)
|| self.crate_name().to_lowercase().contains(lowered_term)
|| self
.keywords
.iter()
.any(|k| k.to_lowercase().contains(lowered_term))
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RegistryIndex {
#[serde(default = "default_version")]
pub version: u32,
pub connectors: Vec<ConnectorEntry>,
}
fn default_version() -> u32 {
1
}
impl RegistryIndex {
pub fn embedded() -> Self {
serde_json::from_str(EMBEDDED_INDEX)
.expect("embedded cli/connectors/registry.json is valid")
}
pub fn load(path: Option<&Path>) -> CliResult<Self> {
match path {
None => Ok(Self::embedded()),
Some(p) => {
let text = std::fs::read_to_string(p)?;
serde_json::from_str(&text).map_err(|e| {
CliError::Config(format!("invalid connector index `{}`: {e}", p.display()))
})
}
}
}
pub fn search(&self, term: &str) -> Vec<&ConnectorEntry> {
let t = term.to_lowercase();
let mut hits: Vec<&ConnectorEntry> =
self.connectors.iter().filter(|c| c.matches(&t)).collect();
hits.sort_by(|a, b| {
(a.kind.as_str(), a.name.as_str()).cmp(&(b.kind.as_str(), b.name.as_str()))
});
hits
}
pub fn find(&self, name: &str, kind: Option<&str>) -> Vec<&ConnectorEntry> {
self.connectors
.iter()
.filter(|c| c.name == name && kind.map(|k| k == c.kind).unwrap_or(true))
.collect()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum InstallRecipe {
AlreadyAvailable { feature: String },
CargoInstall { feature: String },
CustomBinary { krate: String, feature: String },
}
pub fn install_recipe(entry: &ConnectorEntry, compiled_in: bool) -> InstallRecipe {
let feature = entry.feature_flag();
if !entry.verified {
return InstallRecipe::CustomBinary {
krate: entry.crate_name(),
feature,
};
}
if compiled_in {
InstallRecipe::AlreadyAvailable { feature }
} else {
InstallRecipe::CargoInstall { feature }
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn embedded_index_parses_and_is_non_empty() {
let idx = RegistryIndex::embedded();
assert_eq!(idx.version, 1);
assert!(idx.connectors.len() > 30, "expected the built-in catalog");
let kafka = idx
.find("kafka", Some("source"))
.into_iter()
.next()
.unwrap();
assert_eq!(kafka.crate_name(), "faucet-source-kafka");
assert_eq!(kafka.feature_flag(), "source-kafka");
assert!(kafka.verified);
}
#[test]
fn search_is_case_insensitive_over_fields() {
let idx = RegistryIndex::embedded();
let hits = idx.search("KAFKA");
assert!(hits.iter().any(|c| c.name == "kafka" && c.kind == "source"));
assert!(hits.iter().any(|c| c.name == "kafka" && c.kind == "sink"));
assert!(!idx.search("cdc").is_empty());
assert!(idx.search("definitely-not-a-connector").is_empty());
}
#[test]
fn install_recipe_for_builtin_compiled_and_not() {
let idx = RegistryIndex::embedded();
let entry = idx
.find("bigquery", Some("sink"))
.into_iter()
.next()
.unwrap();
assert_eq!(
install_recipe(entry, true),
InstallRecipe::AlreadyAvailable {
feature: "sink-bigquery".into()
}
);
assert_eq!(
install_recipe(entry, false),
InstallRecipe::CargoInstall {
feature: "sink-bigquery".into()
}
);
}
#[test]
fn install_recipe_for_community_is_custom_binary() {
let entry = ConnectorEntry {
name: "acme".into(),
kind: "source".into(),
verified: false,
description: "community".into(),
krate: None,
feature: None,
keywords: vec![],
core_compat: None,
};
assert_eq!(
install_recipe(&entry, false),
InstallRecipe::CustomBinary {
krate: "faucet-source-acme".into(),
feature: "source-acme".into()
}
);
}
#[test]
fn load_from_path_parses_custom_index() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("idx.json");
std::fs::write(
&p,
r#"{"version":1,"connectors":[{"name":"acme","kind":"source","verified":false,"description":"Acme","crate":"acme-faucet"}]}"#,
)
.unwrap();
let idx = RegistryIndex::load(Some(&p)).unwrap();
let e = &idx.connectors[0];
assert_eq!(e.crate_name(), "acme-faucet"); assert_eq!(e.feature_flag(), "source-acme"); assert!(!e.verified);
}
#[test]
fn index_covers_every_compiled_builtin() {
let idx = RegistryIndex::embedded();
for k in crate::registry::source_kinds() {
assert!(
idx.find(k, Some("source")).iter().any(|c| c.verified),
"built-in source `{k}` is missing a verified entry in cli/connectors/registry.json"
);
}
for k in crate::registry::sink_kinds() {
assert!(
idx.find(k, Some("sink")).iter().any(|c| c.verified),
"built-in sink `{k}` is missing a verified entry in cli/connectors/registry.json"
);
}
}
#[test]
fn load_bad_index_errors() {
let dir = tempfile::tempdir().unwrap();
let p = dir.path().join("bad.json");
std::fs::write(&p, "{ not json").unwrap();
assert!(matches!(
RegistryIndex::load(Some(&p)),
Err(CliError::Config(_))
));
}
}