kernel/
adapter.rs

1//! What an adapter declares about itself: which port it implements, its kind
2//! string, and the shape of its options.
3//!
4//! An adapter owns its configuration (ADR 0001). It writes one struct, derives
5//! [`AdapterConfig`] on it, and from that single declaration the composition
6//! root gets everything it renders: the editor JSON schema, the Reference
7//! option tables, the designer's forms, and the environment variables that
8//! override the adapter's secrets. Nothing in the kernel or the config layer
9//! names the adapter.
10//!
11//! ```
12//! use kernel::{AdapterConfig, Options, Port, Secret};
13//! use schemars::JsonSchema;
14//! use serde::{Deserialize, Serialize};
15//!
16//! /// Where the documents go.
17//! #[derive(Debug, Serialize, Deserialize, JsonSchema)]
18//! #[serde(deny_unknown_fields)]
19//! struct DemoConfig {
20//!     /// The cluster URL.
21//!     url: Secret,
22//!     /// Documents per request.
23//!     #[serde(default = "default_batch")]
24//!     batch_size: u32,
25//! }
26//! fn default_batch() -> u32 { 1000 }
27//!
28//! impl AdapterConfig for DemoConfig {
29//!     const PORT: Port = Port::Sink;
30//!     const KIND: &'static str = "demo";
31//!     fn example() -> Self {
32//!         Self { url: Secret::Value("https://search:9200".into()), batch_size: 500 }
33//!     }
34//! }
35//!
36//! let description = DemoConfig::description();
37//! assert_eq!(description.kind, "demo");
38//! // Every `Secret` field gets an override variable, named per entry.
39//! let vars: Vec<String> = description.override_vars("primary").collect();
40//! assert_eq!(vars, ["PRIMARY_DEMO_URL"]);
41//!
42//! let mut options = Options::empty();
43//! options.insert("url", "https://other:9200");
44//! let config = DemoConfig::from_options(options).unwrap();
45//! assert_eq!(config.batch_size, 1000);
46//! ```
47
48use std::fmt;
49
50use schemars::{JsonSchema, Schema};
51use serde::de::DeserializeOwned;
52use serde::{Deserialize, Serialize};
53
54use crate::options::{Options, OptionsError};
55
56/// The three contracts the engine drives. Every adapter implements exactly one.
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, JsonSchema)]
58#[serde(rename_all = "snake_case")]
59pub enum Port {
60    Source,
61    Stream,
62    Sink,
63}
64
65impl Port {
66    /// The entry name of a singleton port in `flusso.toml`, also its `<NAME>`
67    /// in override variables: `source` and `stream`. Sinks are named by the
68    /// user, so this is `None` for them.
69    pub fn singleton_entry(self) -> Option<&'static str> {
70        match self {
71            Port::Source => Some("source"),
72            Port::Stream => Some("stream"),
73            Port::Sink => None,
74        }
75    }
76}
77
78impl fmt::Display for Port {
79    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
80        f.write_str(match self {
81            Port::Source => "source",
82            Port::Stream => "stream",
83            Port::Sink => "sink",
84        })
85    }
86}
87
88/// The schema-property marker a [`Secret`](crate::Secret) plants so a
89/// description can find the fields that take an override variable.
90pub const SECRET_MARKER: &str = "x-flusso-secret";
91
92/// An adapter's configuration type: its port, its kind, and an example.
93///
94/// The provided [`description`](Self::description) and
95/// [`from_options`](Self::from_options) are what the composition root uses;
96/// adapters implement only the three required items, usually through
97/// `#[derive(AdapterConfig)]`.
98pub trait AdapterConfig: JsonSchema + Serialize + DeserializeOwned + Sized {
99    /// Which port this adapter implements.
100    const PORT: Port;
101
102    /// The `type = "…"` token that selects this adapter in `flusso.toml`.
103    const KIND: &'static str;
104
105    /// A realistic, complete example, rendered into the docs and the schema.
106    fn example() -> Self;
107
108    /// Turn a port entry's options into this config, strictly.
109    fn from_options(options: Options) -> Result<Self, OptionsError> {
110        options.deserialize_into()
111    }
112
113    /// Everything the composition root renders from this declaration.
114    fn description() -> AdapterDescription {
115        // Draft-07 (`definitions`, no `$defs`): what TOML/YAML editor tooling
116        // supports most widely, and the dialect the assembled config schema uses.
117        let schema = schemars::generate::SchemaSettings::draft07()
118            .into_generator()
119            .into_root_schema_for::<Self>();
120        let secrets = secret_paths(&schema);
121        AdapterDescription {
122            port: Self::PORT,
123            kind: Self::KIND.to_owned(),
124            schema,
125            example: Options::from_serialize(&Self::example()).unwrap_or_default(),
126            secrets,
127        }
128    }
129}
130
131/// The rendered facts about one adapter's options: the JSON schema (draft-07;
132/// doc comments as descriptions, serde defaults as defaults), an example, and
133/// the field paths that accept an override variable.
134#[derive(Debug, Clone, Serialize, Deserialize)]
135pub struct AdapterDescription {
136    /// Which port the adapter implements.
137    pub port: Port,
138    /// The adapter's `type` token.
139    pub kind: String,
140    /// The JSON schema of the options struct.
141    pub schema: Schema,
142    /// A complete example, as it would appear under the entry.
143    pub example: Options,
144    /// The `Secret` fields, as `.`-joined paths from the entry root.
145    pub secrets: Vec<String>,
146}
147
148impl AdapterDescription {
149    /// The override variable of every secret field for the entry named
150    /// `entry` (a sink's name, or `source`/`stream` for the singletons).
151    pub fn override_vars<'a>(&'a self, entry: &'a str) -> impl Iterator<Item = String> + 'a {
152        self.secrets
153            .iter()
154            .map(move |path| override_var(entry, &self.kind, path))
155    }
156}
157
158/// The environment variable that overrides one option of one port entry:
159/// `<ENTRY>_<KIND>_<FIELD>`, uppercased, nested path segments joined with `_`.
160///
161/// ```
162/// assert_eq!(kernel::override_var("primary", "opensearch", "url"), "PRIMARY_OPENSEARCH_URL");
163/// assert_eq!(
164///     kernel::override_var("source", "postgres", "connection_url.password"),
165///     "SOURCE_POSTGRES_CONNECTION_URL_PASSWORD",
166/// );
167/// ```
168pub fn override_var(entry: &str, kind: &str, field_path: &str) -> String {
169    format!("{entry}_{kind}_{}", field_path.replace('.', "_")).to_uppercase()
170}
171
172/// Walk a schema's `properties` (through `$ref`s into `$defs`, `anyOf`/`oneOf`
173/// alternatives, and `Option` wrappers) and collect the paths of every property
174/// carrying [`SECRET_MARKER`].
175fn secret_paths(schema: &Schema) -> Vec<String> {
176    let root = schema.as_value();
177    let defs = root
178        .get("$defs")
179        .or_else(|| root.get("definitions"))
180        .and_then(serde_json::Value::as_object);
181    let mut out = Vec::new();
182    walk(root, defs, &mut Vec::new(), &mut out, 0);
183    out.sort();
184    out.dedup();
185    out
186}
187
188fn walk(
189    node: &serde_json::Value,
190    defs: Option<&serde_json::Map<String, serde_json::Value>>,
191    path: &mut Vec<String>,
192    out: &mut Vec<String>,
193    depth: usize,
194) {
195    if depth > 16 {
196        return;
197    }
198    if node.get(SECRET_MARKER).and_then(serde_json::Value::as_bool) == Some(true)
199        && !path.is_empty()
200    {
201        out.push(path.join("."));
202        return;
203    }
204    if let Some(reference) = node.get("$ref").and_then(serde_json::Value::as_str) {
205        let name = reference.rsplit('/').next().unwrap_or(reference);
206        if let Some(target) = defs.and_then(|d| d.get(name)) {
207            walk(target, defs, path, out, depth + 1);
208        }
209        return;
210    }
211    for key in ["anyOf", "oneOf", "allOf"] {
212        if let Some(alternatives) = node.get(key).and_then(serde_json::Value::as_array) {
213            for alternative in alternatives {
214                walk(alternative, defs, path, out, depth + 1);
215            }
216        }
217    }
218    if let Some(properties) = node
219        .get("properties")
220        .and_then(serde_json::Value::as_object)
221    {
222        for (name, property) in properties {
223            path.push(name.clone());
224            walk(property, defs, path, out, depth + 1);
225            path.pop();
226        }
227    }
228}
229
230#[cfg(test)]
231#[allow(clippy::unwrap_used)]
232mod tests {
233    use super::*;
234    use crate::Secret;
235
236    #[derive(Debug, Serialize, Deserialize, JsonSchema)]
237    #[serde(deny_unknown_fields)]
238    struct Nested {
239        password: Option<Secret>,
240        host: String,
241    }
242
243    #[derive(Debug, Serialize, Deserialize, JsonSchema)]
244    #[serde(deny_unknown_fields)]
245    struct Demo {
246        url: Secret,
247        #[serde(default)]
248        token: Option<Secret>,
249        connection: Nested,
250        #[serde(default = "default_retries")]
251        retries: u32,
252    }
253
254    fn default_retries() -> u32 {
255        3
256    }
257
258    impl AdapterConfig for Demo {
259        const PORT: Port = Port::Sink;
260        const KIND: &'static str = "demo";
261
262        fn example() -> Self {
263            Self {
264                url: Secret::Value("https://x".into()),
265                token: None,
266                connection: Nested {
267                    password: Some(Secret::Env("PW".into())),
268                    host: "db".into(),
269                },
270                retries: 5,
271            }
272        }
273    }
274
275    #[test]
276    fn description_lists_secret_paths_including_nested_and_optional() {
277        let description = Demo::description();
278        assert_eq!(
279            description.secrets,
280            vec!["connection.password", "token", "url"]
281        );
282        let vars: Vec<String> = description.override_vars("primary").collect();
283        assert_eq!(
284            vars,
285            [
286                "PRIMARY_DEMO_CONNECTION_PASSWORD",
287                "PRIMARY_DEMO_TOKEN",
288                "PRIMARY_DEMO_URL"
289            ]
290        );
291    }
292
293    #[test]
294    fn description_carries_defaults_and_docs_from_the_struct() {
295        let description = Demo::description();
296        let retries = description.schema.pointer("/properties/retries").unwrap();
297        assert_eq!(retries.get("default"), Some(&serde_json::json!(3)));
298        assert_eq!(
299            description.example.get("retries").and_then(|v| v.as_i64()),
300            Some(5)
301        );
302        assert_eq!(description.port, Port::Sink);
303        assert_eq!(Port::Source.singleton_entry(), Some("source"));
304        assert_eq!(Port::Sink.singleton_entry(), None);
305        assert_eq!(Port::Stream.to_string(), "stream");
306    }
307
308    #[test]
309    fn from_options_is_strict() {
310        let mut options = Options::empty();
311        options.insert("url", "https://x");
312        options.insert("retriez", 1i64);
313        let error = Demo::from_options(options).unwrap_err().to_string();
314        assert!(error.contains("unknown field `retriez`"), "{error}");
315    }
316}
kernel/adapter.rs

kernel/
adapter.rs
