dre_core/
options.rs

1//! Plugin options: a format's `output:` keys and a destination entry's keys. Core owns only the
2//! keys every output shares (`format`, `destination`, `template`, `extension`, `profile`,
3//! `path`); everything else belongs to the plugin, which checks it (the protocol's `validate`).
4//! A new format or destination therefore needs no change to core.
5
6use std::collections::BTreeMap;
7use std::path::PathBuf;
8use std::sync::Arc;
9
10use dre_protocol::host::LogSink;
11use serde_json::{Map, Value};
12
13use crate::Diagnostics;
14use crate::codes::Code;
15use crate::profiles::{LOCAL_TYPE, Role};
16use crate::project::{PluginKind, Project};
17use crate::run::find_plugin;
18
19/// Parse `B12` into zero-based `(row, col)`.
20pub use dre_protocol::util::parse_cell;
21
22pub fn is_cell(s: &str) -> bool {
23    parse_cell(s).is_some()
24}
25
26/// Where a config block is used: the report (and Set), its file, and for a destination, the
27/// profile it names.
28struct Use {
29    ctx: String,
30    file: PathBuf,
31    profile: Option<String>,
32}
33
34/// One destination entry: its output's format, and whether it attaches other outputs.
35struct Route {
36    at: Use,
37    format: String,
38    attach: bool,
39}
40
41/// One plugin's distinct option blocks (keyed by their JSON and template), each with its
42/// template check (html and pdf outputs) and where it's used.
43type Blocks = BTreeMap<String, (Map<String, Value>, Option<Value>, Vec<Use>)>;
44
45/// Ask each format and destination plugin to check every config block the project gives it,
46/// found by kind and name alone: core knows nothing of any plugin's options. Each destination
47/// profile's output is its entry for the run. A declared plugin that can't be found is an error, or
48/// a warning with `offline` (`dre validate --no-auto-install`, which doesn't install plugins).
49pub fn check(project: &Project, offline: bool, diags: &mut Diagnostics) {
50    // (kind, plugin) -> distinct option blocks -> where each is used.
51    let mut blocks: BTreeMap<(PluginKind, String), Blocks> = BTreeMap::new();
52    // Destination type -> what each entry of it sends: checked against its capabilities.
53    let mut routes: BTreeMap<String, Vec<Route>> = BTreeMap::new();
54    let mut add = |kind, name: &str, options: &Map<String, Value>, template: Option<Value>, u: Use| {
55        let key = format!(
56            "{}{}",
57            serde_json::to_string(options).unwrap_or_default(),
58            template.as_ref().map(Value::to_string).unwrap_or_default()
59        );
60        blocks
61            .entry((kind, name.to_string()))
62            .or_default()
63            .entry(key)
64            .or_insert_with(|| (options.clone(), template, Vec::new()))
65            .2
66            .push(u);
67    };
68    for (format, options) in &project.format_options {
69        let u = Use {
70            ctx: format!("`format_options.{format}`"),
71            file: PathBuf::from(crate::project::PROJECT_FILE),
72            profile: None,
73        };
74        add(PluginKind::Format, format, options, None, u);
75    }
76    for r in &project.reports {
77        for b in &r.bindings {
78            let ctx = match &b.set {
79                Some(s) => format!("report `{}`, Set `{s}`", r.name),
80                None => format!("report `{}`", r.name),
81            };
82            let at = |profile: Option<&str>| Use {
83                ctx: ctx.clone(),
84                file: r.file.clone(),
85                profile: profile.map(str::to_string),
86            };
87            for o in b.outputs.iter().filter(|o| !o.is_message()) {
88                let template = crate::document::FORMATS.contains(&o.format.as_str()).then(|| {
89                    // A file that can't be resolved is reported by the template checks.
90                    let file = o
91                        .template
92                        .as_ref()
93                        .and_then(|t| crate::project::resolve_template(&project.root, &r.file, &t.file).ok());
94                    let queries: Vec<String> = b
95                        .queries
96                        .iter()
97                        .filter(|q| q.tab && o.feeds(&q.query))
98                        .map(|q| q.query.clone())
99                        .collect();
100                    let abs = |p: &std::path::Path| {
101                        std::path::absolute(p)
102                            .unwrap_or_else(|_| p.to_path_buf())
103                            .to_string_lossy()
104                            .into_owned()
105                    };
106                    serde_json::to_value(crate::document::Check {
107                        file: file.as_deref().map(abs),
108                        project_root: abs(&project.root),
109                        queries,
110                    })
111                    .unwrap_or_default()
112                });
113                if template.as_ref().is_some_and(|t| t["file"].is_null()) && o.template.is_some() {
114                    // Its file is missing: nothing to check.
115                    add(PluginKind::Format, &o.format, &o.options, None, at(None));
116                    continue;
117                }
118                add(PluginKind::Format, &o.format, &o.options, template, at(None));
119            }
120            let message_of: Vec<&crate::project::Output> = b
121                .outputs
122                .iter()
123                .flat_map(|o| o.destinations.iter().map(move |_| o))
124                .collect();
125            for (i, d) in b.destinations().enumerate() {
126                // The profile as the parse pass rendered it (a Jinja `profile:`).
127                let rendered = b
128                    .parsed
129                    .as_ref()
130                    .and_then(|p| p.destinations.get(i).cloned().flatten());
131                let profile = rendered.as_deref().unwrap_or(&d.profile);
132                let kind = if project.profiles.is_builtin_local(profile) {
133                    LOCAL_TYPE
134                } else {
135                    match project.profiles.target(Role::Destination, profile) {
136                        Some(t) => t.kind.as_str(),
137                        // A missing profile or entry is reported elsewhere; `deliver: false` takes no options.
138                        None => continue,
139                    }
140                };
141                add(PluginKind::Destination, kind, &d.options, None, at(Some(profile)));
142                routes.entry(kind.to_string()).or_default().push(Route {
143                    at: at(Some(profile)),
144                    format: message_of[i].format.clone(),
145                    // A zipping entry packs its attachments into the zip: any destination takes it.
146                    attach: !d.attach.is_empty() && d.zip.is_none(),
147                });
148            }
149        }
150    }
151
152    let log: LogSink = Arc::new(|_, _| {});
153    for ((kind, name), blocks) in blocks {
154        let report = |diags: &mut Diagnostics, u: &Use, e: &str| {
155            let (code, msg) = match &u.profile {
156                Some(p) => (
157                    Code::InvalidDestinationOption,
158                    format!("{}: destination `{p}`: {e}", u.ctx),
159                ),
160                None => (Code::InvalidOutputOption, format!("{}: {e}", u.ctx)),
161            };
162            diags.error(code, Some(u.file.clone()), None, msg);
163        };
164        // Core's own destination.
165        if kind == PluginKind::Destination && name == LOCAL_TYPE {
166            for r in routes.get(&name).into_iter().flatten().filter(|r| r.attach) {
167                report(
168                    diags,
169                    &r.at,
170                    "`attach:` needs a destination that takes messages and files, but `local` takes files only",
171                );
172            }
173            for (options, _, uses) in blocks.values() {
174                for e in local_option_errors(options) {
175                    for u in uses {
176                        report(diags, u, &e);
177                    }
178                }
179            }
180            continue;
181        }
182        let plugin = match find_plugin(project, kind, &name) {
183            Ok(p) => p,
184            // Already an error (`undeclared-plugin`).
185            Err(crate::plugins::LocateError::NotProvided(_)) => continue,
186            Err(e) => {
187                let first = blocks.values().flat_map(|(_, _, u)| u).next();
188                let file = first.map(|u| u.file.clone());
189                let at = first.map(|u| format!("{}: ", u.ctx)).unwrap_or_default();
190                if offline {
191                    diags.warning(
192                        Code::OptionsUnchecked,
193                        file,
194                        None,
195                        format!("{at}{kind} `{name}`'s options weren't checked: {e}"),
196                    );
197                } else {
198                    diags.error(
199                        Code::PluginNotFound,
200                        file,
201                        None,
202                        format!("{at}{kind} `{name}` has no plugin to check it against: {e}"),
203                    );
204                }
205                continue;
206            }
207        };
208        let mut p = match plugin.start(log.clone(), Some(&project.root)) {
209            Ok(p) => p,
210            Err(e) => {
211                diags.warning(
212                    Code::OptionsUnchecked,
213                    None,
214                    None,
215                    format!("can't check the options of {kind} `{name}`: {e}"),
216                );
217                continue;
218            }
219        };
220        if kind == PluginKind::Destination {
221            for r in routes.get(&name).into_iter().flatten() {
222                let message = r.format == crate::project::MESSAGE_FORMAT;
223                let problem = if p.has(dre_protocol::CAP_MESSAGE_ONLY) && !message {
224                    Some(format!(
225                        "`{name}` only takes messages, but this output is `{}`; deliver the file to a file destination (s3, sftp, a Databricks Volume...) and link it from a message output with `outputs.<name>.location`",
226                        r.format
227                    ))
228                } else if r.attach && p.has(dre_protocol::CAP_MESSAGE_ONLY) {
229                    Some(format!(
230                        "`attach:` needs a destination that takes files, but `{name}` only takes messages; deliver the files elsewhere and link them with `outputs.<name>.location`"
231                    ))
232                } else if r.attach && !p.has(dre_protocol::CAP_MESSAGE) {
233                    Some(format!(
234                        "`attach:` needs a destination that takes messages and files, but `{name}` takes files only"
235                    ))
236                } else {
237                    None
238                };
239                if let Some(e) = problem {
240                    report(diags, &r.at, &e);
241                }
242            }
243        }
244        if !p.has(dre_protocol::CAP_VALIDATE) {
245            if blocks.values().any(|(o, _, _)| !o.is_empty()) {
246                diags.warning(
247                    Code::OptionsUnchecked,
248                    None,
249                    None,
250                    format!(
251                        "{kind} `{name}` {} predates option checks, so its options weren't checked; run `dre plugin update {kind}/{name}`",
252                        p.info().version
253                    ),
254                );
255            }
256            let _ = p.close();
257            continue;
258        }
259        for (options, template, uses) in blocks.into_values() {
260            match p.validate_with(options, template) {
261                Ok((errors, warnings)) => {
262                    for u in &uses {
263                        for e in &errors {
264                            report(diags, u, e);
265                        }
266                        for w in &warnings {
267                            let at = format!("{}: {w}", u.ctx);
268                            diags.warning(Code::InvalidTemplate, Some(u.file.clone()), None, at);
269                        }
270                    }
271                }
272                Err(e) => {
273                    diags.warning(
274                        Code::OptionsUnchecked,
275                        None,
276                        None,
277                        format!("{kind} `{name}` failed to check its options: {e}"),
278                    );
279                    break;
280                }
281            }
282        }
283        let _ = p.close();
284    }
285}
286
287/// The options core's built-in `local` destination takes: the shared delivery rules'
288/// `if_exists`, `atomic` and `temp_dir`.
289pub fn local_options() -> Vec<dre_protocol::options::OptionField> {
290    dre_protocol::delivery::option_fields()
291}
292
293/// Every problem with a `local` destination entry's options.
294pub fn local_option_errors(options: &serde_json::Map<String, serde_json::Value>) -> Vec<String> {
295    dre_protocol::options::check(
296        dre_protocol::Kind::Destination,
297        LOCAL_TYPE,
298        &local_options(),
299        options,
300    )
301}
302
303/// One profile entry to check: its role, profile and entry (target) name.
304pub type EntryRef = (Role, String, String);
305
306/// Every entry of every profile, for `dre validate --all-targets`.
307pub fn all_entries(project: &Project) -> Vec<EntryRef> {
308    let mut out = Vec::new();
309    for (role, profiles) in [
310        (Role::Connection, &project.profiles.connections),
311        (Role::Destination, &project.profiles.destinations),
312    ] {
313        for (name, p) in profiles {
314            for target in p.targets.keys() {
315                out.push((role, name.clone(), target.clone()));
316            }
317        }
318    }
319    out
320}
321
322/// Ask each entry's plugin to check its connection settings without connecting (the protocol's
323/// `validate_connection`). A value from an unset `env_var()` is reported once as an
324/// `unset-env-var` warning (CI often validates without secrets) and sent unresolved, so it isn't
325/// reported again as missing.
326pub fn check_connections(project: &Project, entries: &[EntryRef], diags: &mut Diagnostics) {
327    let log: LogSink = Arc::new(|_, _| {});
328    let file = project.profiles.file.clone();
329    let mut unset_reported = std::collections::BTreeSet::new();
330    // Plugin -> the entries it checks, so each plugin starts once.
331    let mut by_plugin: BTreeMap<(PluginKind, String), Vec<&EntryRef>> = BTreeMap::new();
332    for e in entries {
333        let (role, profile, target) = e;
334        let Some(t) = project
335            .profiles
336            .get(*role, profile)
337            .and_then(|p| p.targets.get(target))
338        else {
339            continue;
340        };
341        if t.kind == LOCAL_TYPE {
342            continue;
343        }
344        let kind = match role {
345            Role::Connection => PluginKind::Source,
346            Role::Destination => PluginKind::Destination,
347        };
348        by_plugin.entry((kind, t.kind.clone())).or_default().push(e);
349    }
350    for ((kind, name), entries) in by_plugin {
351        // A missing plugin is reported by the plugin checks.
352        let Ok(plugin) = find_plugin(project, kind, &name) else {
353            continue;
354        };
355        let mut p = match plugin.start(log.clone(), Some(&project.root)) {
356            Ok(p) => p,
357            Err(e) => {
358                diags.warning(
359                    Code::OptionsUnchecked,
360                    file.clone(),
361                    None,
362                    format!("can't check the settings of {kind} `{name}`: {e}"),
363                );
364                continue;
365            }
366        };
367        for (role, profile, target) in entries {
368            let t = &project.profiles.get(*role, profile).unwrap().targets[target];
369            let line = project.profiles.line_of(*role, profile);
370            let at = format!("{} `{profile}` (entry `{target}`)", role.as_str());
371            // Values with an unset `env_var()` go unresolved; the rest are rendered as for a run.
372            let mut known = t.clone();
373            let mut unresolved = Vec::new();
374            for (k, v) in &t.fields {
375                let unset: Vec<String> = unset_env_vars(v);
376                if unset.is_empty() {
377                    continue;
378                }
379                known.fields.insert(k.clone(), Value::Null);
380                unresolved.push(k.clone());
381                for var in unset {
382                    // A warning: CI often validates without the secrets a run needs.
383                    if unset_reported.insert(var.clone()) {
384                        diags.warning(
385                            Code::UnsetEnvVar,
386                            file.clone(),
387                            line,
388                            format!(
389                                "{at}: `env_var('{var}')`: environment variable `{var}` is not set and no default is given; `dre run` will need it"
390                            ),
391                        );
392                    }
393                }
394            }
395            let rendered = match crate::run::render_connection(&known) {
396                Ok(r) => r,
397                Err(e) => {
398                    diags.error(
399                        Code::InvalidConnectionSetting,
400                        file.clone(),
401                        line,
402                        format!("{at}: {e}"),
403                    );
404                    continue;
405                }
406            };
407            match p.validate_connection(rendered, unresolved) {
408                Ok((errors, warnings)) => {
409                    for e in errors {
410                        diags.error(
411                            Code::InvalidConnectionSetting,
412                            file.clone(),
413                            line,
414                            format!("{at}: {e}"),
415                        );
416                    }
417                    for w in warnings {
418                        diags.warning(
419                            Code::UnknownConnectionKey,
420                            file.clone(),
421                            line,
422                            format!("{at}: {w}"),
423                        );
424                    }
425                }
426                Err(e) => {
427                    diags.warning(
428                        Code::OptionsUnchecked,
429                        file.clone(),
430                        line,
431                        format!(
432                            "{at}: {kind} `{name}` {} couldn't check its settings ({e}); run `dre plugin update {kind}/{name}`",
433                            p.info().version
434                        ),
435                    );
436                    break;
437                }
438            }
439        }
440        let _ = p.close();
441    }
442}
443
444/// The variables of `env_var()` calls without a default that aren't set, anywhere in `v`.
445fn unset_env_vars(v: &Value) -> Vec<String> {
446    match v {
447        Value::String(s) if crate::preflight::is_templated(s) => crate::preflight::calls(s)
448            .into_iter()
449            .filter(|c| c.func == "env_var" && !c.has_default && std::env::var_os(&c.name).is_none())
450            .map(|c| c.name)
451            .collect(),
452        Value::Array(a) => a.iter().flat_map(unset_env_vars).collect(),
453        Value::Object(o) => o.values().flat_map(unset_env_vars).collect(),
454        _ => Vec::new(),
455    }
456}
dre_core/options.rs

dre_core/
options.rs
