Skip to main content

dre_core/
run.rs

1//! The run path: turn a resolved project into delivered files.
2//!
3//! Per Binding: parse pass → render → split → (unmanaged check) → execute, each query on its
4//! connection's session (one per connection, opened on first use, in strict YAML order) →
5//! format into `target/run/` → schema-drift check → deliver → snapshot → `run_results.json`.
6
7use std::collections::BTreeMap;
8use std::fs::File;
9use std::path::{Path, PathBuf};
10use std::sync::atomic::{AtomicBool, Ordering};
11use std::sync::{Arc, Mutex};
12use std::time::{Duration, Instant};
13
14use arrow::array::RecordBatch;
15use arrow::datatypes::SchemaRef;
16use arrow::ipc::reader::FileReader;
17use arrow::ipc::writer::FileWriter;
18use chrono::NaiveDate;
19use dre_protocol::host::{Execution, HostError, LogSink, PluginProcess};
20use dre_protocol::msg::{ColumnOptions, DeliveryFile, ResultSetMeta};
21use dre_protocol::{CAP_LOAD, CAP_MESSAGE, CAP_MESSAGE_ONLY, CAP_MULTI_FILE, CAP_READ_ONLY, CAP_SESSIONS};
22use serde::Serialize;
23use serde_json::{Map as JsonMap, Value as Json, json};
24
25use crate::codes::{Code, ErrorCode};
26use crate::dates::Calendar;
27use crate::engine::{CancelToken, Events, RunEvent, RunStore};
28use crate::lookups::Table;
29use crate::message::QueryResult;
30use crate::parse::ParsedBinding;
31use crate::profiles::{Entry, LOCAL_TYPE, ProfileTarget, Profiles, Role};
32use crate::project::{Binding, PluginKind, Project, QueryEntry, Report};
33use crate::render::{
34    Column, Connection, Connections, Mode, QueryRows, QueryRunner, RenderError, Renderer, RendererConfig,
35};
36use crate::run_results::{self, Delivery, RunResults};
37use crate::selector;
38use crate::sqlsplit::{self, StatementKind};
39
40/// What the caller asked for.
41#[derive(Debug, Clone, Default)]
42pub struct RunOptions {
43    pub selector: Option<String>,
44    /// `--set <name>` or `--set all`.
45    pub set: Option<String>,
46    /// `--target` as given; the run's target is [`Project::target_name`].
47    pub target: Option<String>,
48    /// `--profile`: the inherited connection, for every Binding.
49    pub profile: Option<String>,
50    pub vars: BTreeMap<String, String>,
51    pub output_name: Option<String>,
52    pub output_path: Option<String>,
53    pub dry_run: bool,
54    pub no_deliver: bool,
55    /// Optional isolated run store, plus typed rows under report/binding/query.json.
56    pub output_json: Option<PathBuf>,
57    /// `--preview [N]`: row limit per query.
58    pub preview: Option<u64>,
59    pub accept_schema_change: bool,
60    /// Whether a person is at a terminal to answer prompts.
61    pub interactive: bool,
62    /// `run.date` (`DRE_RUN_DATE`).
63    pub date: Option<NaiveDate>,
64    /// The instant this run was scheduled for (`DRE_RUN_AT`): `run.now` and `run.scheduled_at`,
65    /// and `run.date` in the run's timezone unless `date` is set.
66    pub scheduled_at: Option<chrono::DateTime<chrono::Utc>>,
67    /// `validate --live`: check statements instead of executing them.
68    pub live_check: bool,
69    /// `--schedule <name>`: run exactly the Bindings that schedule targets, with its vars.
70    pub schedule: Option<String>,
71    /// `--timezone` or `DRE_TIMEZONE`: above every configured `timezone:`.
72    pub timezone: Option<String>,
73    /// SHA-256 of the manifest this command wrote, for `run_results.json`.
74    pub manifest_checksum: Option<String>,
75    /// Stops the run: no further Binding, statement or delivery starts once it's cancelled.
76    pub cancel: CancelToken,
77    /// `--keep-runs` or `DRE_KEEP_RUNS`: above `flags: keep_runs`.
78    pub keep_runs: Option<usize>,
79    /// `--threads` or `DRE_THREADS`: how many Bindings may run at once in the whole run, over
80    /// each connection's `threads:`.
81    pub threads: Option<usize>,
82}
83
84impl RunOptions {
85    /// What the run was asked to do, for `run_results.json` and the log.
86    pub fn params(&self, date: NaiveDate) -> run_results::Params {
87        run_results::Params {
88            selector: self.selector.clone(),
89            set: self.set.clone(),
90            schedule: self.schedule.clone(),
91            target: self.target.clone(),
92            profile: self.profile.clone(),
93            vars: self.vars.clone(),
94            run_date: date.to_string(),
95            scheduled_at: self.scheduled_at.map(rfc3339),
96            timezone: self.timezone.clone(),
97            output_name: self.output_name.clone(),
98            output_path: self.output_path.clone(),
99            dry_run: self.dry_run,
100            preview: self.preview,
101            accept_schema_change: self.accept_schema_change,
102        }
103    }
104}
105
106/// How much a message matters: `Info` is shown by default, `Debug` with `-v`.
107#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
108pub enum Level {
109    Debug,
110    Info,
111}
112
113/// How the run reports progress. The engine emits structured events; the CLI decides how
114/// they look (colour, progress bar, JSON, log file).
115pub trait Ui {
116    /// The run's target and each used profile's entry, once they're all known to exist.
117    fn targets(&mut self, _targets: &RunTargets) {}
118    /// The number of Bindings about to run, once Sets are resolved.
119    fn plan(&mut self, _bindings: usize) {}
120    fn binding_start(&mut self, _report: &str, _set: Option<&str>) {}
121    /// The running Binding's log file, in its run folder (a real run only).
122    fn binding_log(&mut self, _path: &Path) {}
123    /// Right after `binding_start`: the schedule it runs under (if any) and every var it uses.
124    fn binding_vars(
125        &mut self,
126        _schedule: Option<&str>,
127        _schedule_vars: Option<&JsonMap<String, Json>>,
128        _vars: &JsonMap<String, Json>,
129    ) {
130    }
131    /// One step inside the current Binding: a short verb, a detail, and how long it took.
132    fn step(&mut self, level: Level, verb: &str, detail: &str, elapsed: Option<Duration>);
133    fn warn(&mut self, msg: &str);
134    fn binding_end(&mut self, _outcome: &BindingOutcome) {}
135    /// A Binding was compiled (`--dry-run`, `dre compile`, `dre validate`): what it would do.
136    fn compiled(&mut self, _plan: &BindingPlan) {}
137    /// `--preview`: a rendered message, shown instead of sent.
138    fn message(&mut self, _message: &ShownMessage) {}
139    /// Ask which Set to run; `None` means "all".
140    fn choose_set(&mut self, report: &str, sets: &[String]) -> Result<Option<String>, String>;
141    /// Bindings running at once: `label` started (`running`) or finished. Its own events come
142    /// later, together.
143    fn running(&mut self, _label: &str, _running: bool) {}
144    /// Where plugin stderr goes.
145    fn plugin_log(&self) -> LogSink;
146    /// Where the full text of every statement sent to a source goes: `(label, sql)`.
147    fn sql_log(&self) -> LogSink {
148        Arc::new(|_, _| {})
149    }
150}
151
152/// A message `--preview` shows.
153#[derive(Debug, Clone, Serialize)]
154pub struct ShownMessage {
155    /// The output's `name:`.
156    pub output: Option<String>,
157    pub title: String,
158    /// Portable Markdown.
159    pub text: String,
160    /// Length, the row sample, `when:` and each destination's limit.
161    pub notes: Vec<String>,
162}
163
164pub use crate::run_results::Status;
165
166#[derive(Debug, Clone, Serialize)]
167pub struct BindingOutcome {
168    pub report: String,
169    #[serde(skip_serializing_if = "Option::is_none")]
170    pub set: Option<String>,
171    pub binding: String,
172    pub status: Status,
173    #[serde(skip_serializing_if = "Option::is_none")]
174    pub error: Option<String>,
175    /// `error`'s code.
176    #[serde(skip_serializing_if = "Option::is_none")]
177    pub error_code: Option<ErrorCode>,
178    pub files: Vec<PathBuf>,
179    /// One line describing what the Binding produced (result sets, rows, outputs, delivery).
180    pub summary: String,
181    /// The schedule this ran under (`--schedule`), and the vars it layered in.
182    pub schedule: Option<String>,
183    pub schedule_vars: Option<JsonMap<String, Json>>,
184    /// Every var the Binding rendered with: its own, the schedule's, then `--var`.
185    pub vars: JsonMap<String, Json>,
186    /// The run's timezone (IANA name); empty when the Binding never started.
187    #[serde(skip_serializing_if = "String::is_empty")]
188    pub timezone: String,
189    #[serde(skip)]
190    pub elapsed: Duration,
191}
192
193/// What a compiled Binding would use and produce if it ran.
194#[derive(Debug, Clone, Serialize)]
195pub struct BindingPlan {
196    pub report: String,
197    pub set: Option<String>,
198    /// Rendered SQL files, relative to the project.
199    pub compiled: Vec<PathBuf>,
200    /// The inherited connection, if any.
201    pub profile: Option<String>,
202    /// Each query's connection, its type and the sources it reads.
203    pub queries: Vec<PlannedQuery>,
204    pub target: String,
205    pub format: String,
206    /// The file written under `target/run/` (a single-table format writes one per result set
207    /// when there are several).
208    pub output: PathBuf,
209    pub destinations: Vec<PlannedDestination>,
210    /// Every output (the fields above repeat the first one).
211    pub outputs: Vec<PlannedOutput>,
212    pub schedules: Vec<String>,
213}
214
215#[derive(Debug, Clone, Serialize)]
216pub struct PlannedOutput {
217    pub name: Option<String>,
218    pub format: String,
219    /// The file written under `target/run/`.
220    pub output: PathBuf,
221    pub destinations: Vec<PlannedDestination>,
222}
223
224#[derive(Debug, Clone, Serialize)]
225pub struct PlannedQuery {
226    pub query: String,
227    pub connection: String,
228    #[serde(rename = "type")]
229    pub kind: String,
230    /// The connection's entry: the run's target, else its own `target:`, else `dev`.
231    pub target: String,
232    #[serde(skip_serializing_if = "Vec::is_empty")]
233    pub sources: Vec<String>,
234}
235
236#[derive(Debug, Clone, Serialize)]
237pub struct PlannedDestination {
238    pub profile: String,
239    /// The destination type of its entry for this run; `None` for `deliver: false`.
240    pub kind: Option<String>,
241    /// Its entry: the run's target, else the profile's own `target:`, else `dev`.
242    pub target: Option<String>,
243    pub path: Option<String>,
244    /// False for a `deliver: false` entry: the output stays in the target path.
245    pub delivers: bool,
246}
247
248#[derive(Debug, Default)]
249pub struct RunSummary {
250    pub outcomes: Vec<BindingOutcome>,
251    /// Selection failed, or a profile has no entry for the run: nothing ran.
252    pub error: Option<String>,
253    /// `error` is about a profile's missing entry.
254    pub missing_entry: bool,
255    /// Why the run was cancelled, if it was.
256    pub cancelled: Option<crate::engine::CancelReason>,
257    /// Bindings that never started because the run was cancelled.
258    pub not_run: usize,
259}
260
261impl RunSummary {
262    pub fn failed(&self) -> bool {
263        self.error.is_some()
264            || self.cancelled.is_some()
265            || self
266                .outcomes
267                .iter()
268                .any(|o| matches!(o.status, Status::Error | Status::Cancelled | Status::TimedOut))
269    }
270
271    /// The code of what failed first: why nothing ran, else the first failed Binding's.
272    pub fn error_code(&self) -> Option<ErrorCode> {
273        if self.error.is_some() {
274            return Some(ErrorCode::Core(if self.missing_entry {
275                Code::MissingTargetEntry
276            } else {
277                Code::InvalidSelector
278            }));
279        }
280        self.outcomes
281            .iter()
282            .find(|o| matches!(o.status, Status::Error | Status::Cancelled | Status::TimedOut))
283            .map(|o| o.error_code.clone().unwrap_or(ErrorCode::Core(Code::RunFailed)))
284            .or_else(|| {
285                self.cancelled.map(|r| {
286                    ErrorCode::Core(match r {
287                        crate::engine::CancelReason::Timeout => Code::RunTimedOut,
288                        _ => Code::RunCancelled,
289                    })
290                })
291            })
292    }
293
294    /// The exit code (see docs/exit-codes.md): 130, 143 or 124 for a cancelled or timed-out run;
295    /// 2 when nothing ran because the selection or a profile is wrong; 1 when a Binding failed;
296    /// else 0.
297    pub fn exit_code(&self) -> u8 {
298        if let Some(r) = self.cancelled {
299            return r.exit_code();
300        }
301        if self.error.is_some() {
302            // Nothing ran: the selection or a profile is wrong.
303            return self.error_code().map_or(2, |c| c.kind().exit_code());
304        }
305        let failed: Vec<&BindingOutcome> = self
306            .outcomes
307            .iter()
308            .filter(|o| matches!(o.status, Status::Error | Status::Cancelled | Status::TimedOut))
309            .collect();
310        // Every Binding that failed was refused before it started (already running): nothing ran.
311        if !failed.is_empty()
312            && failed.iter().all(|o| {
313                o.error_code
314                    .as_ref()
315                    .is_some_and(|c| c.kind() == crate::codes::Kind::Refused)
316            })
317            && self.outcomes.len() == failed.len()
318        {
319            return 2;
320        }
321        // The run happened; a failed Binding exits 1 whatever its kind.
322        if failed.is_empty() { 0 } else { 1 }
323    }
324}
325
326/// What a run will do: the Bindings to run, and the run's target with every profile's entry.
327pub struct Plan<'a> {
328    pub bindings: Vec<(&'a Report, Binding)>,
329    pub targets: RunTargets,
330}
331
332/// Run what `opts` selects: [`plan`] it, then run the plan on a worker thread while this thread
333/// shows its events through `ui`.
334pub fn run(project: &Project, opts: &RunOptions, ui: &mut dyn Ui) -> RunSummary {
335    let mut summary = RunSummary::default();
336    let Some(plan) = plan(project, opts, ui, &mut summary) else {
337        return summary;
338    };
339    ui.targets(&plan.targets);
340    ui.plan(plan.bindings.len());
341    let (plugin_log, sql_log) = (ui.plugin_log(), ui.sql_log());
342    let (tx, rx) = std::sync::mpsc::channel();
343    let cancel = opts.cancel.clone();
344    let outcomes = std::thread::scope(|s| {
345        let worker = s.spawn(|| execute(project, &plan, opts, &cancel, Events::new(tx)));
346        for e in rx {
347            crate::engine::dispatch(ui, e, &plugin_log, &sql_log);
348        }
349        worker.join().expect("the run's worker doesn't panic")
350    });
351    summary.not_run = plan.bindings.len().saturating_sub(outcomes.len());
352    summary.cancelled = cancel.reason();
353    summary.outcomes.extend(outcomes);
354    summary
355}
356
357/// Which Bindings `opts` selects, and every profile's entry for the run. Asking which Set to run
358/// happens here, through `ui`. A selection that fails for one report is recorded as that
359/// report's failed outcome in `summary`; a failure that stops the whole run is `None`, with
360/// `summary.error` set.
361pub fn plan<'a>(
362    project: &'a Project,
363    opts: &RunOptions,
364    ui: &mut dyn Ui,
365    summary: &mut RunSummary,
366) -> Option<Plan<'a>> {
367    if let Err(e) = crate::target::ensure(&project.target_dir) {
368        summary.error = Some(format!(
369            "can't write to the target path {} (from {}): {e}",
370            project.target_dir.display(),
371            project.target_source
372        ));
373        return None;
374    }
375    let planned: Vec<(&Report, Binding)> = match &opts.schedule {
376        Some(name) => {
377            if !project.schedules.iter().any(|e| &e.name == name) {
378                summary.error = Some(unknown_schedule(project, name));
379                return None;
380            }
381            match schedule_bindings(project, name, opts) {
382                Ok(p) => p,
383                Err(e) => {
384                    summary.error = Some(e);
385                    return None;
386                }
387            }
388        }
389        None => {
390            let reports: Vec<&Report> = match &opts.selector {
391                None => project.reports.iter().collect(),
392                Some(s) => match selector::resolve(project, s) {
393                    Ok(r) if r.is_empty() => {
394                        summary.error = Some(format!("selector `{s}` matches no report"));
395                        return None;
396                    }
397                    Ok(r) => r,
398                    Err(e) => {
399                        summary.error = Some(e.to_string());
400                        return None;
401                    }
402                },
403            };
404            // Resolve every report's Bindings first (prompts happen here), so progress has a
405            // total.
406            let mut planned = Vec::new();
407            for report in reports {
408                match choose_bindings(project, report, opts, ui) {
409                    Ok(bs) => planned.extend(bs.into_iter().map(|b| (report, b))),
410                    Err(e) => {
411                        let outcome = BindingOutcome {
412                            report: report.name.clone(),
413                            set: None,
414                            binding: "-".into(),
415                            status: Status::Error,
416                            error: Some(e),
417                            error_code: Some(Code::InvalidSelector.into()),
418                            files: Vec::new(),
419                            summary: String::new(),
420                            schedule: opts.schedule.clone(),
421                            schedule_vars: None,
422                            vars: JsonMap::new(),
423                            timezone: String::new(),
424                            elapsed: Duration::ZERO,
425                        };
426                        ui.binding_end(&outcome);
427                        summary.outcomes.push(outcome);
428                    }
429                }
430            }
431            planned
432        }
433    };
434    // Every profile the run uses needs an entry for it, before anything runs.
435    match run_targets(project, &planned, opts) {
436        Ok(targets) => Some(Plan {
437            bindings: planned,
438            targets,
439        }),
440        Err(e) => {
441            summary.error = Some(e);
442            summary.missing_entry = true;
443            None
444        }
445    }
446}
447
448/// Run a plan's Bindings, reporting through `events`; a cancelled run starts no further
449/// Binding. One at a time, in order, unless a connection's entry sets `threads:` above 1 or the
450/// run sets `--threads` (see [`Limits`]).
451pub fn execute(
452    project: &Project,
453    plan: &Plan<'_>,
454    opts: &RunOptions,
455    cancel: &CancelToken,
456    events: Events,
457) -> Vec<BindingOutcome> {
458    let limits = Limits::new(project, plan, opts);
459    let target = opts
460        .output_json
461        .as_ref()
462        .map(|p| project.root.join(p))
463        .unwrap_or_else(|| project.target_dir.clone());
464    if limits.sequential() {
465        let store = RunStore::new(&target);
466        let mut outcomes = Vec::new();
467        for (report, b) in &plan.bindings {
468            if cancel.is_cancelled() {
469                break;
470            }
471            events.emit(RunEvent::BindingStart {
472                report: report.name.clone(),
473                set: b.set.clone(),
474            });
475            let mut r = BindingRun::new(project, report, b, opts, events.clone(), &store);
476            let outcome = r.run();
477            events.emit(RunEvent::BindingEnd(outcome.clone()));
478            outcomes.push(outcome);
479        }
480        return outcomes;
481    }
482    let store = RunStore::concurrent(&target);
483    let n = plan.bindings.len();
484    let state = std::sync::Mutex::new(Running::new(n));
485    let wake = std::sync::Condvar::new();
486    let outcomes: std::sync::Mutex<Vec<Option<BindingOutcome>>> = std::sync::Mutex::new(vec![None; n]);
487    std::thread::scope(|s| {
488        for _ in 0..limits.cap.min(n) {
489            s.spawn(|| {
490                loop {
491                    let i = {
492                        let mut st = state.lock().unwrap();
493                        loop {
494                            if cancel.is_cancelled() || st.started.iter().all(|x| *x) {
495                                break None;
496                            }
497                            match limits.next(&st) {
498                                Some(i) => {
499                                    st.start(i, &limits.uses[i]);
500                                    break Some(i);
501                                }
502                                None => st = wake.wait(st).unwrap(),
503                            }
504                        }
505                    };
506                    let Some(i) = i else {
507                        wake.notify_all();
508                        return;
509                    };
510                    let (report, b) = &plan.bindings[i];
511                    let label = match &b.set {
512                        Some(set) => format!("{} [{set}]", report.name),
513                        None => report.name.clone(),
514                    };
515                    let held = events.held();
516                    held.emit_now(RunEvent::Running {
517                        label: label.clone(),
518                        running: true,
519                    });
520                    held.emit(RunEvent::BindingStart {
521                        report: report.name.clone(),
522                        set: b.set.clone(),
523                    });
524                    let mut r = BindingRun::new(project, report, b, opts, held.clone(), &store);
525                    let outcome = r.run();
526                    held.emit(RunEvent::BindingEnd(outcome.clone()));
527                    held.emit_now(RunEvent::Running {
528                        label,
529                        running: false,
530                    });
531                    held.flush();
532                    outcomes.lock().unwrap()[i] = Some(outcome);
533                    state.lock().unwrap().finish(&limits.uses[i]);
534                    wake.notify_all();
535                }
536            });
537        }
538    });
539    // Declaration order, whatever order they finished in.
540    outcomes.into_inner().unwrap().into_iter().flatten().collect()
541}
542
543/// How many Bindings may run at once: each connection entry's `threads:` (default 1; a DuckDB
544/// file is always 1, as only one process can write it), and `--threads` (`DRE_THREADS`) over
545/// the whole run.
546struct Limits {
547    /// The connections each Binding uses.
548    uses: Vec<Vec<String>>,
549    per_connection: BTreeMap<String, usize>,
550    cap: usize,
551}
552
553/// Which Bindings have started, and what's running on each connection.
554struct Running {
555    started: Vec<bool>,
556    total: usize,
557    on: BTreeMap<String, usize>,
558    /// Connections a Binding has used to the end: until then, a connection runs one Binding,
559    /// so its sign-in (a browser, a token refresh) happens once.
560    warm: std::collections::BTreeSet<String>,
561}
562
563impl Running {
564    fn new(n: usize) -> Running {
565        Running {
566            started: vec![false; n],
567            total: 0,
568            on: BTreeMap::new(),
569            warm: Default::default(),
570        }
571    }
572
573    fn start(&mut self, i: usize, uses: &[String]) {
574        self.started[i] = true;
575        self.total += 1;
576        for c in uses {
577            *self.on.entry(c.clone()).or_default() += 1;
578        }
579    }
580
581    fn finish(&mut self, uses: &[String]) {
582        self.total -= 1;
583        for c in uses {
584            *self.on.entry(c.clone()).or_default() -= 1;
585            self.warm.insert(c.clone());
586        }
587    }
588}
589
590impl Limits {
591    fn new(project: &Project, plan: &Plan<'_>, opts: &RunOptions) -> Limits {
592        let schedule_vars = opts
593            .schedule
594            .as_ref()
595            .and_then(|n| project.schedules.iter().find(|e| &e.name == n))
596            .map(|e| e.vars.clone());
597        let inputs = crate::parse::Inputs {
598            target: project.target_name.clone(),
599            cli_vars: opts.vars.clone(),
600            date: opts.date,
601            scheduled_at: opts.scheduled_at,
602            timezone: opts.timezone.clone(),
603            schedule: opts.schedule.clone(),
604            started_at: None,
605        };
606        let uses: Vec<Vec<String>> = plan
607            .bindings
608            .iter()
609            .map(|(report, b)| {
610                let mut vars = b.vars.clone();
611                vars.extend(schedule_vars.clone().unwrap_or_default());
612                crate::parse::binding(project, report, b, &vars, &inputs)
613                    .connections()
614                    .into_iter()
615                    .map(str::to_string)
616                    .collect()
617            })
618            .collect();
619        let mut per_connection = BTreeMap::new();
620        for c in uses.iter().flatten() {
621            per_connection
622                .entry(c.clone())
623                .or_insert_with(|| connection_threads(project, c));
624        }
625        // Only a real `dre run` (`--preview` included) runs Bindings at once.
626        let cap = if opts.dry_run || opts.live_check {
627            1
628        } else {
629            opts.threads.unwrap_or(usize::MAX).max(1)
630        };
631        Limits {
632            uses,
633            per_connection,
634            cap,
635        }
636    }
637
638    /// One at a time: `--threads 1`, or no `--threads` and no connection above 1.
639    fn sequential(&self) -> bool {
640        self.cap == 1 || (self.cap == usize::MAX && self.per_connection.values().all(|n| *n <= 1))
641    }
642
643    /// The first Binding, in declaration order, that may start now.
644    fn next(&self, st: &Running) -> Option<usize> {
645        if st.total >= self.cap {
646            return None;
647        }
648        (0..st.started.len()).find(|&i| {
649            !st.started[i]
650                && self.uses[i].iter().all(|c| {
651                    let on = st.on.get(c).copied().unwrap_or(0);
652                    let limit = if st.warm.contains(c) {
653                        self.per_connection[c]
654                    } else {
655                        1
656                    };
657                    on < limit
658                })
659        })
660    }
661}
662
663/// A connection entry's `threads:` (default 1); a DuckDB file is 1 whatever it says.
664fn connection_threads(project: &Project, connection: &str) -> usize {
665    let Some(t) = project.profiles.target(Role::Connection, connection) else {
666        return 1;
667    };
668    let n = match t.fields.get("threads") {
669        Some(Json::Number(n)) => n.as_u64().unwrap_or(1),
670        Some(Json::String(s)) => s.trim().parse().unwrap_or(1),
671        _ => 1,
672    } as usize;
673    let duckdb_file = t.kind == "duckdb"
674        && t.fields
675            .get("path")
676            .and_then(Json::as_str)
677            .is_some_and(|p| p != ":memory:");
678    if duckdb_file { 1 } else { n.max(1) }
679}
680
681/// A profile a run uses, and its entry.
682#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
683pub struct UsedProfile {
684    /// `connection` or `destination`.
685    pub role: &'static str,
686    pub profile: String,
687    /// Its entry: the run's target, else the profile's own `target:`, else `dev`.
688    pub target: String,
689    /// False for a destination entry written `deliver: false`.
690    pub deliver: bool,
691}
692
693/// The run's target and the entry each profile it uses picked.
694#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
695pub struct RunTargets {
696    /// `target.name`.
697    pub name: String,
698    /// `--target`, `DRE_TARGET` or `default`.
699    pub from: String,
700    pub profiles: Vec<UsedProfile>,
701}
702
703impl RunTargets {
704    /// `dev (default)`, then each profile whose entry differs: `; connection `warehouse`: prod`.
705    pub fn line(&self) -> String {
706        let mut out = format!("{} ({})", self.name, self.from);
707        for p in self.profiles.iter().filter(|p| p.target != self.name) {
708            out.push_str(&format!("; {} `{}`: {}", p.role, p.profile, p.target));
709        }
710        out
711    }
712
713    /// When every profile the run uses is on one target other than the run's, `target.name`
714    /// probably isn't what the templates expect.
715    pub fn mismatch(&self) -> Option<String> {
716        let first = &self.profiles.first()?.target;
717        if first == &self.name || self.profiles.iter().any(|p| &p.target != first) {
718            return None;
719        }
720        Some(format!(
721            "every profile is on `{first}` but the run's target is `{}`: pass `--target {first}` or set `{}`",
722            self.name,
723            crate::profiles::TARGET_ENV
724        ))
725    }
726}
727
728/// The profiles the planned Bindings use (by the parse pass, with this run's inputs) and each
729/// one's entry. A profile without an entry for the run is an error naming them all; profiles
730/// nothing planned uses aren't checked.
731pub fn run_targets(
732    project: &Project,
733    planned: &[(&Report, Binding)],
734    opts: &RunOptions,
735) -> Result<RunTargets, String> {
736    let profiles = &project.profiles;
737    let schedule_vars = opts
738        .schedule
739        .as_ref()
740        .and_then(|n| project.schedules.iter().find(|e| &e.name == n))
741        .map(|e| e.vars.clone());
742    let inputs = crate::parse::Inputs {
743        target: project.target_name.clone(),
744        cli_vars: opts.vars.clone(),
745        date: opts.date,
746        scheduled_at: opts.scheduled_at,
747        timezone: opts.timezone.clone(),
748        schedule: opts.schedule.clone(),
749        started_at: None,
750    };
751    let mut used: Vec<(Role, String)> = Vec::new();
752    for (report, b) in planned {
753        let mut vars = b.vars.clone();
754        vars.extend(schedule_vars.clone().unwrap_or_default());
755        // A Binding whose parse fails reports that when it runs.
756        let parsed = crate::parse::binding(project, report, b, &vars, &inputs);
757        let names = parsed
758            .connections()
759            .into_iter()
760            .map(|c| (Role::Connection, c.to_string()))
761            .chain(
762                parsed
763                    .destinations
764                    .iter()
765                    .flatten()
766                    .filter(|_| !opts.no_deliver)
767                    .filter(|d| !profiles.is_builtin_local(d))
768                    .map(|d| (Role::Destination, d.clone())),
769            );
770        for u in names {
771            if !used.contains(&u) {
772                used.push(u);
773            }
774        }
775    }
776    let mut missing = Vec::new();
777    let mut out = Vec::new();
778    for (role, name) in &used {
779        let deliver = match profiles.entry(*role, name) {
780            Entry::Use(_) => true,
781            Entry::Nowhere => false,
782            Entry::Missing => {
783                missing.push(profiles.missing_entry(*role, name));
784                continue;
785            }
786            // Reported by the load.
787            Entry::Unknown => continue,
788        };
789        out.push(UsedProfile {
790            role: role.as_str(),
791            profile: name.clone(),
792            target: profiles.target_of(*role, name),
793            deliver,
794        });
795    }
796    match missing.len() {
797        0 => Ok(RunTargets {
798            name: project.target_name.clone(),
799            from: project.target_from.to_string(),
800            profiles: out,
801        }),
802        1 => Err(format!("{}; nothing was run", missing[0])),
803        n => Err(format!(
804            "{n} profiles this run uses have no entry for it; nothing was run:\n    {}",
805            missing.join("\n    ")
806        )),
807    }
808}
809
810/// The Bindings `--schedule <name>` runs: all of them, or those a selector and/or `--set` pick.
811/// Picking something the schedule doesn't run is an error naming what it does run.
812fn schedule_bindings<'a>(
813    project: &'a Project,
814    name: &str,
815    opts: &RunOptions,
816) -> Result<Vec<(&'a Report, Binding)>, String> {
817    let reports: Option<Vec<&str>> = match &opts.selector {
818        None => None,
819        Some(s) => Some(
820            selector::resolve(project, s)
821                .map_err(|e| e.to_string())?
822                .into_iter()
823                .map(|r| r.name.as_str())
824                .collect(),
825        ),
826    };
827    let set = opts.set.as_deref().filter(|s| *s != "all");
828    let all: Vec<(&Report, &Binding)> = project
829        .reports
830        .iter()
831        .flat_map(|r| {
832            r.bindings
833                .iter()
834                .filter(|b| b.schedules.iter().any(|s| s == name))
835                .map(move |b| (r, b))
836        })
837        .collect();
838    let picked: Vec<(&Report, Binding)> = all
839        .iter()
840        .filter(|(r, _)| reports.as_ref().is_none_or(|rs| rs.contains(&r.name.as_str())))
841        .filter(|(_, b)| set.is_none() || b.set.as_deref() == set)
842        .map(|(r, b)| {
843            let mut b = (*b).clone();
844            if let Some(p) = &opts.profile {
845                b.profile = Some(p.clone());
846                b.profile_at = Some(profile_flag_at());
847            }
848            (*r, b)
849        })
850        .collect();
851    if picked.is_empty() && (reports.is_some() || set.is_some()) {
852        let mut asked = Vec::new();
853        if let Some(s) = &opts.selector {
854            asked.push(format!("`{s}`"));
855        }
856        if let Some(s) = set {
857            asked.push(format!("Set `{s}`"));
858        }
859        let runs: Vec<String> = all
860            .iter()
861            .map(|(r, b)| match &b.set {
862                Some(s) => format!("{}/{s}", r.name),
863                None => r.name.clone(),
864            })
865            .collect();
866        return Err(format!(
867            "schedule `{name}` doesn't run {}; it runs: {}",
868            asked.join(" with "),
869            if runs.is_empty() {
870                "nothing".to_string()
871            } else {
872                runs.join(", ")
873            }
874        ));
875    }
876    Ok(picked)
877}
878
879/// The usage error for `--schedule` with a name that isn't declared.
880pub fn unknown_schedule(project: &Project, name: &str) -> String {
881    let names: Vec<&str> = project.schedules.iter().map(|e| e.name.as_str()).collect();
882    if names.is_empty() {
883        format!("no schedule `{name}`: this project's schedules.yml declares none")
884    } else {
885        format!("no schedule `{name}`; valid names: {}", names.join(", "))
886    }
887}
888
889/// Which Bindings of a report run, following ADR 0009's Set rules.
890pub fn choose_bindings(
891    project: &Project,
892    report: &Report,
893    opts: &RunOptions,
894    ui: &mut dyn Ui,
895) -> Result<Vec<Binding>, String> {
896    let mut out: Vec<Binding> = match opts.set.as_deref() {
897        Some("all") => report.bindings.clone(),
898        Some(name) => match report.binding(name) {
899            Some(b) => vec![b.clone()],
900            None => vec![ad_hoc(project, report, name)],
901        },
902        None if !report.has_sets => report.bindings.clone(),
903        None => {
904            let names: Vec<String> = report.bindings.iter().filter_map(|b| b.set.clone()).collect();
905            let pick = report
906                .default_set
907                .clone()
908                .or_else(|| project.default_set.clone().filter(|d| names.contains(d)))
909                .or_else(|| (names.len() == 1).then(|| names[0].clone()));
910            match pick {
911                Some(p) => report.binding(&p).cloned().into_iter().collect(),
912                None if opts.interactive => match ui.choose_set(&report.name, &names)? {
913                    Some(p) => report.binding(&p).cloned().into_iter().collect(),
914                    None => report.bindings.clone(),
915                },
916                None => {
917                    return Err(format!(
918                        "report `{}` has several Sets ({}) and no `default_set`; declare `default_set:` or pass `--set <name>` or `--set all`",
919                        report.name,
920                        names.join(", ")
921                    ));
922                }
923            }
924        }
925    };
926    if let Some(p) = &opts.profile {
927        for b in &mut out {
928            b.profile = Some(p.clone());
929            b.profile_at = Some(profile_flag_at());
930        }
931    }
932    Ok(out)
933}
934
935/// A Set that isn't declared on the report: start from the report itself, then apply the
936/// `sets.yml` entry of that name if there is one. `--profile`/`--var` complete it.
937fn ad_hoc(project: &Project, report: &Report, name: &str) -> Binding {
938    let mut b = report.base.clone();
939    b.set = Some(name.to_string());
940    if let Some(reg) = project.sets.get(name) {
941        if let Some(p) = &reg.profile {
942            b.profile = Some(p.clone());
943            b.profile_at = Some(crate::project::ProfileAt {
944                file: reg.file.clone(),
945                line: reg.line,
946                key: format!("`profile` of Set `{name}`"),
947            });
948        }
949        b.vars.extend(reg.vars.clone());
950    }
951    b
952}
953
954/// `--profile` as the source of a Binding's inherited connection, for messages.
955fn profile_flag_at() -> crate::project::ProfileAt {
956    crate::project::ProfileAt {
957        file: PathBuf::from("--profile"),
958        line: None,
959        key: "`--profile`".into(),
960    }
961}
962
963/// The tab one query produced: its last statement's result set, spooled to disk.
964struct Produced {
965    query: String,
966    connection: String,
967    schema: SchemaRef,
968    rows: u64,
969    spool: PathBuf,
970    name: String,
971    anchor: Option<String>,
972    header: Option<bool>,
973    columns: BTreeMap<String, ColumnOptions>,
974    autofit: Option<bool>,
975    style: Option<dre_protocol::style::SheetStyle>,
976}
977
978struct Statement {
979    query: String,
980    /// The connection it runs on.
981    connection: String,
982    file: PathBuf,
983    line: usize,
984    text: String,
985    kind: StatementKind,
986    /// The complete rendered query contained a registered secret before it was split. Every
987    /// statement inherits this because splitting can itself separate a secret into fragments.
988    sensitive: bool,
989    /// Whether this statement's result is its query's tab: the file's last statement, when the
990    /// query has `tab: true`.
991    tab: bool,
992}
993
994struct BindingRun<'a> {
995    project: &'a Project,
996    report: &'a Report,
997    b: &'a Binding,
998    /// The Binding's vars with the schedule's layered on (`--var` stays separate, on top).
999    vars: JsonMap<String, Json>,
1000    /// Everything `var()` sees, `--var` included: what's recorded.
1001    rendered_vars: JsonMap<String, Json>,
1002    schedule_vars: Option<JsonMap<String, Json>>,
1003    opts: &'a RunOptions,
1004    /// `run.date`: `DRE_RUN_DATE`, else today in the run's timezone.
1005    date: NaiveDate,
1006    calendar: Calendar,
1007    ui: Events,
1008    store: &'a RunStore,
1009    compiled_dir: PathBuf,
1010    /// The Binding's folder in `target/run/`, with its runs and `current` pointer.
1011    runs: crate::runs::BindingRuns,
1012    /// This run's id and folder (`runs/<run id>/`).
1013    run_id: String,
1014    run_dir: PathBuf,
1015    /// Whether this run becomes the current one when it finishes: false when a run for a later
1016    /// instant already is (it then also leaves the drift snapshot alone).
1017    becomes_current: bool,
1018    schema_dir: PathBuf,
1019    /// The run's target (environment).
1020    target: String,
1021    /// The parse pass for this run's inputs.
1022    parsed: ParsedBinding,
1023    /// The Binding's connections and their sessions.
1024    pool: Option<Arc<Pool>>,
1025    /// One renderer per connection (`None`: the inherited one, for paths and template values).
1026    renderers: BTreeMap<Option<String>, Arc<Renderer>>,
1027    started: Instant,
1028    started_at: chrono::DateTime<chrono::Utc>,
1029    produced: Vec<Produced>,
1030    /// Each of the Binding's outputs, in order.
1031    outs: Vec<OutputRun>,
1032    /// How many tries the delivery being made took (a plugin retrying a temporary error).
1033    attempts: u32,
1034    drift: Vec<String>,
1035    /// The current schema with stable identities and any protection carried from the baseline.
1036    protected_snapshot: Option<Json>,
1037}
1038
1039/// One output's progress through the run.
1040#[derive(Default)]
1041struct OutputRun {
1042    /// Its destinations with paths and options rendered.
1043    dests: Vec<RenderedDest>,
1044    /// Each file it wrote, and where it was first delivered.
1045    files: Vec<(PathBuf, Option<String>)>,
1046    /// One record per destination: `run_results.json`'s `deliveries`.
1047    deliveries: Vec<Delivery>,
1048    delivery_note: Option<String>,
1049    /// `None` until it's done.
1050    status: Option<OutputStatus>,
1051    error: Option<String>,
1052    /// The result of its `when:`, once evaluated.
1053    when: Option<bool>,
1054    /// A message output's rendered title and text.
1055    message: Option<(String, String)>,
1056}
1057
1058/// How one output ended, as `run_results.json` records it.
1059#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1060enum OutputStatus {
1061    Delivered,
1062    /// It stays in the target path: no destination, `deliver: false`, or a preview.
1063    Kept,
1064    /// Its `when:` was false, or its message rendered empty.
1065    Skipped,
1066    Failed,
1067}
1068
1069impl OutputStatus {
1070    fn as_str(self) -> &'static str {
1071        match self {
1072            OutputStatus::Delivered => "delivered",
1073            OutputStatus::Kept => "kept",
1074            OutputStatus::Skipped => "skipped",
1075            OutputStatus::Failed => "failed",
1076        }
1077    }
1078}
1079
1080/// A destination entry after rendering.
1081struct RenderedDest {
1082    profile: String,
1083    path: Option<String>,
1084    options: JsonMap<String, Json>,
1085    /// Other outputs whose files go with a message.
1086    attach: Vec<String>,
1087}
1088
1089/// Why a Binding (or one of its steps) failed.
1090type Fail = crate::error::Error;
1091
1092fn masked_source_error(error: &HostError, sensitive: bool) -> String {
1093    let error = error.to_string();
1094    crate::secrets::mask_source_error(&error, sensitive).into_owned()
1095}
1096
1097fn masked_sql_summary(sql: &str, max: usize, sensitive: bool) -> String {
1098    if sensitive {
1099        crate::secrets::MASK.to_string()
1100    } else {
1101        dre_protocol::util::summarize(&crate::secrets::mask(sql), max)
1102    }
1103}
1104
1105fn protected_sql(sql: &str, sensitive: bool) -> std::borrow::Cow<'_, str> {
1106    if sensitive {
1107        std::borrow::Cow::Borrowed(crate::secrets::MASK)
1108    } else {
1109        crate::secrets::mask(sql)
1110    }
1111}
1112
1113impl<'a> BindingRun<'a> {
1114    fn new(
1115        project: &'a Project,
1116        report: &'a Report,
1117        b: &'a Binding,
1118        opts: &'a RunOptions,
1119        ui: Events,
1120        store: &'a RunStore,
1121    ) -> Self {
1122        let schedule = opts
1123            .schedule
1124            .as_ref()
1125            .and_then(|n| project.schedules.iter().find(|e| &e.name == n));
1126        let schedule_vars = schedule.map(|e| e.vars.clone());
1127        // Names were checked when the project and the command line were read.
1128        let calendar: Calendar = crate::parse::calendar(
1129            project,
1130            report,
1131            opts.timezone.as_deref(),
1132            opts.schedule.as_deref(),
1133        );
1134        let date = crate::parse::run_date(&calendar, opts.date, opts.scheduled_at);
1135        // Binding vars, then the schedule's, then `--var` on top.
1136        let mut vars = b.vars.clone();
1137        vars.extend(schedule_vars.clone().unwrap_or_default());
1138        let rendered_vars = {
1139            let mut v = vars.clone();
1140            v.extend(
1141                opts.vars
1142                    .iter()
1143                    .map(|(k, x)| (k.clone(), Json::String(x.clone()))),
1144            );
1145            v
1146        };
1147        ui.binding_vars(opts.schedule.as_deref(), schedule_vars.as_ref(), &rendered_vars);
1148        let started_at = chrono::Utc::now();
1149        let runs = crate::runs::BindingRuns::new(&store.run_dir(&report.name, b.dir_name()));
1150        // A dry run (compile, validate) makes no run folder: its plan shows where files would go.
1151        let run_id = if opts.dry_run || opts.live_check {
1152            "<run id>".to_string()
1153        } else {
1154            crate::runs::new_run_id(started_at)
1155        };
1156        BindingRun {
1157            project,
1158            report,
1159            b,
1160            vars,
1161            rendered_vars,
1162            schedule_vars,
1163            opts,
1164            date,
1165            calendar,
1166            ui,
1167            store,
1168            compiled_dir: store.compiled_dir(&report.name, b.dir_name()),
1169            run_dir: runs.run_dir(&run_id),
1170            runs,
1171            run_id,
1172            becomes_current: true,
1173            schema_dir: store.schema_dir(&report.name, b.dir_name()),
1174            target: project.target_name.clone(),
1175            parsed: ParsedBinding::default(),
1176            pool: None,
1177            renderers: BTreeMap::new(),
1178            started: Instant::now(),
1179            started_at,
1180            produced: Vec::new(),
1181            outs: b.outputs.iter().map(|_| OutputRun::default()).collect(),
1182            attempts: 1,
1183            drift: Vec::new(),
1184            protected_snapshot: None,
1185        }
1186    }
1187
1188    /// What `connection`, `destination` and `profile()` read for this Binding.
1189    fn connections(&self) -> Arc<ProfileConnections> {
1190        Arc::new(ProfileConnections {
1191            profiles: self.project.profiles.clone(),
1192            root: self.project.root.clone(),
1193            plugins: self.project.plugins.clone(),
1194            log: self.ui.plugin_log(),
1195        })
1196    }
1197
1198    /// The renderer for SQL on `connection` (`None`: the inherited connection, for output paths
1199    /// and template values), made on first use.
1200    fn renderer(&mut self, connection: Option<&str>) -> Result<Arc<Renderer>, Fail> {
1201        let key = connection.map(str::to_string);
1202        if let Some(r) = self.renderers.get(&key) {
1203            return Ok(r.clone());
1204        }
1205        let pool = self.pool.clone().expect("the pool is made before rendering");
1206        // Outside query SQL, `connection.*` and `run_query()` mean the inherited connection.
1207        let name = key.clone().or_else(|| self.parsed.inherited.clone());
1208        let source_type = match (&key, &name) {
1209            (Some(n), _) => pool.kind(n)?,
1210            // The inherited connection's type matters only if something uses it.
1211            (None, Some(n)) => pool.kind(n).unwrap_or_default(),
1212            (None, None) => String::new(),
1213        };
1214        let context =
1215            crate::parse::run_context(self.project, self.report, self.b, &self.inputs(), self.started_at);
1216        let limited = Arc::new(crate::render::Limited::new(
1217            context.clone(),
1218            self.vars.clone(),
1219            self.opts.vars.clone(),
1220        ));
1221        let r = Renderer::new(RendererConfig {
1222            root: &self.project.root,
1223            macros: &self.project.macros,
1224            context,
1225            vars: self.vars.clone(),
1226            cli_vars: self.opts.vars.clone(),
1227            runner: Some(Arc::new(PoolRunner {
1228                pool: pool.clone(),
1229                default: name.clone(),
1230                log: self.ui.sql_log(),
1231            })),
1232            connections: Some(pool.connections.clone()),
1233            mode: Mode::Run,
1234            connection: name,
1235            source_type,
1236            sources: crate::parse::resolver(self.project, limited),
1237            run_query_max_rows: self.project.run_query_max_rows,
1238            sql: self.project.sql.clone(),
1239            lookups: self.project.lookups.clone(),
1240            lookup_inline_max_rows: self.project.lookup_inline_max_rows,
1241            packages: self.project.packages.clone(),
1242            project_name: self.project.name.clone(),
1243            dispatch: self.project.dispatch.clone(),
1244        })
1245        .map_err(|e| e.to_string())?;
1246        let r = Arc::new(r);
1247        self.renderers.insert(key, r.clone());
1248        Ok(r)
1249    }
1250
1251    /// The parse pass's inputs for this run.
1252    fn inputs(&self) -> crate::parse::Inputs {
1253        crate::parse::Inputs {
1254            target: self.target.clone(),
1255            cli_vars: self.opts.vars.clone(),
1256            date: self.opts.date,
1257            scheduled_at: self.opts.scheduled_at,
1258            timezone: self.opts.timezone.clone(),
1259            schedule: self.opts.schedule.clone(),
1260            started_at: Some(self.started_at),
1261        }
1262    }
1263
1264    /// Each query's connection, from the parse pass.
1265    fn connection_of(&self, query: &str) -> Result<String, Fail> {
1266        self.parsed
1267            .query(query)
1268            .and_then(|q| q.connection.clone())
1269            .ok_or_else(|| format!("query `{query}` has no connection").into())
1270    }
1271
1272    /// `report` or `report (Set s)`, for messages.
1273    fn label(&self) -> String {
1274        match &self.b.set {
1275            Some(s) => format!("{} (Set {s})", self.report.name),
1276            None => self.report.name.clone(),
1277        }
1278    }
1279
1280    /// Why the Binding's lock couldn't be taken, with how to clear it.
1281    fn lock_failure(&self, e: crate::runs::LockError) -> Fail {
1282        let unlock = match &self.b.set {
1283            Some(s) => format!("dre unlock {} --binding {s}", self.report.name),
1284            None => format!("dre unlock {}", self.report.name),
1285        };
1286        let message = match e {
1287            crate::runs::LockError::Held(h) => format!(
1288                "{} is already running ({}); this run changed nothing. Wait for it, or if it's gone, `{unlock}`",
1289                self.label(),
1290                h.describe()
1291            ),
1292            crate::runs::LockError::Unreadable(p) => format!(
1293                "{} has an unreadable lock at {}; if no run is going on, `{unlock}`",
1294                self.label(),
1295                p.display()
1296            ),
1297            crate::runs::LockError::Io(e) => format!("can't lock {}: {e}", self.label()),
1298        };
1299        Fail::new(Code::RunInProgress, message)
1300    }
1301
1302    /// Whether the current run is for a later instant than this one (its `DRE_RUN_AT`, else
1303    /// its start): then this run doesn't replace it.
1304    fn older_than_current(&self) -> bool {
1305        let Some(dir) = self.runs.current_dir() else {
1306            return false;
1307        };
1308        let Ok(text) = std::fs::read_to_string(dir.join("run_results.json")) else {
1309            return false;
1310        };
1311        let Ok(v) = serde_json::from_str::<Json>(&text) else {
1312            return false;
1313        };
1314        let theirs = v
1315            .get("scheduled_at")
1316            .and_then(Json::as_str)
1317            .or_else(|| v.get("started_at").and_then(Json::as_str))
1318            .and_then(|t| chrono::DateTime::parse_from_rfc3339(t).ok())
1319            .map(|t| t.with_timezone(&chrono::Utc));
1320        let mine = self.opts.scheduled_at.unwrap_or(self.started_at);
1321        theirs.is_some_and(|t| mine < t)
1322    }
1323
1324    /// Fail once the run is cancelled, so no further statement, file or delivery starts.
1325    fn not_cancelled(&self) -> Result<(), Fail> {
1326        match self.opts.cancel.reason() {
1327            None => Ok(()),
1328            Some(crate::engine::CancelReason::Timeout) => {
1329                Err(Fail::new(Code::RunTimedOut, "the run timed out"))
1330            }
1331            Some(_) => Err(Fail::new(Code::RunCancelled, "the run was cancelled")),
1332        }
1333    }
1334
1335    fn outcome(&self, status: Status, error: Option<Fail>) -> BindingOutcome {
1336        BindingOutcome {
1337            report: self.report.name.clone(),
1338            set: self.b.set.clone(),
1339            binding: self.b.dir_name().to_string(),
1340            status,
1341            error_code: error.as_ref().map(|e| e.code.clone()),
1342            error: error.map(|e| e.message),
1343            files: self.files().map(|(p, _)| p.clone()).collect(),
1344            summary: self.summary_line(),
1345            schedule: self.opts.schedule.clone(),
1346            schedule_vars: self.schedule_vars.clone(),
1347            vars: self.rendered_vars.clone(),
1348            timezone: self.calendar.tz.name().to_string(),
1349            elapsed: self.started.elapsed(),
1350        }
1351    }
1352
1353    fn run(&mut self) -> BindingOutcome {
1354        let dry = self.opts.dry_run || self.opts.live_check;
1355        // A real run holds the Binding's lock: the same Binding never runs twice at once.
1356        let guard = if dry {
1357            None
1358        } else {
1359            match self.runs.lock(&self.run_id, self.started_at) {
1360                Ok(g) => Some(g),
1361                Err(e) => {
1362                    let error = self.lock_failure(e);
1363                    return self.outcome(Status::Error, Some(error));
1364                }
1365            }
1366        };
1367        if let Some(g) = &guard {
1368            if let Some(h) = &g.took_over {
1369                self.ui.warn(&format!(
1370                    "took over the lock of {} for {}: that process is gone (it crashed or was killed)",
1371                    h.describe(),
1372                    self.label()
1373                ));
1374            }
1375            match self.runs.migrate_legacy(self.started_at) {
1376                Ok(Some(id)) => self.ui.step(
1377                    Level::Debug,
1378                    "Moved",
1379                    &format!("the files of the run before run folders into runs/{id}/"),
1380                    None,
1381                ),
1382                Ok(None) => {}
1383                Err(e) => self
1384                    .ui
1385                    .warn(&format!("can't move the earlier run's files into runs/: {e}")),
1386            }
1387            self.becomes_current = !self.older_than_current();
1388            // The run's own log, in its folder.
1389            if std::fs::create_dir_all(&self.run_dir).is_ok() {
1390                self.ui.emit(RunEvent::BindingLog(self.run_dir.join("dre.log")));
1391            }
1392        }
1393        let result = self.run_inner();
1394        // A Binding that fails after the run was cancelled was stopped by it (a plugin's
1395        // cancelled reply, or a check between steps).
1396        let stopped =
1397            matches!(&result, Err(e) if e.code == Code::RunCancelled || e.code == Code::RunTimedOut);
1398        let result = match (result, self.opts.cancel.reason()) {
1399            (Err(e), Some(reason)) if !stopped => {
1400                let (code, what) = match reason {
1401                    crate::engine::CancelReason::Timeout => (Code::RunTimedOut, "timed out"),
1402                    _ => (Code::RunCancelled, "was cancelled"),
1403                };
1404                Err(Fail::new(code, format!("the run {what}: {}", e.message)))
1405            }
1406            (r, _) => r,
1407        };
1408        let status = match (&result, dry, self.opts.live_check) {
1409            (Err(e), _, _) if e.code == Code::RunCancelled => Status::Cancelled,
1410            (Err(e), _, _) if e.code == Code::RunTimedOut => Status::TimedOut,
1411            (Err(_), _, _) => Status::Error,
1412            (Ok(()), _, true) => Status::Checked,
1413            (Ok(()), true, _) => Status::DryRun,
1414            (Ok(()), false, _) => Status::Success,
1415        };
1416        let err = result.err();
1417        if !dry && let Err(e) = self.write_results(&status, err.as_ref()) {
1418            self.ui.warn(&format!("can't write run_results.json: {e}"));
1419        }
1420        if guard.is_some() {
1421            if self.becomes_current {
1422                if let Err(e) = self.runs.switch(&self.run_id) {
1423                    self.ui.warn(&format!("can't make this run the current one: {e}"));
1424                }
1425            } else {
1426                self.ui.warn(&format!(
1427                    "{}: a run for a later instant is current, so this one doesn't replace it; its files are in {}",
1428                    self.label(),
1429                    rel(&self.project.root, &self.run_dir).display()
1430                ));
1431            }
1432            for id in self
1433                .runs
1434                .prune(self.opts.keep_runs.unwrap_or(self.project.keep_runs))
1435            {
1436                self.ui.step(Level::Debug, "Removed", &format!("run {id}"), None);
1437            }
1438        }
1439        // Spools are scratch space.
1440        let _ = std::fs::remove_dir_all(self.run_dir.join(".spool"));
1441        self.outcome(status, err)
1442    }
1443
1444    fn run_inner(&mut self) -> Result<(), Fail> {
1445        let errors = crate::variables::check_run_values(&self.report.var_schema, &self.vars, &self.opts.vars);
1446        if !errors.is_empty() {
1447            return Err(errors.join("; ").into());
1448        }
1449        // The parse pass with this run's inputs: each query's connection and sources.
1450        self.parsed = crate::parse::binding(self.project, self.report, self.b, &self.vars, &self.inputs());
1451        if !self.parsed.errors.is_empty() {
1452            let msgs: Vec<String> = self
1453                .parsed
1454                .errors
1455                .iter()
1456                .map(|p| match p.line {
1457                    Some(l) => format!("{}:{l}: {}", p.file.display(), p.message),
1458                    None => format!("{}: {}", p.file.display(), p.message),
1459                })
1460                .collect();
1461            return Err(Fail::new(Code::ParseFailed, msgs.join("\n    ")));
1462        }
1463        // Sessions are opened only when something needs the database, so compiling a report
1464        // whose templates don't query it works without the plugin or credentials.
1465        let pool = Arc::new(Pool::new(
1466            self.connections(),
1467            self.project.root.clone(),
1468            self.ui.plugin_log(),
1469            !self.report.managed,
1470        ));
1471        self.pool = Some(pool.clone());
1472
1473        if !self.report.managed {
1474            self.ui.warn(&format!(
1475                "`{}` is an unmanaged report ({}): for quick tests only — add a YAML to make it a managed report",
1476                self.report.name,
1477                self.report.file.display()
1478            ));
1479        }
1480
1481        // 1. Render.
1482        std::fs::create_dir_all(&self.compiled_dir).map_err(|e| e.to_string())?;
1483        // How many queries run on each connection: one session holds them all.
1484        let mut per_connection: BTreeMap<String, usize> = BTreeMap::new();
1485        for q in &self.b.queries {
1486            *per_connection.entry(self.connection_of(&q.query)?).or_default() += 1;
1487        }
1488        // A real run of a managed report renders each query just before running it, so a
1489        // template can look at what earlier queries made (`columns()` of a temp table).
1490        // Compiling, checking and unmanaged reports render everything first.
1491        let interleave = !self.opts.dry_run && !self.opts.live_check && self.report.managed;
1492        let mut statements = Vec::new();
1493        // Reported after the unmanaged-report check, which matters more.
1494        let mut two_tabs: Option<String> = None;
1495        if interleave {
1496            self.render_destinations().map_err(|e| e.or(Code::RenderFailed))?;
1497            let mut checked: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
1498            for (c, n) in &per_connection {
1499                if *n > 1 {
1500                    self.check_sessions(&pool, c, *n)?;
1501                    checked.insert(c.clone());
1502                }
1503            }
1504            std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
1505            let mut i = 0;
1506            let queries = self.b.queries.clone();
1507            for q in &queries {
1508                let conn = self.connection_of(&q.query)?;
1509                let renderer = self.renderer(Some(&conn))?;
1510                let sts = self.render_query(&renderer, q, &conn, &mut two_tabs)?;
1511                for w in renderer.take_warnings() {
1512                    self.ui.warn(&w);
1513                }
1514                if let Some(e) = two_tabs.take() {
1515                    return Err(e.into());
1516                }
1517                if sts.len() > 1 && checked.insert(conn.clone()) {
1518                    self.check_sessions(&pool, &conn, sts.len())?;
1519                }
1520                let session = pool.session(&conn)?;
1521                for st in &sts {
1522                    self.execute_statement(&session, i, st)?;
1523                    i += 1;
1524                }
1525            }
1526        } else {
1527            let queries = self.b.queries.clone();
1528            for q in &queries {
1529                let conn = self.connection_of(&q.query)?;
1530                let renderer = self.renderer(Some(&conn))?;
1531                statements.extend(self.render_query(&renderer, q, &conn, &mut two_tabs)?);
1532            }
1533            self.render_destinations().map_err(|e| e.or(Code::RenderFailed))?;
1534            for r in self.renderers.values() {
1535                for w in r.take_warnings() {
1536                    self.ui.warn(&w);
1537                }
1538            }
1539
1540            // 3. Unmanaged: every rendered statement must only read (or create temp objects).
1541            if !self.report.managed {
1542                if let Some(bad) = statements.iter().find(|s| !s.kind.is_read_only_safe()) {
1543                    return Err(format!(
1544                        "{}:{}: unmanaged report `{}` may only run SELECT/WITH or CREATE [OR REPLACE] TEMP|TEMPORARY TABLE|VIEW, but found `{}`; nothing was run — rewrite the statement, or give the report a YAML to declare it",
1545                        bad.file.display(),
1546                        bad.line,
1547                        self.report.name,
1548                        masked_sql_summary(&bad.text, 60, bad.sensitive)
1549                    ).into());
1550                }
1551                for conn in per_connection.keys() {
1552                    let session = pool.session(conn)?;
1553                    let creates_temp = statements
1554                        .iter()
1555                        .any(|s| &s.connection == conn && s.kind == StatementKind::TempCreate)
1556                        || session.lock().unwrap().loaded;
1557                    session.lock().unwrap().want_read_only(!creates_temp);
1558                }
1559            }
1560            if let Some(e) = two_tabs {
1561                return Err(e.into());
1562            }
1563
1564            if self.opts.dry_run {
1565                let plan = self.plan();
1566                self.ui.compiled(&plan);
1567                return Ok(());
1568            }
1569
1570            // The one-session-per-connection guarantee.
1571            for conn in per_connection.keys() {
1572                let n = statements.iter().filter(|s| &s.connection == conn).count();
1573                if n > 1 {
1574                    self.check_sessions(&pool, conn, n)?;
1575                }
1576            }
1577
1578            if self.opts.live_check {
1579                return self.live_check(&statements);
1580            }
1581
1582            // 4. Execute in YAML order, each on its connection's session.
1583            std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
1584            for (i, st) in statements.iter().enumerate() {
1585                let session = pool.session(&st.connection)?;
1586                self.execute_statement(&session, i, st)?;
1587            }
1588        }
1589        pool.close_all();
1590
1591        self.name_result_sets()?;
1592        if let Some(dir) = &self.opts.output_json {
1593            let dir = self
1594                .project
1595                .root
1596                .join(dir)
1597                .join(&self.report.name)
1598                .join(self.b.dir_name());
1599            std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
1600            for result in &self.produced {
1601                crate::rows::write_spool(
1602                    &result.spool,
1603                    &dir.join(format!("{}.json", result.name)),
1604                    result.rows,
1605                )
1606                .map_err(|e| e.to_string())?;
1607            }
1608        }
1609
1610        // 5. Format every file output into target/run/; one whose `when:` is false is skipped.
1611        let (file_outs, message_outs): (Vec<usize>, Vec<usize>) =
1612            (0..self.outs.len()).partition(|&i| !self.b.outputs[i].is_message());
1613        for &oi in &file_outs {
1614            if !self.passes_when(oi)? {
1615                continue;
1616            }
1617            let filename = self.file_names(oi);
1618            self.format(oi, &filename).map_err(|e| e.or(Code::FormatFailed))?;
1619        }
1620
1621        // Schema drift, before delivery.
1622        if self.opts.preview.is_none() {
1623            self.drift = self.schema_drift();
1624            if !self.drift.is_empty() {
1625                if self.opts.accept_schema_change {
1626                    self.ui.warn(&format!(
1627                        "  schema changed since the last successful run (accepted): {}",
1628                        self.drift.join("; ")
1629                    ));
1630                } else {
1631                    return Err(format!(
1632                        "schema drift since the last successful run: {}; the output is in {} but was not delivered — pass --accept-schema-change to deliver it and accept the new schema",
1633                        self.drift.join("; "),
1634                        rel(&self.project.root, &self.run_dir).display()
1635                    ).into());
1636                }
1637            }
1638        }
1639
1640        // 6. Deliver the file outputs, then render and deliver each message, so a message can
1641        // link to what was delivered. A failed output doesn't stop the next one.
1642        let preview = self.opts.preview.is_some() || self.opts.no_deliver;
1643        let mut failures = Vec::new();
1644        for oi in file_outs.into_iter().chain(message_outs) {
1645            if self.outs[oi].status == Some(OutputStatus::Skipped) {
1646                continue;
1647            }
1648            if self.b.outputs[oi].is_message() {
1649                match self.render_message(oi) {
1650                    Ok(true) => {}
1651                    Ok(false) => continue,
1652                    Err(e) => {
1653                        let e = e.or(Code::RenderFailed);
1654                        let e = Fail::new(e.code.clone(), format!("{}: {e}", self.b.outputs[oi].label(oi)));
1655                        self.outs[oi].status = Some(OutputStatus::Failed);
1656                        self.outs[oi].error = Some(e.message.clone());
1657                        failures.push(e);
1658                        continue;
1659                    }
1660                }
1661            }
1662            if preview {
1663                self.outs[oi].delivery_note = Some(if self.opts.no_deliver {
1664                    "--no-deliver: output kept locally".into()
1665                } else {
1666                    "preview: not delivered".into()
1667                });
1668                self.outs[oi].status = Some(OutputStatus::Kept);
1669            } else if let Err(e) = self.deliver(oi) {
1670                failures.push(e.or(Code::DeliveryFailed));
1671            }
1672        }
1673        if preview {
1674            let dir = rel(&self.project.root, &self.run_dir);
1675            self.ui.step(
1676                Level::Info,
1677                "Preview",
1678                &format!("not delivered; output stays in {}", dir.display()),
1679                None,
1680            );
1681        }
1682        if let Some(first) = failures.first() {
1683            let messages: Vec<&str> = failures.iter().map(|f| f.message.as_str()).collect();
1684            return Err(Fail::new(first.code.clone(), messages.join("; ")));
1685        }
1686
1687        // 7. Snapshot the schema for the next drift check (not from a run for an earlier instant
1688        // than the current one).
1689        if self.opts.preview.is_none()
1690            && self.becomes_current
1691            && let Err(error) = self.write_snapshot()
1692        {
1693            self.ui.warn(&format!(
1694                "delivery completed, but the schema snapshot could not be written: {error}; the next run may compare against an older schema"
1695            ));
1696        }
1697        Ok(())
1698    }
1699
1700    /// Output `oi`'s queries' results, `rows` capped at `max_rows`.
1701    fn results_for(&mut self, oi: usize, max_rows: u64) -> Result<Vec<(String, Arc<QueryResult>)>, Fail> {
1702        let o = &self.b.outputs[oi];
1703        let mut out = Vec::new();
1704        for p in self.produced.iter().filter(|p| o.feeds(&p.query)) {
1705            let r = QueryResult::read(&p.spool, p.rows, max_rows, self.calendar)
1706                .map_err(|e| format!("can't read the result of `{}`: {e}", p.query))?;
1707            out.push((p.query.clone(), Arc::new(r)));
1708        }
1709        Ok(out)
1710    }
1711
1712    /// What output `oi`'s templates read besides the usual context: `results` and `outputs`
1713    /// (every other named output: `location`, `files` and `status`).
1714    fn template_context(&self, oi: usize, results: &[(String, Arc<QueryResult>)]) -> minijinja::Value {
1715        use minijinja::Value;
1716        let results: BTreeMap<String, Value> = results
1717            .iter()
1718            .map(|(q, r)| (q.clone(), r.clone().value()))
1719            .collect();
1720        let mut outputs: BTreeMap<String, Value> = BTreeMap::new();
1721        for (j, (o, run)) in self.b.outputs.iter().zip(&self.outs).enumerate() {
1722            let Some(name) = &o.name else { continue };
1723            if j == oi {
1724                continue;
1725            }
1726            let locations: Vec<&str> = run
1727                .deliveries
1728                .iter()
1729                .filter_map(|d| d.location.as_deref())
1730                .collect();
1731            let files: Vec<String> = run
1732                .files
1733                .iter()
1734                .map(|(f, _)| record_path(self.project, f).to_string_lossy().to_string())
1735                .collect();
1736            let status = run.status.map_or("pending", OutputStatus::as_str);
1737            outputs.insert(
1738                name.clone(),
1739                Value::from_serialize(json!({
1740                    "location": locations.join(", "),
1741                    "files": files,
1742                    "status": status,
1743                })),
1744            );
1745        }
1746        Value::from(BTreeMap::from([
1747            ("results".to_string(), Value::from(results)),
1748            ("outputs".to_string(), Value::from(outputs)),
1749        ]))
1750    }
1751
1752    /// Evaluate output `oi`'s `when:`. False: the output is skipped (recorded, not a failure).
1753    fn passes_when(&mut self, oi: usize) -> Result<bool, Fail> {
1754        if self.b.outputs[oi].when.is_none() {
1755            return Ok(true);
1756        }
1757        let results = self.results_for(oi, crate::message::DEFAULT_MAX_ROWS)?;
1758        self.when_holds(oi, &results)
1759    }
1760
1761    /// Output `oi`'s `when:` over `results` (the same ones its message reads).
1762    fn when_holds(&mut self, oi: usize, results: &[(String, Arc<QueryResult>)]) -> Result<bool, Fail> {
1763        let Some(when) = self.b.outputs[oi].when.clone() else {
1764            return Ok(true);
1765        };
1766        let ctx = self.template_context(oi, results);
1767        let renderer = self.renderer(None)?;
1768        let label = self.b.outputs[oi].label(oi);
1769        let src = format!("{{% if {when} %}}true{{% endif %}}");
1770        let passed = renderer
1771            .render_with(&self.report.file, &src, ctx, false)
1772            .map_err(|e| format!("{label}: `when` `{when}`: {}", e.message))?
1773            == "true";
1774        self.outs[oi].when = Some(passed);
1775        if !passed {
1776            self.skip(oi, &format!("`when` is false ({when})"));
1777        }
1778        Ok(passed)
1779    }
1780
1781    fn skip(&mut self, oi: usize, why: &str) {
1782        let label = self.b.outputs[oi].label(oi);
1783        self.outs[oi].status = Some(OutputStatus::Skipped);
1784        self.outs[oi].delivery_note = Some(format!("skipped: {why}"));
1785        self.ui
1786            .step(Level::Info, "Skipped", &format!("{label}: {why}"), None);
1787    }
1788
1789    /// Render message output `oi` and write its `.md` file. `Ok(false)`: skipped (`when:` false,
1790    /// or the text rendered empty).
1791    fn render_message(&mut self, oi: usize) -> Result<bool, Fail> {
1792        let o = &self.b.outputs[oi];
1793        let opt = |k: &str| o.options.get(k).and_then(Json::as_str).map(str::to_string);
1794        let (text_src, file_src, title_src) = (opt("text"), opt("file"), opt("title"));
1795        let max_rows = o
1796            .options
1797            .get("max_rows")
1798            .and_then(Json::as_u64)
1799            .unwrap_or(crate::message::DEFAULT_MAX_ROWS);
1800        let results = self.results_for(oi, max_rows)?;
1801        if !self.when_holds(oi, &results)? {
1802            return Ok(false);
1803        }
1804        for (q, r) in results.iter().filter(|(_, r)| r.capped()) {
1805            self.ui.warn(&format!(
1806                "  results.{q}.rows holds the first {} of {} rows (`max_rows`); aggregate in SQL, or raise `max_rows`",
1807                thousands(r.rows.len() as u64),
1808                thousands(r.row_count)
1809            ));
1810        }
1811        let ctx = self.template_context(oi, &results);
1812        let renderer = self.renderer(None)?;
1813        let file = self.report.file.clone();
1814        let text = match (text_src, file_src) {
1815            (Some(t), _) => renderer
1816                .render_with(&file, &t, ctx.clone(), true)
1817                .map_err(|e| e.message)?,
1818            (None, Some(f)) => {
1819                let root = &self.project.root;
1820                let path = crate::project::find_template(root, &f)
1821                    .ok_or_else(|| format!("message file `{f}` doesn't exist"))?;
1822                let src = std::fs::read_to_string(&path).map_err(|e| format!("{}: {e}", path.display()))?;
1823                let shown = rel(root, &path);
1824                renderer
1825                    .render_with(&shown, &src, ctx.clone(), true)
1826                    .map_err(|e| e.to_string())?
1827            }
1828            (None, None) => crate::message::default_text(&results, self.locale()),
1829        };
1830        let text = text.trim().to_string();
1831        if text.is_empty() {
1832            self.skip(oi, "the message is empty");
1833            return Ok(false);
1834        }
1835        let title = match title_src {
1836            Some(t) => renderer
1837                .render_with(&file, &t, ctx, false)
1838                .map_err(|e| e.message)?,
1839            None => format!("{}: {}", self.report.name, self.date),
1840        };
1841        let title = title.split_whitespace().collect::<Vec<_>>().join(" ");
1842        let name = self.file_names(oi);
1843        let path = self.run_dir.join(&name);
1844        if let Some(prev) = (0..oi).find(|&j| self.outs[j].files.iter().any(|(f, _)| f == &path)) {
1845            return Err(format!(
1846                "it writes {name}, as {} does; give them different `name:`s",
1847                self.b.outputs[prev].label(prev)
1848            )
1849            .into());
1850        }
1851        std::fs::create_dir_all(&self.run_dir).map_err(|e| e.to_string())?;
1852        std::fs::write(&path, crate::message::file_body(&title, &text)).map_err(|e| e.to_string())?;
1853        self.outs[oi].files.push((path, None));
1854        // Protect preview and metadata copies; report files and deliveries retain their values.
1855        let shown_title = crate::secrets::mask_message_title(&title);
1856        let shown_text = crate::secrets::mask_markdown(&text);
1857        if self.opts.preview.is_some() {
1858            let mut notes = Vec::new();
1859            if let Some(n) = self.opts.preview {
1860                notes.push(format!(
1861                    "numbers come from a sample of at most {} rows per query (--preview)",
1862                    thousands(n)
1863                ));
1864            }
1865            if let Some(w) = &self.b.outputs[oi].when {
1866                notes.push(format!("`when` passed ({w})"));
1867            }
1868            let length = text.chars().count();
1869            notes.push(format!("{} characters", thousands(length as u64)));
1870            notes.extend(self.destination_limits(oi, length));
1871            self.ui.message(&ShownMessage {
1872                output: self.b.outputs[oi].name.clone(),
1873                title: shown_title.into_owned(),
1874                text: shown_text.into_owned(),
1875                notes,
1876            });
1877        } else {
1878            self.ui.step(
1879                Level::Info,
1880                "Message",
1881                &crate::message::excerpt(&title, &text, 100),
1882                None,
1883            );
1884        }
1885        self.outs[oi].message = Some((title, text));
1886        Ok(true)
1887    }
1888
1889    /// For `--preview`: how each destination of message output `oi` takes it, and its limit.
1890    fn destination_limits(&self, oi: usize, length: usize) -> Vec<String> {
1891        let mut out = Vec::new();
1892        let Some(pool) = &self.pool else { return out };
1893        for d in &self.outs[oi].dests {
1894            let Some(t) = self.project.profiles.target(Role::Destination, &d.profile) else {
1895                out.push(format!("{}: delivers nowhere on this target", d.profile));
1896                continue;
1897            };
1898            let described = pool.connections.described(Role::Destination, &t.kind);
1899            let takes = described
1900                .as_ref()
1901                .is_some_and(|x| x.capabilities.iter().any(|c| c == CAP_MESSAGE));
1902            let limit = described.and_then(|x| x.message_limit);
1903            out.push(match (takes, limit) {
1904                (false, _) => format!("{} ({}): delivers the .md file", d.profile, t.kind),
1905                (true, Some(l)) if length as u64 > l => format!(
1906                    "{} ({}): {} of {} characters: over the limit, so it will be cut short",
1907                    d.profile,
1908                    t.kind,
1909                    thousands(length as u64),
1910                    thousands(l)
1911                ),
1912                (true, Some(l)) => format!(
1913                    "{} ({}): {} of {} characters",
1914                    d.profile,
1915                    t.kind,
1916                    thousands(length as u64),
1917                    thousands(l)
1918                ),
1919                (true, None) => format!("{} ({}): posts the message", d.profile, t.kind),
1920            });
1921        }
1922        out
1923    }
1924
1925    /// The Binding's locale for the number filters.
1926    fn locale(&self) -> crate::numbers::Locale {
1927        self.b
1928            .locale
1929            .as_deref()
1930            .and_then(|l| crate::numbers::Locale::parse(l).ok())
1931            .unwrap_or_default()
1932    }
1933
1934    /// Render one query file into its statements, writing it to `target/compiled/`.
1935    fn render_query(
1936        &mut self,
1937        renderer: &Renderer,
1938        q: &QueryEntry,
1939        connection: &str,
1940        two_tabs: &mut Option<String>,
1941    ) -> Result<Vec<Statement>, Fail> {
1942        let mut out = Vec::new();
1943        let src = std::fs::read_to_string(self.project.root.join(&q.path))
1944            .map_err(|e| format!("{}: {e}", q.path.display()))?;
1945        let _ = renderer.take_sources();
1946        let sql = renderer
1947            .render(&q.path, &src)
1948            .map_err(|e: RenderError| e.to_string())?;
1949        // The manifest's `depends_on` and the query's connection come from the parse pass, so
1950        // a `source()` it couldn't see would make both wrong.
1951        let known = self
1952            .parsed
1953            .query(&q.query)
1954            .map(|p| p.sources.clone())
1955            .unwrap_or_default();
1956        if let Some((s, t)) = renderer
1957            .take_sources()
1958            .into_iter()
1959            .find(|(s, t)| !known.contains(&format!("{s}.{t}")))
1960        {
1961            return Err(format!(
1962                "{}: `source('{s}', '{t}')` was reached while rendering `{}`, but the parse pass didn't find it. The parse pass renders without data (`run_query()` returns no rows, `connection.*` nothing), so it can't see a `source()` that depends on them; call it where it's reached either way",
1963                q.path.display(),
1964                q.query
1965            ).into());
1966        }
1967        let sensitive = crate::secrets::contains_secret(&sql);
1968        std::fs::write(
1969            self.compiled_dir.join(format!("{}.sql", q.query)),
1970            crate::secrets::mask(&sql).as_bytes(),
1971        )
1972        .map_err(|e| e.to_string())?;
1973        // Hash the exact rendered SQL before masking display artifacts. Integrations can
1974        // identify executed SQL without receiving any secret embedded in it.
1975        {
1976            use sha2::Digest;
1977            let digest = sha2::Sha256::digest(sql.as_bytes());
1978            let hash = digest
1979                .iter()
1980                .map(|byte| format!("{byte:02x}"))
1981                .collect::<String>();
1982            std::fs::write(self.compiled_dir.join(format!("{}.sha256", q.query)), hash)
1983                .map_err(|e| e.to_string())?;
1984        }
1985        self.ui
1986            .step(Level::Debug, "Rendered", &q.path.display().to_string(), None);
1987        // 2. Split. One file makes at most one tab, from its last statement, so any earlier
1988        // SELECT would be lost: that's an error, not a guess.
1989        let parts = sqlsplit::split(&sql);
1990        if q.tab && parts.is_empty() {
1991            return Err(format!(
1992                "{}: `{}` makes a tab but has no statements; add a query, or `tab: false` if it's meant to be empty",
1993                q.path.display(),
1994                q.query
1995            ).into());
1996        }
1997        let n = parts.len();
1998        for (i, st) in parts.into_iter().enumerate() {
1999            let body = sqlsplit::strip_leading_comments(&st.text);
2000            let line = st.line + st.text[..st.text.len() - body.len()].matches('\n').count();
2001            let kind = sqlsplit::classify(&st.text);
2002            let last = i + 1 == n;
2003            if q.tab && !last && kind == StatementKind::Read && two_tabs.is_none() {
2004                *two_tabs = Some(format!(
2005                    "{}:{line}: `{}` has more than one SELECT, but one .sql file makes one tab (from its last statement); put each tab's query in its own .sql file and list each in `queries:`",
2006                    q.path.display(),
2007                    q.query
2008                ));
2009            }
2010            out.push(Statement {
2011                query: q.query.clone(),
2012                connection: connection.to_string(),
2013                file: q.path.clone(),
2014                line,
2015                kind,
2016                text: st.text,
2017                sensitive,
2018                tab: q.tab && last,
2019            });
2020        }
2021        Ok(out)
2022    }
2023
2024    /// Fail unless `connection`'s plugin can hold one session across `n` statements.
2025    fn check_sessions(&self, pool: &Pool, connection: &str, n: usize) -> Result<(), Fail> {
2026        let session = pool.session(connection)?;
2027        let mut s = session.lock().unwrap();
2028        if !s
2029            .get()
2030            .map_err(|e| Fail::new(Code::ConnectionFailed, e))?
2031            .has(CAP_SESSIONS)
2032        {
2033            let kind = pool.kind(connection).unwrap_or_default();
2034            return Err(format!(
2035                "this Binding runs {n} statements on connection `{connection}`, but the `{kind}` source plugin can't hold one session across them; nothing was run"
2036            ).into());
2037        }
2038        Ok(())
2039    }
2040
2041    /// Run statement `i`, spooling its result if it makes a tab.
2042    fn execute_statement(
2043        &mut self,
2044        session: &Arc<Mutex<Session>>,
2045        i: usize,
2046        st: &Statement,
2047    ) -> Result<(), Fail> {
2048        self.not_cancelled()?;
2049        let sql_log = self.ui.sql_log();
2050        sql_log(
2051            &format!("{}:{}", st.file.display(), st.line),
2052            &protected_sql(&st.text, st.sensitive),
2053        );
2054        let spool_path = self.run_dir.join(".spool").join(format!("{i}.arrow"));
2055        let mut writer: Option<FileWriter<File>> = None;
2056        let t = Instant::now();
2057        let exec = {
2058            let mut s = session.lock().unwrap();
2059            let (p, _log_scope) = s
2060                .statement(st.sensitive)
2061                .map_err(|e| Fail::new(Code::ConnectionFailed, e))?;
2062            p.execute(&st.text, self.opts.preview, |schema, batch| {
2063                if writer.is_none() {
2064                    let f = File::create(&spool_path).map_err(|e| e.to_string())?;
2065                    writer = Some(FileWriter::try_new(f, schema).map_err(|e| e.to_string())?);
2066                }
2067                writer.as_mut().unwrap().write(&batch).map_err(|e| e.to_string())
2068            })
2069            .map_err(|e| {
2070                let message = format!(
2071                    "{}:{}: {}",
2072                    st.file.display(),
2073                    st.line,
2074                    masked_source_error(&e, st.sensitive)
2075                );
2076                Fail::from_plugin(&e, message).or(Code::QueryFailed)
2077            })?
2078        };
2079        let what = match &exec {
2080            Execution::Result { rows, .. } => {
2081                format!("{} row{}", thousands(*rows), if *rows == 1 { "" } else { "s" })
2082            }
2083            Execution::NoResult {
2084                rows_affected: Some(n),
2085            } => format!("no result set ({n} affected)"),
2086            Execution::NoResult { .. } => "no result set".to_string(),
2087        };
2088        let at = format!("{}:{}", st.file.display(), st.line);
2089        self.ui.step(
2090            Level::Debug,
2091            "Executed",
2092            &format!("{at}  {what}"),
2093            Some(t.elapsed()),
2094        );
2095        // The YAML decides the tabs; the result only has to match it. A tab with no rows
2096        // still gets its column names.
2097        let (schema, rows) = match (st.tab, exec) {
2098            (true, Execution::Result { schema, rows }) => (schema, rows),
2099            (true, Execution::NoResult { .. }) => {
2100                return Err(Fail::new(
2101                    Code::NoResultSet,
2102                    format!(
2103                        "{at}: `{}` makes a tab, but its last statement returned no result set; if the query only prepares data (a temp view, a SET), add `tab: false` to it in the YAML",
2104                        st.query
2105                    ),
2106                ));
2107            }
2108            (false, _) => {
2109                drop(writer);
2110                let _ = std::fs::remove_file(&spool_path);
2111                return Ok(());
2112            }
2113        };
2114        {
2115            let mut w = match writer {
2116                Some(w) => w,
2117                None => FileWriter::try_new(File::create(&spool_path).map_err(|e| e.to_string())?, &schema)
2118                    .map_err(|e| e.to_string())?,
2119            };
2120            w.finish().map_err(|e| e.to_string())?;
2121            self.produced.push(Produced {
2122                query: st.query.clone(),
2123                connection: st.connection.clone(),
2124                schema,
2125                rows,
2126                spool: spool_path,
2127                name: String::new(),
2128                anchor: None,
2129                header: None,
2130                columns: Default::default(),
2131                autofit: None,
2132                style: None,
2133            });
2134        }
2135        Ok(())
2136    }
2137
2138    /// Each destination's rendered `profile`, `path` and options; `destination.*` is that
2139    /// destination while its values render.
2140    fn render_destinations(&mut self) -> Result<(), Fail> {
2141        if self.opts.no_deliver {
2142            return Ok(());
2143        }
2144        if self.b.destinations().next().is_none() {
2145            return Ok(());
2146        }
2147        let renderer = self.renderer(None)?;
2148        let connections = self.pool.as_ref().expect("pool").connections.clone();
2149        // Destinations are numbered across every output, as the parse pass renders them.
2150        let numbered: Vec<(usize, usize, &crate::project::Destination)> = self
2151            .b
2152            .outputs
2153            .iter()
2154            .enumerate()
2155            .flat_map(|(oi, o)| o.destinations.iter().map(move |d| (oi, d)))
2156            .enumerate()
2157            .map(|(i, (oi, d))| (i, oi, d))
2158            .collect();
2159        for (i, oi, d) in numbered {
2160            let profile = self
2161                .parsed
2162                .destinations
2163                .get(i)
2164                .cloned()
2165                .flatten()
2166                .ok_or_else(|| format!("destination `{}` has no profile", d.profile))?;
2167            // Nothing is delivered, so there's no path or options to render.
2168            if matches!(
2169                self.project.profiles.entry(Role::Destination, &profile),
2170                Entry::Nowhere
2171            ) {
2172                self.outs[oi].dests.push(RenderedDest {
2173                    profile,
2174                    path: None,
2175                    options: JsonMap::new(),
2176                    attach: Vec::new(),
2177                });
2178                continue;
2179            }
2180            renderer.set_destination(connections.profile(&profile, Some("destination")).ok());
2181            let rendered = (|| {
2182                let path = match &d.path {
2183                    Some(p) => Some(
2184                        renderer
2185                            .render(&self.report.file, p)
2186                            .map_err(|e| format!("output path: {e}"))?,
2187                    ),
2188                    None => None,
2189                };
2190                let mut options = JsonMap::new();
2191                for (k, v) in &d.options {
2192                    let v = render_json(&renderer, &self.report.file, v)
2193                        .map_err(|e| format!("destination `{profile}` option `{k}`: {e}"))?;
2194                    options.insert(k.clone(), v);
2195                }
2196                Ok::<_, Fail>(RenderedDest {
2197                    profile: profile.clone(),
2198                    path,
2199                    options,
2200                    attach: d.attach.clone(),
2201                })
2202            })();
2203            renderer.set_destination(None);
2204            self.outs[oi].dests.push(rendered?);
2205        }
2206        Ok(())
2207    }
2208
2209    /// Tab names: `tab_name`, else the query's basename.
2210    fn name_result_sets(&mut self) -> Result<(), Fail> {
2211        let entries: BTreeMap<&str, &QueryEntry> =
2212            self.b.queries.iter().map(|q| (q.query.as_str(), q)).collect();
2213        for p in &mut self.produced {
2214            let q = entries[p.query.as_str()];
2215            p.name = q.tab_name.clone().unwrap_or_else(|| q.query.clone());
2216            p.anchor = q.anchor.clone();
2217            p.header = q.header;
2218            p.columns = q.columns.clone();
2219            p.autofit = q.autofit;
2220            p.style = q.style.clone();
2221        }
2222        for o in self.b.outputs.iter().filter(|o| o.format == "xlsx") {
2223            let mut seen: BTreeMap<String, String> = BTreeMap::new();
2224            for p in self.produced.iter().filter(|p| o.feeds(&p.query)) {
2225                check_sheet_name(&p.name)?;
2226                if let Some(prev) = seen.insert(p.name.to_lowercase(), p.query.clone()) {
2227                    return Err(format!(
2228                        "sheet name `{}` is used twice (by `{prev}` and `{}`); Excel sheet names must be unique",
2229                        p.name, p.query
2230                    ).into());
2231                }
2232            }
2233        }
2234        Ok(())
2235    }
2236
2237    /// What a real run of this Binding would use and produce, for `dre compile` and `validate`.
2238    fn plan(&mut self) -> BindingPlan {
2239        let compiled = self
2240            .b
2241            .queries
2242            .iter()
2243            .map(|q| {
2244                rel(
2245                    &self.project.root,
2246                    &self.compiled_dir.join(format!("{}.sql", q.query)),
2247                )
2248            })
2249            .collect();
2250        let mut outputs = Vec::new();
2251        for oi in 0..self.outs.len() {
2252            let file = self.file_names(oi);
2253            let o = &self.b.outputs[oi];
2254            let destinations = self.outs[oi]
2255                .dests
2256                .iter()
2257                .map(|d| {
2258                    let out = self.project.profiles.target(Role::Destination, &d.profile);
2259                    PlannedDestination {
2260                        profile: d.profile.clone(),
2261                        kind: out.map(|o| o.kind.clone()),
2262                        target: Some(self.project.profiles.target_of(Role::Destination, &d.profile)),
2263                        path: d.path.clone(),
2264                        delivers: out.is_some(),
2265                    }
2266                })
2267                .collect();
2268            outputs.push(PlannedOutput {
2269                name: o.name.clone(),
2270                format: o.format.clone(),
2271                output: rel(&self.project.root, &self.run_dir.join(file)),
2272                destinations,
2273            });
2274        }
2275        let first = outputs.first().cloned();
2276        let pool = self.pool.clone().expect("pool");
2277        let queries = self
2278            .parsed
2279            .queries
2280            .iter()
2281            .map(|q| {
2282                let connection = q.connection.clone().unwrap_or_default();
2283                PlannedQuery {
2284                    query: q.query.clone(),
2285                    kind: pool.kind(&connection).unwrap_or_default(),
2286                    target: self.project.profiles.target_of(Role::Connection, &connection),
2287                    connection,
2288                    sources: q.sources.clone(),
2289                }
2290            })
2291            .collect();
2292        BindingPlan {
2293            report: self.report.name.clone(),
2294            set: self.b.set.clone(),
2295            compiled,
2296            profile: self.parsed.inherited.clone(),
2297            queries,
2298            target: self.target.clone(),
2299            format: first.as_ref().map(|o| o.format.clone()).unwrap_or_default(),
2300            output: first.as_ref().map(|o| o.output.clone()).unwrap_or_default(),
2301            destinations: first.map(|o| o.destinations).unwrap_or_default(),
2302            outputs,
2303            schedules: self.b.schedules.clone(),
2304        }
2305    }
2306
2307    /// Apply `--output-path`/`--output-name` to every destination of the first output that has a
2308    /// path, and return output `oi`'s local file name: `--output-name` (first output only), else
2309    /// the first destination path's file name, else `<name>.<ext>` (the output's `name:`, else
2310    /// the report's).
2311    fn file_names(&mut self, oi: usize) -> String {
2312        let first = oi == 0;
2313        let (output_path, output_name) = if first {
2314            (self.opts.output_path.clone(), self.opts.output_name.clone())
2315        } else {
2316            (None, None)
2317        };
2318        for d in self.outs[oi].dests.iter_mut().filter(|d| d.path.is_some()) {
2319            if let Some(p) = &output_path {
2320                d.path = Some(p.clone());
2321            }
2322            if let (Some(n), Some(r)) = (&output_name, d.path.as_mut()) {
2323                *r = match r.rfind(['/', '\\']) {
2324                    Some(i) => format!("{}{n}", &r[..=i]),
2325                    None => n.clone(),
2326                };
2327            }
2328        }
2329        let o = &self.b.outputs[oi];
2330        let ext = match &o.extension {
2331            Some(e) => e.as_str(),
2332            None => extension(&o.format),
2333        };
2334        let stem = o.name.clone().unwrap_or_else(|| self.report.name.clone());
2335        let from_remote = self.outs[oi]
2336            .dests
2337            .iter()
2338            .find_map(|d| d.path.as_deref())
2339            .and_then(|r| r.rsplit(['/', '\\']).next())
2340            .filter(|s| !s.is_empty())
2341            .map(str::to_string);
2342        output_name.or(from_remote).unwrap_or_else(|| match ext {
2343            "" => stem,
2344            ext => format!("{stem}.{ext}"),
2345        })
2346    }
2347
2348    /// Write output `oi`'s files from the result sets of its queries.
2349    fn format(&mut self, oi: usize, filename: &str) -> Result<(), Fail> {
2350        self.not_cancelled()?;
2351        std::fs::create_dir_all(&self.run_dir).map_err(|e| e.to_string())?;
2352        let b = self.b;
2353        let out = &b.outputs[oi];
2354        let mine: Vec<usize> = (0..self.produced.len())
2355            .filter(|&i| out.feeds(&self.produced[i].query))
2356            .collect();
2357        if mine.is_empty() {
2358            let what = if self.outs.len() > 1 {
2359                format!("{} gets no tab", out.label(oi))
2360            } else {
2361                "no query makes a tab (every query has `tab: false`)".to_string()
2362            };
2363            self.ui.warn(&format!("  {what}; no output file was written"));
2364            return Ok(());
2365        }
2366        let multi = out.format == "xlsx" || out.template.is_some() || !is_single_table(&out.format);
2367        let groups: Vec<(String, Vec<usize>)> = if multi || mine.len() == 1 {
2368            vec![(filename.to_string(), mine)]
2369        } else {
2370            let (stem, ext) = split_ext(filename);
2371            let mut g = Vec::new();
2372            for i in mine {
2373                let r = &self.produced[i];
2374                if r.name.contains(['/', '\\']) {
2375                    return Err(format!("`{}` can't be used in a file name", r.name).into());
2376                }
2377                g.push((format!("{stem}_{}{ext}", r.name), vec![i]));
2378            }
2379            g
2380        };
2381        // Two outputs writing one file would overwrite each other.
2382        for (name, _) in &groups {
2383            let path = self.run_dir.join(name);
2384            if let Some(prev) = (0..oi).find(|&j| self.outs[j].files.iter().any(|(f, _)| f == &path)) {
2385                return Err(format!(
2386                    "{} and {} both write {name}; give them different `name:`s or destination paths",
2387                    self.b.outputs[prev].label(prev),
2388                    out.label(oi)
2389                )
2390                .into());
2391            }
2392        }
2393        let plugin = find_plugin(self.project, PluginKind::Format, &out.format)
2394            .map_err(|e| Fail::new(Code::FormatFailed, e.to_string()))?;
2395        let mut p = plugin
2396            .start(self.ui.plugin_log(), Some(&self.project.root))
2397            .map_err(|e| e.to_string())?;
2398        let template = match &out.template {
2399            Some(t) => Some(self.template_payload(t)?),
2400            None => None,
2401        };
2402        for (name, idxs) in groups {
2403            let t = Instant::now();
2404            let path = self.run_dir.join(&name);
2405            let metas: Vec<ResultSetMeta> = idxs
2406                .iter()
2407                .map(|&i| {
2408                    let r = &self.produced[i];
2409                    ResultSetMeta {
2410                        name: r.name.clone(),
2411                        query: r.query.clone(),
2412                        result_index: 1,
2413                        anchor: r.anchor.clone(),
2414                        header: r.header,
2415                        columns: r.columns.clone(),
2416                        autofit: r.autofit,
2417                        style: r.style.clone(),
2418                    }
2419                })
2420                .collect();
2421            p.write_begin(
2422                &path.to_string_lossy(),
2423                &out.format,
2424                out.options.clone(),
2425                metas,
2426                template.clone(),
2427            )
2428            .map_err(|e| e.to_string())?;
2429            for &i in &idxs {
2430                let r = &self.produced[i];
2431                let reader = FileReader::try_new(File::open(&r.spool).map_err(|e| e.to_string())?, None)
2432                    .map_err(|e| e.to_string())?;
2433                let schema = r.schema.clone();
2434                // Stream batch by batch so memory stays flat.
2435                let mut any = false;
2436                for b in reader {
2437                    let b = b.map_err(|e| e.to_string())?;
2438                    any = true;
2439                    p.send_batch(&b)
2440                        .map_err(|e| format!("{} format: {e}", out.format))?;
2441                }
2442                if !any {
2443                    p.send_batch(&RecordBatch::new_empty(schema))
2444                        .map_err(|e| format!("{} format: {e}", out.format))?;
2445                }
2446                p.send(&dre_protocol::msg::Request::ResultSetEnd {})
2447                    .map_err(|e| format!("{} format: {e}", out.format))?;
2448            }
2449            let (files, warnings) = p
2450                .write_finish_with_warnings()
2451                .map_err(|e| Fail::from_plugin(&e, format!("{} format: {e}", out.format)))?;
2452            for w in warnings {
2453                self.ui.warn(&format!("  {} format: {w}", out.format));
2454            }
2455            for f in files {
2456                let f = PathBuf::from(f);
2457                let size = std::fs::metadata(&f).map(|m| m.len()).unwrap_or(0);
2458                let shown = rel(&self.project.root, &f);
2459                self.ui.step(
2460                    Level::Debug,
2461                    "Wrote",
2462                    &format!("{} ({})", shown.display(), human_bytes(size)),
2463                    Some(t.elapsed()),
2464                );
2465                self.outs[oi].files.push((f, None));
2466            }
2467        }
2468        let _ = p.close();
2469        Ok(())
2470    }
2471
2472    fn template_payload(&mut self, t: &crate::project::Template) -> Result<Json, Fail> {
2473        let file = crate::project::find_template(&self.project.root, &t.file)
2474            .ok_or_else(|| format!("template file `{}` doesn't exist", t.file))?;
2475        let renderer = self.renderer(None)?;
2476        let mut values = JsonMap::new();
2477        for b in &t.bindings {
2478            if let (Some(cell), Some(v)) = (&b.cell, &b.value) {
2479                let rendered = renderer
2480                    .render(&self.report.file, v)
2481                    .map_err(|e| format!("template value: {e}"))?;
2482                values.insert(format!("{}!{}", b.sheet, cell), Json::String(rendered));
2483            }
2484        }
2485        Ok(json!({"file": file.to_string_lossy(), "bindings": t.bindings, "values": values}))
2486    }
2487
2488    /// Deliver output `oi` to each of its destinations in order. A failure is recorded and the
2489    /// rest are still attempted; the output fails if any did.
2490    fn deliver(&mut self, oi: usize) -> Result<(), Fail> {
2491        let result = self.deliver_output(oi);
2492        let o = &mut self.outs[oi];
2493        o.status = Some(match &result {
2494            Err(_) => OutputStatus::Failed,
2495            Ok(()) if o.deliveries.iter().any(|d| d.status == "delivered") => OutputStatus::Delivered,
2496            Ok(()) => OutputStatus::Kept,
2497        });
2498        o.error = result.as_ref().err().map(|f| f.message.clone());
2499        result
2500    }
2501
2502    fn deliver_output(&mut self, oi: usize) -> Result<(), Fail> {
2503        if self.outs[oi].dests.is_empty() {
2504            self.outs[oi].delivery_note = Some("no destination declared; output stays in target/".into());
2505            let dir = rel(&self.project.root, &self.run_dir);
2506            self.ui.step(
2507                Level::Debug,
2508                "Kept",
2509                &format!("no destination declared: output stays in {}", dir.display()),
2510                None,
2511            );
2512            return Ok(());
2513        }
2514        if self.outs[oi].files.is_empty() {
2515            return Ok(());
2516        }
2517        let dests = std::mem::take(&mut self.outs[oi].dests);
2518        let mut failures = Vec::new();
2519        let mut nowhere = Vec::new();
2520        for d in &dests {
2521            let profiles = &self.project.profiles;
2522            let target = profiles.target_of(Role::Destination, &d.profile);
2523            // `None` for an entry that delivers nowhere.
2524            let kind = profiles
2525                .target(Role::Destination, &d.profile)
2526                .map(|o| o.kind.clone());
2527            self.attempts = 1;
2528            let (status, location, error) =
2529                match self.deliver_one(oi, d).map_err(|e| e.or(Code::DeliveryFailed)) {
2530                    Ok(Some(loc)) => ("delivered", Some(loc), None),
2531                    Ok(None) => {
2532                        nowhere.push(nowhere_note(&d.profile, &target));
2533                        ("not_delivered", None, None)
2534                    }
2535                    Err(e) => {
2536                        failures.push(e.clone());
2537                        ("failed", None, Some(e))
2538                    }
2539                };
2540            let record = Delivery {
2541                profile: d.profile.clone(),
2542                kind,
2543                target,
2544                status,
2545                location,
2546                attempts: Some(self.attempts).filter(|n| *n > 1),
2547                error_code: error.as_ref().map(|e| e.code.clone()),
2548                error: error.map(|e| e.message),
2549            };
2550            self.outs[oi].deliveries.push(record);
2551        }
2552        let o = &mut self.outs[oi];
2553        o.dests = dests;
2554        if o.files.iter().all(|(_, d)| d.is_none()) && !nowhere.is_empty() {
2555            o.delivery_note = Some(nowhere.join("; "));
2556        }
2557        match failures.len() {
2558            0 => Ok(()),
2559            1 if o.dests.len() == 1 => Err(failures.remove(0)),
2560            n => Err(Fail::new(
2561                failures[0].code.clone(),
2562                format!(
2563                    "{n} of {} destinations failed: {}",
2564                    o.dests.len(),
2565                    failures
2566                        .iter()
2567                        .map(|f| f.message.as_str())
2568                        .collect::<Vec<_>>()
2569                        .join("; ")
2570                ),
2571            )),
2572        }
2573    }
2574
2575    /// Deliver every file to one destination. `Ok(None)`: its entry for this run is
2576    /// `deliver: false`, so nothing was sent.
2577    fn deliver_one(&mut self, oi: usize, d: &RenderedDest) -> Result<Option<String>, Fail> {
2578        // An output is never delivered once the run is cancelled.
2579        self.not_cancelled()?;
2580        let profiles = &self.project.profiles;
2581        let out = match profiles.entry(Role::Destination, &d.profile) {
2582            Entry::Use(o) => o,
2583            Entry::Nowhere => {
2584                let note = nowhere_note(&d.profile, &profiles.target_of(Role::Destination, &d.profile));
2585                self.ui.step(Level::Info, "Kept", &note, None);
2586                return Ok(None);
2587            }
2588            // Both are checked before the run starts.
2589            Entry::Missing => {
2590                return Err(Fail::new(
2591                    Code::MissingTargetEntry,
2592                    profiles.missing_entry(Role::Destination, &d.profile),
2593                ));
2594            }
2595            Entry::Unknown => {
2596                return Err(format!("destination profile `{}` isn't in profiles.yml", d.profile).into());
2597            }
2598        };
2599        let kind = out.kind.clone();
2600        let connection = render_connection(out)?;
2601        let files = &self.outs[oi].files;
2602        let targets: Vec<DeliveryFile> = files
2603            .iter()
2604            .map(|(f, _)| {
2605                let name = f.file_name().unwrap().to_string_lossy().to_string();
2606                let r = d.path.as_deref().map(|r| {
2607                    if files.len() == 1 {
2608                        r.to_string()
2609                    } else {
2610                        match r.rfind(['/', '\\']) {
2611                            Some(i) => format!("{}{name}", &r[..=i]),
2612                            None => name.clone(),
2613                        }
2614                    }
2615                });
2616                DeliveryFile {
2617                    local_path: f.to_string_lossy().to_string(),
2618                    remote_path: r,
2619                }
2620            })
2621            .collect();
2622        // Bindings running at once never deliver to the same file.
2623        for t in &targets {
2624            if let Some(r) = &t.remote_path
2625                && let Some(other) = self.store.claim(&d.profile, r, &self.label())
2626            {
2627                return Err(Fail::new(
2628                    Code::DeliveryFailed,
2629                    format!(
2630                        "`{other}` delivers to `{r}` on destination `{}` in this run too; Bindings running at once (`threads:`) need different paths (put the Set or a var in the path)",
2631                        d.profile
2632                    ),
2633                ));
2634            }
2635        }
2636        let mut locations = Vec::new();
2637        if kind == LOCAL_TYPE {
2638            let errors = crate::options::local_option_errors(&d.options);
2639            if !errors.is_empty() {
2640                return Err(format!("destination `{}`: {}", d.profile, errors.join("; ")).into());
2641            }
2642            // `if_exists`, `atomic` (default true: written as `.<name>.dre-part`, then renamed)
2643            // and `temp_dir`.
2644            let (rules, _) = dre_protocol::delivery::Rules::from_settings(
2645                dre_protocol::delivery::Rules {
2646                    retries: 0,
2647                    ..Default::default()
2648                },
2649                &JsonMap::new(),
2650                &d.options,
2651                &[],
2652            )?;
2653            for (i, f) in targets.iter().enumerate() {
2654                let t = Instant::now();
2655                let r = f
2656                    .remote_path
2657                    .as_ref()
2658                    .ok_or("the local destination needs `output.destination.path`")?;
2659                let dst = self.project.root.join(r).to_string_lossy().to_string();
2660                let delivered = dre_protocol::delivery::deliver(
2661                    &mut dre_protocol::delivery::LocalStore,
2662                    Path::new(&f.local_path),
2663                    &dst,
2664                    &rules,
2665                )
2666                .map_err(|e| {
2667                    let message = format!("delivery to {dst} failed: {e}; the output is still in target/");
2668                    if e.exists {
2669                        // `if_exists: error`, coded as a plugin's would be.
2670                        let code = crate::codes::ErrorCode::Plugin {
2671                            code: "local/file-exists".into(),
2672                            kind: crate::codes::Kind::Delivery,
2673                        };
2674                        Fail::new(code, message)
2675                    } else {
2676                        Fail::new(Code::DeliveryFailed, message)
2677                    }
2678                })?;
2679                let loc = delivered.path;
2680                self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
2681                self.outs[oi].files[i].1.get_or_insert_with(|| loc.clone());
2682                locations.push(loc);
2683            }
2684            return Ok(Some(locations.join(", ")));
2685        }
2686        let failed = |e: dre_protocol::host::HostError| {
2687            Fail::from_plugin(
2688                &e,
2689                format!("delivery through `{kind}` failed: {e}; the output is still in target/"),
2690            )
2691        };
2692        let plugin = find_plugin(self.project, PluginKind::Destination, &kind)
2693            .map_err(|e| Fail::new(Code::DeliveryFailed, e.to_string()))?;
2694        let mut p = plugin
2695            .start(self.ui.plugin_log(), Some(&self.project.root))
2696            .map_err(|e| e.to_string())?;
2697        let is_message = self.b.outputs[oi].is_message();
2698        // A message goes to a destination that takes messages as a message; to any other, as
2699        // its `.md` file.
2700        if is_message
2701            && p.has(CAP_MESSAGE)
2702            && let Some((title, text)) = self.outs[oi].message.clone()
2703        {
2704            let t = Instant::now();
2705            let message = dre_protocol::msg::Message {
2706                title,
2707                text,
2708                html: None,
2709                path: targets[0].local_path.clone(),
2710            };
2711            let attach: Vec<DeliveryFile> = d
2712                .attach
2713                .iter()
2714                .filter_map(|name| {
2715                    self.b
2716                        .outputs
2717                        .iter()
2718                        .position(|o| o.name.as_deref() == Some(name))
2719                })
2720                .flat_map(|j| self.outs[j].files.iter())
2721                .map(|(f, _)| DeliveryFile {
2722                    local_path: f.to_string_lossy().to_string(),
2723                    remote_path: None,
2724                })
2725                .collect();
2726            let loc = p
2727                .deliver_message(&message, &attach, connection, d.options.clone())
2728                .map_err(failed)?;
2729            self.attempts = p.last_attempts();
2730            self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
2731            self.outs[oi].files[0].1.get_or_insert_with(|| loc.clone());
2732            let _ = p.close();
2733            return Ok(Some(loc));
2734        }
2735        if !is_message && p.has(CAP_MESSAGE_ONLY) {
2736            let _ = p.close();
2737            return Err(format!(
2738                "`{kind}` only takes messages, but this output is `{}`; deliver the file elsewhere and link it from a message",
2739                self.b.outputs[oi].format
2740            ).into());
2741        }
2742        let batches: Vec<Vec<usize>> = if targets.len() > 1 && p.has(CAP_MULTI_FILE) {
2743            vec![(0..targets.len()).collect()]
2744        } else {
2745            (0..targets.len()).map(|i| vec![i]).collect()
2746        };
2747        for batch in batches {
2748            let t = Instant::now();
2749            let files: Vec<DeliveryFile> = batch.iter().map(|&i| targets[i].clone()).collect();
2750            let loc = p
2751                .deliver_files(&files, connection.clone(), d.options.clone())
2752                .map_err(failed)?;
2753            self.attempts = self.attempts.max(p.last_attempts());
2754            self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
2755            for i in batch {
2756                self.outs[oi].files[i].1.get_or_insert_with(|| loc.clone());
2757            }
2758            locations.push(loc);
2759        }
2760        let _ = p.close();
2761        Ok(Some(locations.join(", ")))
2762    }
2763
2764    fn snapshot(&self) -> Json {
2765        let sets: Vec<Json> = self
2766            .produced
2767            .iter()
2768            .map(|p| {
2769                json!({
2770                    "name": p.name,
2771                    "columns": p.schema.fields().iter().map(|f| json!({
2772                        "name": f.name(),
2773                        "type": f.data_type().to_string(),
2774                        "type_shape": crate::schema::type_shape(f.data_type()),
2775                    })).collect::<Vec<_>>(),
2776                })
2777            })
2778            .collect();
2779        json!({ "result_sets": sets })
2780    }
2781
2782    fn schema_drift(&mut self) -> Vec<String> {
2783        let path = self.schema_dir.join("last_success.json");
2784        let mut current = self.snapshot();
2785        let Ok(prev) = std::fs::read_to_string(&path) else {
2786            crate::schema::redact(&mut current);
2787            self.protected_snapshot = Some(current);
2788            return Vec::new();
2789        };
2790        let Ok(mut prev) = serde_json::from_str::<Json>(&prev) else {
2791            crate::schema::redact(&mut current);
2792            self.protected_snapshot = Some(current);
2793            return Vec::new();
2794        };
2795        let original = prev.clone();
2796        crate::schema::redact(&mut prev);
2797        crate::schema::redact(&mut current);
2798        if crate::secrets::enabled() {
2799            crate::schema::align_protection(&mut prev, &mut current);
2800        }
2801        if prev != original {
2802            let protected =
2803                serde_json::to_string_pretty(&prev).expect("serializing a serde_json::Value cannot fail");
2804            if let Err(error) = crate::fs::write_atomic(&path, protected.as_bytes()) {
2805                self.ui.warn(&format!(
2806                    "can't protect the existing schema snapshot at {}: {error}; continuing without rewriting it",
2807                    path.display()
2808                ));
2809            }
2810        }
2811        let drift = crate::schema::drift(&prev, &current);
2812        self.protected_snapshot = Some(current);
2813        drift
2814    }
2815
2816    fn write_snapshot(&self) -> std::io::Result<()> {
2817        std::fs::create_dir_all(&self.schema_dir)?;
2818        let snapshot = self.protected_snapshot.clone().unwrap_or_else(|| {
2819            let mut snapshot = self.snapshot();
2820            crate::schema::redact(&mut snapshot);
2821            snapshot
2822        });
2823        crate::fs::write_atomic(
2824            &self.schema_dir.join("last_success.json"),
2825            serde_json::to_string_pretty(&snapshot)?.as_bytes(),
2826        )
2827    }
2828
2829    fn write_results(&self, status: &Status, error: Option<&Fail>) -> std::io::Result<()> {
2830        std::fs::create_dir_all(&self.run_dir)?;
2831        // `outputs` lists every file (as before 0.3); `output_results` has one entry per output.
2832        let mut outputs: Vec<run_results::OutputFile> = Vec::new();
2833        let mut output_results: Vec<run_results::OutputResult> = Vec::new();
2834        for (o, run) in self.b.outputs.iter().zip(&self.outs) {
2835            let files: Vec<run_results::OutputFileRef> = run
2836                .files
2837                .iter()
2838                .map(|(f, delivered)| run_results::OutputFileRef {
2839                    path: record_path(self.project, f),
2840                    size: std::fs::metadata(f).map(|m| m.len()).unwrap_or(0),
2841                    delivered_to: delivered.clone(),
2842                })
2843                .collect();
2844            outputs.extend(files.iter().map(|f| run_results::OutputFile {
2845                path: f.path.clone(),
2846                size: f.size,
2847                delivered_to: f.delivered_to.clone(),
2848                output: o.name.clone(),
2849            }));
2850            let queries: Vec<String> = self
2851                .b
2852                .queries
2853                .iter()
2854                .filter(|q| q.tab && o.feeds(&q.query))
2855                .map(|q| q.query.clone())
2856                .collect();
2857            let status = match (run.status, status) {
2858                (Some(s), _) => s,
2859                (None, Status::Error) => OutputStatus::Failed,
2860                (None, _) => OutputStatus::Kept,
2861            }
2862            .as_str();
2863            output_results.push(run_results::OutputResult {
2864                name: o.name.clone(),
2865                format: o.format.clone(),
2866                queries,
2867                status,
2868                error: run.error.clone(),
2869                files,
2870                delivery: run.delivery_note.clone(),
2871                deliveries: run.deliveries.clone(),
2872                when: run.when,
2873                message: run.message.clone().map(|(title, text)| run_results::Message {
2874                    title: crate::secrets::mask_message_title(&title).into_owned(),
2875                    text: crate::secrets::mask_markdown(&text).into_owned(),
2876                }),
2877            });
2878        }
2879        // In delivery order: file outputs, then messages.
2880        let order = (0..self.outs.len())
2881            .filter(|&i| !self.b.outputs[i].is_message())
2882            .chain((0..self.outs.len()).filter(|&i| self.b.outputs[i].is_message()));
2883        let deliveries: Vec<Delivery> = order
2884            .flat_map(|i| self.outs[i].deliveries.iter().cloned())
2885            .collect();
2886        let notes: Vec<&str> = self
2887            .outs
2888            .iter()
2889            .filter_map(|o| o.delivery_note.as_deref())
2890            .collect();
2891        let delivery = (!notes.is_empty()).then(|| notes.join("; "));
2892        let results = RunResults {
2893            schema_version: run_results::SCHEMA_VERSION,
2894            run_id: self.run_id.clone(),
2895            report: self.report.name.clone(),
2896            set: self.b.set.clone(),
2897            binding: self.b.dir_name().to_string(),
2898            managed: self.report.managed,
2899            profile: self.parsed.inherited.clone(),
2900            connections: self
2901                .parsed
2902                .connections()
2903                .into_iter()
2904                .map(str::to_string)
2905                .collect(),
2906            target: self.target.clone(),
2907            schedule: self.opts.schedule.clone(),
2908            schedule_vars: self.schedule_vars.clone(),
2909            vars: self.rendered_vars.clone(),
2910            run_date: self.date.to_string(),
2911            scheduled_at: self.opts.scheduled_at.map(rfc3339),
2912            timezone: self.calendar.tz.name().to_string(),
2913            params: self.opts.params(self.date),
2914            status: *status,
2915            error: error.map(|e| e.message.clone()),
2916            error_code: error.map(|e| e.code.clone()),
2917            error_kind: error.map(|e| e.code.kind()),
2918            preview: self.opts.preview.is_some(),
2919            row_limit: self.opts.preview,
2920            started_at: self.started_at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
2921            duration_ms: self.started.elapsed().as_millis() as u64,
2922            result_sets: self
2923                .produced
2924                .iter()
2925                .map(|p| run_results::ResultSet {
2926                    name: p.name.clone(),
2927                    query: p.query.clone(),
2928                    connection: p.connection.clone(),
2929                    rows: p.rows,
2930                    columns: p.schema.fields().iter().map(|f| f.name().clone()).collect(),
2931                })
2932                .collect(),
2933            outputs,
2934            output_results,
2935            delivery,
2936            deliveries,
2937            schema_drift: self.drift.clone(),
2938            target_path: self.project.target_dir.clone(),
2939            settings: self.project.settings.clone(),
2940            manifest_checksum: self.opts.manifest_checksum.clone(),
2941        };
2942        self.store.write_results(&self.run_dir, &results)
2943    }
2944
2945    /// `dre validate --live`: execute temp creates, `check` everything else, each on its
2946    /// connection; then check declared source columns against the database.
2947    fn live_check(&mut self, statements: &[Statement]) -> Result<(), Fail> {
2948        let pool = self.pool.clone().expect("pool");
2949        let mut failures = Vec::new();
2950        let mut unexecuted_setup: BTreeMap<String, (PathBuf, usize)> = BTreeMap::new();
2951        let mut uncheckable: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
2952        let sql_log = self.ui.sql_log();
2953        for st in statements {
2954            if uncheckable.contains(&st.connection) {
2955                continue;
2956            }
2957            let session = pool.session(&st.connection)?;
2958            let mut s = session.lock().unwrap();
2959            if !s.get()?.has(dre_protocol::CAP_CHECK) {
2960                let kind = pool.kind(&st.connection).unwrap_or_default();
2961                self.ui.warn(&format!(
2962                    "  not checkable: the `{kind}` source plugin (connection `{}`) can't check statements without running them",
2963                    st.connection
2964                ));
2965                uncheckable.insert(st.connection.clone());
2966                continue;
2967            }
2968            let verb = if st.kind == StatementKind::TempCreate {
2969                ""
2970            } else {
2971                " (check)"
2972            };
2973            sql_log(
2974                &format!("{}:{}{verb}", st.file.display(), st.line),
2975                &protected_sql(&st.text, st.sensitive),
2976            );
2977            if st.kind == StatementKind::TempCreate {
2978                let result = {
2979                    let (process, _log_scope) = s.statement(st.sensitive)?;
2980                    process.execute(&st.text, None, |_, _| Ok(()))
2981                };
2982                if let Err(e) = result {
2983                    failures.push(format!(
2984                        "{}:{}: {}",
2985                        st.file.display(),
2986                        st.line,
2987                        masked_source_error(&e, st.sensitive)
2988                    ));
2989                }
2990                continue;
2991            }
2992            let result = {
2993                let (process, _log_scope) = s.statement(st.sensitive)?;
2994                process.check(&st.text)
2995            };
2996            if let Err(e) = result {
2997                let mut msg = format!(
2998                    "{}:{}: {}",
2999                    st.file.display(),
3000                    st.line,
3001                    masked_source_error(&e, st.sensitive)
3002                );
3003                if let Some((f, l)) = unexecuted_setup.get(&st.connection) {
3004                    msg.push_str(&format!(
3005                        " (may be a false positive: the setup statement at {}:{l} wasn't executed during the check)",
3006                        f.display()
3007                    ));
3008                }
3009                failures.push(msg);
3010            }
3011            if st.kind == StatementKind::Other {
3012                unexecuted_setup.insert(st.connection.clone(), (st.file.clone(), st.line));
3013            }
3014        }
3015        self.check_source_columns(&pool, &mut failures);
3016        if failures.is_empty() {
3017            Ok(())
3018        } else {
3019            Err(Fail::new(Code::QueryFailed, failures.join("\n    ")))
3020        }
3021    }
3022
3023    /// Declared source `columns` against the real table, once per source table, connection and
3024    /// target in this process: each must exist, and match its `data_type` (loosely) where given.
3025    fn check_source_columns(&mut self, pool: &Arc<Pool>, failures: &mut Vec<String>) {
3026        /// `(project, source.table, connection, target)`.
3027        type Checked = std::collections::BTreeSet<(PathBuf, String, String, String)>;
3028        static CHECKED: std::sync::LazyLock<Mutex<Checked>> = std::sync::LazyLock::new(Mutex::default);
3029        for q in self.parsed.queries.clone() {
3030            let Some(conn) = q.connection.clone() else {
3031                continue;
3032            };
3033            for key in &q.sources {
3034                let Some(r) = self.parsed.sources.get(key).cloned() else {
3035                    continue;
3036                };
3037                let Some(declared) = self
3038                    .project
3039                    .sources
3040                    .get(&r.source)
3041                    .and_then(|s| s.table(&r.table))
3042                    .map(|t| t.columns.clone())
3043                    .filter(|c| !c.is_empty())
3044                else {
3045                    continue;
3046                };
3047                let id = (
3048                    self.project.root.clone(),
3049                    key.clone(),
3050                    conn.clone(),
3051                    self.project.profiles.target_of(Role::Connection, &conn),
3052                );
3053                if !CHECKED.lock().unwrap().insert(id) {
3054                    continue;
3055                }
3056                let quote = if r.quoting.any() {
3057                    let kind = pool.kind(&conn).unwrap_or_default();
3058                    match pool.connections.identifier_quote(&kind) {
3059                        Ok(q) => q,
3060                        Err(e) => {
3061                            failures.push(format!("source `{key}`: can't quote its name: {e}"));
3062                            continue;
3063                        }
3064                    }
3065                } else {
3066                    None
3067                };
3068                let rel = r.relation(quote.as_deref());
3069                let runner = PoolRunner {
3070                    pool: pool.clone(),
3071                    default: Some(conn.clone()),
3072                    log: self.ui.sql_log(),
3073                };
3074                let actual = match runner.columns(&format!("select * from {rel} where 1=0"), None) {
3075                    Ok(c) => c,
3076                    Err(e) => {
3077                        failures.push(format!("source `{key}` ({rel} on connection `{conn}`): {e}"));
3078                        continue;
3079                    }
3080                };
3081                for c in &declared {
3082                    let Some(a) = actual.iter().find(|a| a.name.eq_ignore_ascii_case(&c.name)) else {
3083                        failures.push(format!(
3084                            "source `{key}`: declared column `{}` isn't in {rel} on connection `{conn}` (it has: {})",
3085                            c.name,
3086                            actual.iter().map(|a| a.name.as_str()).collect::<Vec<_>>().join(", ")
3087                        ));
3088                        continue;
3089                    };
3090                    let Some(dt) = &c.data_type else { continue };
3091                    match crate::coltypes::matches(dt, &a.data_type) {
3092                        Some(true) => {}
3093                        Some(false) => failures.push(format!(
3094                            "source `{key}`: column `{}` is declared `{dt}`, but {rel} on connection `{conn}` returns {}",
3095                            c.name, a.data_type
3096                        )),
3097                        None => self.ui.warn(&format!(
3098                            "  source `{key}`: column `{}`'s `data_type: {dt}` isn't a type DRE can compare, so only its presence was checked",
3099                            c.name
3100                        )),
3101                    }
3102                }
3103            }
3104        }
3105    }
3106}
3107
3108/// The source session, started on first use (rendering may need it for `run_query()`).
3109struct Session {
3110    path: Result<crate::plugins::Located, String>,
3111    connection: Result<JsonMap<String, Json>, String>,
3112    cwd: PathBuf,
3113    log: LogSink,
3114    /// Whether plugin diagnostics for the active statement must be hidden rather than exactly
3115    /// masked. Connection diagnostics always use exact-value masking.
3116    opaque_log: Arc<AtomicBool>,
3117    /// Unmanaged reports run read-only unless they create temp objects.
3118    unmanaged: bool,
3119    read_only: bool,
3120    /// A lookup was loaded into a temp table, so the session must stay open as it is.
3121    loaded: bool,
3122    proc_: Option<PluginProcess>,
3123}
3124
3125impl Session {
3126    fn new(
3127        path: Result<crate::plugins::Located, String>,
3128        connection: Result<JsonMap<String, Json>, String>,
3129        cwd: PathBuf,
3130        log: LogSink,
3131        unmanaged: bool,
3132    ) -> Session {
3133        let opaque_log = Arc::new(AtomicBool::new(false));
3134        let protect = Arc::clone(&opaque_log);
3135        let downstream = log;
3136        let log: LogSink = Arc::new(move |plugin, line| {
3137            let line = crate::secrets::mask_source_error(line, protect.load(Ordering::SeqCst));
3138            downstream(plugin, &line);
3139        });
3140        Session {
3141            path,
3142            connection,
3143            cwd,
3144            log,
3145            opaque_log,
3146            unmanaged,
3147            read_only: unmanaged,
3148            loaded: false,
3149            proc_: None,
3150        }
3151    }
3152
3153    fn want_read_only(&mut self, ro: bool) {
3154        if self.read_only != ro {
3155            // Reopen if a session was already started in the other mode.
3156            self.close();
3157            self.read_only = ro;
3158        }
3159    }
3160
3161    fn get(&mut self) -> Result<&mut PluginProcess, String> {
3162        if self.proc_.is_none() {
3163            let plugin = self.path.clone()?;
3164            let connection = self.connection.clone()?;
3165            let mut p = plugin
3166                .start(self.log.clone(), Some(&self.cwd))
3167                .map_err(|e| masked_source_error(&e, false))?;
3168            let ro = self.unmanaged && self.read_only && p.has(CAP_READ_ONLY);
3169            p.open(connection, ro).map_err(|e| {
3170                let protected = masked_source_error(&e, false);
3171                format!("can't open the source connection: {protected}")
3172            })?;
3173            self.proc_ = Some(p);
3174        }
3175        Ok(self.proc_.as_mut().unwrap())
3176    }
3177
3178    fn statement(&mut self, sensitive: bool) -> Result<(&mut PluginProcess, OpaqueLogScope), String> {
3179        let opaque_log = Arc::clone(&self.opaque_log);
3180        let process = self.get()?;
3181        let scope = OpaqueLogScope::new(opaque_log, sensitive);
3182        Ok((process, scope))
3183    }
3184
3185    fn close(&mut self) {
3186        if let Some(p) = self.proc_.take() {
3187            let _ = p.close();
3188        }
3189    }
3190}
3191
3192struct OpaqueLogScope(Arc<AtomicBool>);
3193
3194impl OpaqueLogScope {
3195    fn new(active: Arc<AtomicBool>, sensitive: bool) -> Self {
3196        active.store(sensitive, Ordering::SeqCst);
3197        Self(active)
3198    }
3199}
3200
3201impl Drop for OpaqueLogScope {
3202    fn drop(&mut self) {
3203        self.0.store(false, Ordering::SeqCst);
3204    }
3205}
3206
3207/// What a run logs, and records, for a destination whose entry is `deliver: false`.
3208fn nowhere_note(profile: &str, target: &str) -> String {
3209    format!(
3210        "destination `{profile}`: `{target}` delivers nowhere (`deliver: false`); output stays in target/"
3211    )
3212}
3213
3214/// The Binding's connections: each one's session, opened when a query first needs it and held
3215/// until the report ends.
3216struct Pool {
3217    connections: Arc<ProfileConnections>,
3218    cwd: PathBuf,
3219    log: LogSink,
3220    unmanaged: bool,
3221    sessions: Mutex<BTreeMap<String, Arc<Mutex<Session>>>>,
3222}
3223
3224impl Pool {
3225    fn new(connections: Arc<ProfileConnections>, cwd: PathBuf, log: LogSink, unmanaged: bool) -> Pool {
3226        Pool {
3227            connections,
3228            cwd,
3229            log,
3230            unmanaged,
3231            sessions: Mutex::default(),
3232        }
3233    }
3234
3235    /// A connection's settings for this run.
3236    fn target(&self, name: &str) -> Result<ProfileTarget, String> {
3237        let profiles = &self.connections.profiles;
3238        match profiles.entry(Role::Connection, name) {
3239            Entry::Use(o) => Ok(o.clone()),
3240            Entry::Unknown => Err(format!(
3241                "connection `{name}` isn't under `{}:` in {}",
3242                profiles.section_key(Role::Connection),
3243                profiles.path.display()
3244            )),
3245            // A connection never has `deliver: false`.
3246            Entry::Missing | Entry::Nowhere => Err(profiles.missing_entry(Role::Connection, name)),
3247        }
3248    }
3249
3250    /// A connection's plugin type.
3251    fn kind(&self, name: &str) -> Result<String, String> {
3252        Ok(self.target(name)?.kind)
3253    }
3254
3255    fn session(&self, name: &str) -> Result<Arc<Mutex<Session>>, String> {
3256        if let Some(s) = self.sessions.lock().unwrap().get(name) {
3257            return Ok(s.clone());
3258        }
3259        let output = self.target(name)?;
3260        let id = crate::project::PluginId::new(PluginKind::Source, output.kind.clone());
3261        let path = crate::plugins::locate_in(&self.connections.root, &self.connections.plugins, &id)
3262            .map_err(String::from);
3263        let session = Arc::new(Mutex::new(Session::new(
3264            path,
3265            render_connection(&output),
3266            self.cwd.clone(),
3267            self.log.clone(),
3268            self.unmanaged,
3269        )));
3270        self.sessions
3271            .lock()
3272            .unwrap()
3273            .insert(name.to_string(), session.clone());
3274        Ok(session)
3275    }
3276
3277    fn close_all(&self) {
3278        for s in self.sessions.lock().unwrap().values() {
3279            if let Ok(mut s) = s.lock() {
3280                s.close();
3281            }
3282        }
3283    }
3284}
3285
3286/// `run_query()`, `columns()` and lookups for one renderer: on the named connection's session,
3287/// else on the renderer's own (`default`).
3288struct PoolRunner {
3289    pool: Arc<Pool>,
3290    default: Option<String>,
3291    log: LogSink,
3292}
3293
3294impl PoolRunner {
3295    fn session(&self, profile: Option<&str>) -> Result<Arc<Mutex<Session>>, String> {
3296        let name = profile.or(self.default.as_deref()).ok_or(
3297            "no connection here: pass `profile=`, or give the report a `profile:` (this isn't a query's SQL)",
3298        )?;
3299        self.pool.session(name)
3300    }
3301}
3302
3303impl QueryRunner for PoolRunner {
3304    fn run_query(&self, sql: &str, max_rows: u64, profile: Option<&str>) -> Result<QueryRows, String> {
3305        let session = self.session(profile)?;
3306        let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
3307        let sensitive = crate::secrets::contains_secret(sql);
3308        // An unmanaged report may only read, and run_query() runs before the file's own
3309        // statements are checked, so it gets the same rule up front.
3310        if s.unmanaged
3311            && let Some(bad) = sqlsplit::split(sql)
3312                .into_iter()
3313                .find(|st| sqlsplit::classify(&st.text) != StatementKind::Read)
3314        {
3315            return Err(format!(
3316                "run_query() in an unmanaged report may only read, but got `{}`; give the report a YAML to declare it",
3317                masked_sql_summary(&bad.text, 60, sensitive)
3318            ));
3319        }
3320        (self.log)("run_query()", &protected_sql(sql, sensitive));
3321        let (p, _log_scope) = s.statement(sensitive)?;
3322        let mut out = QueryRows::default();
3323        let mut too_many = false;
3324        let exec = p
3325            .execute(sql, Some(max_rows + 1), |schema, batch| {
3326                if out.columns.is_empty() {
3327                    out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
3328                }
3329                if out.rows.len() as u64 + batch.num_rows() as u64 > max_rows {
3330                    too_many = true;
3331                    return Ok(());
3332                }
3333                out.rows.extend(crate::values::batch_rows(&batch));
3334                Ok(())
3335            })
3336            .map_err(|e| format!("run_query() failed: {}", masked_source_error(&e, sensitive)))?;
3337        if too_many {
3338            return Err(format!(
3339                "run_query() returned more than {max_rows} rows; it's meant for small lookups — raise the cap with `run_query(sql, max_rows=N)` or `run_query_max_rows` in dre_project.yml"
3340            ));
3341        }
3342        if let Execution::Result { schema, .. } = exec
3343            && out.columns.is_empty()
3344        {
3345            out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
3346        }
3347        Ok(out)
3348    }
3349
3350    fn columns(&self, sql: &str, profile: Option<&str>) -> Result<Vec<Column>, String> {
3351        let session = self.session(profile)?;
3352        let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
3353        let sensitive = crate::secrets::contains_secret(sql);
3354        (self.log)("columns()", &protected_sql(sql, sensitive));
3355        let (p, _log_scope) = s.statement(sensitive)?;
3356        let mut schema: Option<SchemaRef> = None;
3357        let exec = p
3358            .execute(sql, Some(1), |sch, _| {
3359                schema.get_or_insert_with(|| sch.clone());
3360                Ok(())
3361            })
3362            .map_err(|e| masked_source_error(&e, sensitive))?;
3363        if let Execution::Result { schema: sch, .. } = exec {
3364            schema.get_or_insert(sch);
3365        }
3366        let schema = schema.ok_or("the query returned no result set")?;
3367        Ok(schema
3368            .fields()
3369            .iter()
3370            .map(|f| Column {
3371                name: f.name().clone(),
3372                data_type: f.data_type().to_string(),
3373            })
3374            .collect())
3375    }
3376
3377    fn load(&self, name: &str, table: &Table) -> Result<Option<(String, Option<String>)>, String> {
3378        let session = self.session(None)?;
3379        let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
3380        // A temp table is allowed even in an unmanaged report; it needs a writable session.
3381        s.want_read_only(false);
3382        if !s.get()?.has(CAP_LOAD) {
3383            return Ok(None);
3384        }
3385        let batch = table.to_batch()?;
3386        let schema = batch.schema();
3387        (self.log)(
3388            &format!("lookup `{name}`"),
3389            &format!(
3390                "-- {} rows loaded through the plugin's `load` request",
3391                batch.num_rows()
3392            ),
3393        );
3394        let r = s.get()?.load(name, &schema, [batch]).map_err(|e| e.to_string())?;
3395        s.loaded = true;
3396        Ok(Some(r))
3397    }
3398}
3399
3400/// Render every string inside a destination option value.
3401fn render_json(renderer: &Renderer, file: &Path, v: &Json) -> Result<Json, RenderError> {
3402    Ok(match v {
3403        Json::String(s) => Json::String(renderer.render(file, s)?),
3404        Json::Array(a) => Json::Array(
3405            a.iter()
3406                .map(|v| render_json(renderer, file, v))
3407                .collect::<Result<_, _>>()?,
3408        ),
3409        Json::Object(o) => Json::Object(
3410            o.iter()
3411                .map(|(k, v)| Ok((k.clone(), render_json(renderer, file, v)?)))
3412                .collect::<Result<_, RenderError>>()?,
3413        ),
3414        other => other.clone(),
3415    })
3416}
3417
3418/// Render `env_var()` (and only that) inside a profile output's string fields.
3419pub(crate) fn render_connection(output: &ProfileTarget) -> Result<JsonMap<String, Json>, String> {
3420    let mut env = minijinja::Environment::new();
3421    env.set_undefined_behavior(minijinja::UndefinedBehavior::Strict);
3422    env.add_function("env_var", |name: String, default: Option<String>| -> Result<String, minijinja::Error> {
3423        std::env::var(&name).ok().or(default).ok_or_else(|| {
3424            minijinja::Error::new(
3425                minijinja::ErrorKind::UndefinedError,
3426                format!("`env_var('{name}')`: environment variable `{name}` is not set and no default is given"),
3427            )
3428        })
3429    });
3430    fn walk(env: &minijinja::Environment<'_>, v: &Json) -> Result<Json, String> {
3431        Ok(match v {
3432            Json::String(s) if crate::preflight::is_templated(s) => Json::String(
3433                env.render_str(s, ())
3434                    .map_err(|e| format!("profiles.yml: {}", e.detail().unwrap_or("render error")))?,
3435            ),
3436            Json::Array(a) => Json::Array(a.iter().map(|x| walk(env, x)).collect::<Result<_, _>>()?),
3437            Json::Object(o) => Json::Object(
3438                o.iter()
3439                    .map(|(k, x)| Ok((k.clone(), walk(env, x)?)))
3440                    .collect::<Result<_, String>>()?,
3441            ),
3442            other => other.clone(),
3443        })
3444    }
3445    match walk(&env, &Json::Object(output.fields.clone()))? {
3446        Json::Object(o) => Ok(o),
3447        _ => unreachable!(),
3448    }
3449}
3450
3451/// Locate the plugin for a declared type among the project's packages, honouring `dre.lock`
3452/// pins and declared constraints.
3453pub fn find_plugin(
3454    project: &Project,
3455    kind: PluginKind,
3456    name: &str,
3457) -> Result<crate::plugins::Located, crate::plugins::LocateError> {
3458    crate::plugins::locate(project, &crate::project::PluginId::new(kind, name))
3459}
3460
3461/// `connection`, `destination` and `profile()` from `profiles.yml`, for the run's target. A
3462/// field is secret when the plugin's `describe` says so; when the plugin can't be asked, when its
3463/// name looks like one; and always when its value comes from a `DRE_SECRET_*` variable.
3464struct ProfileConnections {
3465    /// Each profile's entry for the run comes from here.
3466    profiles: Profiles,
3467    root: PathBuf,
3468    plugins: Vec<crate::project::PluginRequirement>,
3469    log: LogSink,
3470}
3471
3472/// What `describe` said, by `(project, role, type)`, `None` when it couldn't be asked: each
3473/// plugin is asked once per process, not once per Binding.
3474static DESCRIBED: std::sync::LazyLock<Mutex<BTreeMap<DescribeKey, Option<Described>>>> =
3475    std::sync::LazyLock::new(Mutex::default);
3476
3477/// `(project root, role, plugin type)`.
3478type DescribeKey = (PathBuf, Role, String);
3479
3480#[derive(Clone)]
3481struct Described {
3482    secrets: Vec<String>,
3483    identifier_quote: Option<String>,
3484    capabilities: Vec<String>,
3485    /// A message destination's length limit.
3486    message_limit: Option<u64>,
3487}
3488
3489const SECRET_NAME_WORDS: &[&str] = &["password", "secret", "token", "key", "credential"];
3490
3491impl ProfileConnections {
3492    fn view(&self, role: Role, name: &str) -> Result<Connection, String> {
3493        let target = self.profiles.target_of(role, name);
3494        let out: ProfileTarget = match self.profiles.entry(role, name) {
3495            Entry::Use(o) => o.clone(),
3496            Entry::Nowhere => {
3497                return Err(format!(
3498                    "destination `{name}`: `{target}` delivers nowhere (`deliver: false`), so it has no settings"
3499                ));
3500            }
3501            Entry::Missing => return Err(self.profiles.missing_entry(role, name)),
3502            Entry::Unknown => {
3503                return Err(format!(
3504                    "no {} profile `{name}` in {}",
3505                    role.as_str(),
3506                    self.profiles.path.display()
3507                ));
3508            }
3509        };
3510        let mut secrets: Vec<String> = out
3511            .fields
3512            .iter()
3513            .filter(|(_, v)| v.as_str().is_some_and(|s| s.contains(crate::secrets::PREFIX)))
3514            .map(|(k, _)| k.clone())
3515            .collect();
3516        match self.described(role, &out.kind) {
3517            Some(d) => secrets.extend(d.secrets),
3518            None => secrets.extend(
3519                out.fields
3520                    .keys()
3521                    .filter(|k| {
3522                        let k = k.to_lowercase();
3523                        SECRET_NAME_WORDS.iter().any(|w| k.contains(w))
3524                    })
3525                    .cloned(),
3526            ),
3527        }
3528        crate::secrets::register_secret_fields(&out.fields, &secrets);
3529        let fields = render_connection(&out)?;
3530        Ok(Connection {
3531            profile: name.to_string(),
3532            target,
3533            kind: out.kind.clone(),
3534            fields,
3535            secrets,
3536        })
3537    }
3538
3539    /// The plugin's `describe` reply, asked once per plugin type.
3540    fn described(&self, role: Role, kind: &str) -> Option<Described> {
3541        let key = (self.root.clone(), role, kind.to_string());
3542        if let Some(v) = DESCRIBED.lock().unwrap().get(&key) {
3543            return v.clone();
3544        }
3545        let plugin_kind = match role {
3546            Role::Connection => PluginKind::Source,
3547            Role::Destination => PluginKind::Destination,
3548        };
3549        let asked = (|| {
3550            if kind == LOCAL_TYPE {
3551                return Some(Described {
3552                    secrets: Vec::new(),
3553                    identifier_quote: None,
3554                    capabilities: Vec::new(),
3555                    message_limit: None,
3556                });
3557            }
3558            let id = crate::project::PluginId::new(plugin_kind, kind);
3559            let plugin = crate::plugins::locate_in(&self.root, &self.plugins, &id).ok()?;
3560            let mut p = plugin.start(self.log.clone(), Some(&self.root)).ok()?;
3561            let capabilities = p.info().capabilities.clone();
3562            let d = p.description().ok();
3563            let _ = p.close();
3564            let d = d?;
3565            Some(Described {
3566                secrets: d
3567                    .connection_fields
3568                    .into_iter()
3569                    .filter(|f| f.secret)
3570                    .map(|f| f.name)
3571                    .collect(),
3572                identifier_quote: d.identifier_quote,
3573                capabilities,
3574                message_limit: d.message_limit,
3575            })
3576        })();
3577        DESCRIBED.lock().unwrap().insert(key, asked.clone());
3578        asked
3579    }
3580}
3581
3582impl Connections for ProfileConnections {
3583    fn profile(&self, name: &str, role: Option<&str>) -> Result<Connection, String> {
3584        let role = match role {
3585            Some("connection") => Role::Connection,
3586            Some(_) => Role::Destination,
3587            None => {
3588                let c = self.profiles.get(Role::Connection, name).is_some();
3589                let d = self.profiles.get(Role::Destination, name).is_some();
3590                match (c, d) {
3591                    (true, true) => {
3592                        return Err(format!(
3593                            "`{name}` is both a connection and a destination profile; say which with `profile('{name}', role='connection')` or `role='destination'`"
3594                        ));
3595                    }
3596                    (true, false) => Role::Connection,
3597                    _ => Role::Destination,
3598                }
3599            }
3600        };
3601        self.view(role, name)
3602    }
3603
3604    fn identifier_quote(&self, kind: &str) -> Result<Option<String>, String> {
3605        match self.described(Role::Connection, kind) {
3606            Some(d) => Ok(d.identifier_quote),
3607            None => Err(format!(
3608                "the `{kind}` source plugin couldn't be asked for its identifier quote character"
3609            )),
3610        }
3611    }
3612}
3613
3614fn check_sheet_name(n: &str) -> Result<(), String> {
3615    if n.is_empty() || n.chars().count() > 31 {
3616        return Err(format!("sheet name `{n}` must be 1 to 31 characters long"));
3617    }
3618    if let Some(c) = n.chars().find(|c| "[]:*?/\\".contains(*c)) {
3619        return Err(format!(
3620            "sheet name `{n}` contains `{c}`, which Excel doesn't allow in sheet names"
3621        ));
3622    }
3623    if n.starts_with('\'') || n.ends_with('\'') {
3624        return Err(format!("sheet name `{n}` can't start or end with an apostrophe"));
3625    }
3626    if n.eq_ignore_ascii_case("history") {
3627        return Err("`History` is reserved by Excel and can't be a sheet name".into());
3628    }
3629    Ok(())
3630}
3631
3632fn is_single_table(format: &str) -> bool {
3633    matches!(format, "csv" | "delimited" | "fixed_width" | "parquet")
3634}
3635
3636fn extension(format: &str) -> &str {
3637    match format {
3638        "delimited" | "fixed_width" => "txt",
3639        crate::project::MESSAGE_FORMAT => "md",
3640        f => f,
3641    }
3642}
3643
3644fn split_ext(name: &str) -> (&str, &str) {
3645    match name.rfind('.') {
3646        Some(i) if i > 0 => (&name[..i], &name[i..]),
3647        _ => (name, ""),
3648    }
3649}
3650
3651fn rel(root: &Path, p: &Path) -> PathBuf {
3652    crate::slash(p.strip_prefix(root).unwrap_or(p))
3653}
3654
3655/// An instant as DRE records it: RFC 3339 in UTC, to the second.
3656pub fn rfc3339(t: chrono::DateTime<chrono::Utc>) -> String {
3657    t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
3658}
3659
3660/// How a generated file is recorded in `run_results.json`: relative to the project root when
3661/// it's inside it, else relative to the target path, so moving the target folder doesn't
3662/// change what's recorded.
3663fn record_path(project: &Project, p: &Path) -> PathBuf {
3664    match p.strip_prefix(&project.root) {
3665        Ok(r) => crate::slash(r),
3666        Err(_) => rel(&project.target_dir, p),
3667    }
3668}
3669
3670impl BindingRun<'_> {
3671    /// What this Binding produced, for its end-of-run line.
3672    fn summary_line(&self) -> String {
3673        let rows: u64 = self.produced.iter().map(|p| p.rows).sum();
3674        let mut parts = Vec::new();
3675        if !self.produced.is_empty() {
3676            let n = self.produced.len();
3677            parts.push(format!(
3678                "{n} result set{}, {} row{}",
3679                if n == 1 { "" } else { "s" },
3680                thousands(rows),
3681                if rows == 1 { "" } else { "s" }
3682            ));
3683        }
3684        let names: Vec<String> = self
3685            .files()
3686            .map(|(f, _)| f.file_name().unwrap_or_default().to_string_lossy().to_string())
3687            .collect();
3688        if !names.is_empty() {
3689            parts.push(format!("→ {}", names.join(", ")));
3690        }
3691        // Every destination's location, not just the first one each file reached.
3692        let delivered: Vec<&str> = self
3693            .outs
3694            .iter()
3695            .flat_map(|o| o.deliveries.iter())
3696            .filter_map(|d| d.location.as_deref())
3697            .collect();
3698        let notes: Vec<&str> = self
3699            .outs
3700            .iter()
3701            .filter(|o| !o.dests.is_empty())
3702            .filter_map(|o| o.delivery_note.as_deref())
3703            .collect();
3704        if !delivered.is_empty() {
3705            parts.push(format!("→ {}", delivered.join(", ")));
3706        } else if !notes.is_empty() && self.opts.preview.is_none() {
3707            parts.push(format!("({})", notes.join("; ")));
3708        }
3709        parts.join(" ")
3710    }
3711
3712    /// Every file the Binding's outputs wrote, and where each was first delivered.
3713    fn files(&self) -> impl Iterator<Item = &(PathBuf, Option<String>)> {
3714        self.outs.iter().flat_map(|o| o.files.iter())
3715    }
3716}
3717
3718/// `1234567` → `1,234,567`.
3719pub fn thousands(n: u64) -> String {
3720    let s = n.to_string();
3721    let mut out = String::new();
3722    for (i, c) in s.chars().enumerate() {
3723        if i > 0 && (s.len() - i).is_multiple_of(3) {
3724            out.push(',');
3725        }
3726        out.push(c);
3727    }
3728    out
3729}
3730
3731fn human_bytes(n: u64) -> String {
3732    match n {
3733        n if n < 1024 => format!("{n} B"),
3734        n if n < 1024 * 1024 => format!("{:.1} KB", n as f64 / 1024.0),
3735        n => format!("{:.1} MB", n as f64 / (1024.0 * 1024.0)),
3736    }
3737}
3738
3739/// `dre clean`: remove the target folder `t`.
3740pub fn clean(t: &Path) -> std::io::Result<bool> {
3741    if t.exists() {
3742        std::fs::remove_dir_all(t)?;
3743        Ok(true)
3744    } else {
3745        Ok(false)
3746    }
3747}