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 =
1446            crate::variables::check_values(&self.report.var_schema, &self.vars, &self.opts.vars, true);
1447        if !errors.is_empty() {
1448            return Err(errors.join("; ").into());
1449        }
1450        // The parse pass with this run's inputs: each query's connection and sources.
1451        self.parsed = crate::parse::binding(self.project, self.report, self.b, &self.vars, &self.inputs());
1452        if !self.parsed.errors.is_empty() {
1453            let msgs: Vec<String> = self
1454                .parsed
1455                .errors
1456                .iter()
1457                .map(|p| match p.line {
1458                    Some(l) => format!("{}:{l}: {}", p.file.display(), p.message),
1459                    None => format!("{}: {}", p.file.display(), p.message),
1460                })
1461                .collect();
1462            return Err(Fail::new(Code::ParseFailed, msgs.join("\n    ")));
1463        }
1464        // Sessions are opened only when something needs the database, so compiling a report
1465        // whose templates don't query it works without the plugin or credentials.
1466        let pool = Arc::new(Pool::new(
1467            self.connections(),
1468            self.project.root.clone(),
1469            self.ui.plugin_log(),
1470            !self.report.managed,
1471        ));
1472        self.pool = Some(pool.clone());
1473
1474        if !self.report.managed {
1475            self.ui.warn(&format!(
1476                "`{}` is an unmanaged report ({}): for quick tests only — add a YAML to make it a managed report",
1477                self.report.name,
1478                self.report.file.display()
1479            ));
1480        }
1481
1482        // 1. Render.
1483        std::fs::create_dir_all(&self.compiled_dir).map_err(|e| e.to_string())?;
1484        // How many queries run on each connection: one session holds them all.
1485        let mut per_connection: BTreeMap<String, usize> = BTreeMap::new();
1486        for q in &self.b.queries {
1487            *per_connection.entry(self.connection_of(&q.query)?).or_default() += 1;
1488        }
1489        // A real run of a managed report renders each query just before running it, so a
1490        // template can look at what earlier queries made (`columns()` of a temp table).
1491        // Compiling, checking and unmanaged reports render everything first.
1492        let interleave = !self.opts.dry_run && !self.opts.live_check && self.report.managed;
1493        let mut statements = Vec::new();
1494        // Reported after the unmanaged-report check, which matters more.
1495        let mut two_tabs: Option<String> = None;
1496        if interleave {
1497            self.render_destinations().map_err(|e| e.or(Code::RenderFailed))?;
1498            let mut checked: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
1499            for (c, n) in &per_connection {
1500                if *n > 1 {
1501                    self.check_sessions(&pool, c, *n)?;
1502                    checked.insert(c.clone());
1503                }
1504            }
1505            std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
1506            let mut i = 0;
1507            let queries = self.b.queries.clone();
1508            for q in &queries {
1509                let conn = self.connection_of(&q.query)?;
1510                let renderer = self.renderer(Some(&conn))?;
1511                let sts = self.render_query(&renderer, q, &conn, &mut two_tabs)?;
1512                for w in renderer.take_warnings() {
1513                    self.ui.warn(&w);
1514                }
1515                if let Some(e) = two_tabs.take() {
1516                    return Err(e.into());
1517                }
1518                if sts.len() > 1 && checked.insert(conn.clone()) {
1519                    self.check_sessions(&pool, &conn, sts.len())?;
1520                }
1521                let session = pool.session(&conn)?;
1522                for st in &sts {
1523                    self.execute_statement(&session, i, st)?;
1524                    i += 1;
1525                }
1526            }
1527        } else {
1528            let queries = self.b.queries.clone();
1529            for q in &queries {
1530                let conn = self.connection_of(&q.query)?;
1531                let renderer = self.renderer(Some(&conn))?;
1532                statements.extend(self.render_query(&renderer, q, &conn, &mut two_tabs)?);
1533            }
1534            self.render_destinations().map_err(|e| e.or(Code::RenderFailed))?;
1535            for r in self.renderers.values() {
1536                for w in r.take_warnings() {
1537                    self.ui.warn(&w);
1538                }
1539            }
1540
1541            // 3. Unmanaged: every rendered statement must only read (or create temp objects).
1542            if !self.report.managed {
1543                if let Some(bad) = statements.iter().find(|s| !s.kind.is_read_only_safe()) {
1544                    return Err(format!(
1545                        "{}:{}: 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",
1546                        bad.file.display(),
1547                        bad.line,
1548                        self.report.name,
1549                        masked_sql_summary(&bad.text, 60, bad.sensitive)
1550                    ).into());
1551                }
1552                for conn in per_connection.keys() {
1553                    let session = pool.session(conn)?;
1554                    let creates_temp = statements
1555                        .iter()
1556                        .any(|s| &s.connection == conn && s.kind == StatementKind::TempCreate)
1557                        || session.lock().unwrap().loaded;
1558                    session.lock().unwrap().want_read_only(!creates_temp);
1559                }
1560            }
1561            if let Some(e) = two_tabs {
1562                return Err(e.into());
1563            }
1564
1565            if self.opts.dry_run {
1566                let plan = self.plan();
1567                self.ui.compiled(&plan);
1568                return Ok(());
1569            }
1570
1571            // The one-session-per-connection guarantee.
1572            for conn in per_connection.keys() {
1573                let n = statements.iter().filter(|s| &s.connection == conn).count();
1574                if n > 1 {
1575                    self.check_sessions(&pool, conn, n)?;
1576                }
1577            }
1578
1579            if self.opts.live_check {
1580                return self.live_check(&statements);
1581            }
1582
1583            // 4. Execute in YAML order, each on its connection's session.
1584            std::fs::create_dir_all(self.run_dir.join(".spool")).map_err(|e| e.to_string())?;
1585            for (i, st) in statements.iter().enumerate() {
1586                let session = pool.session(&st.connection)?;
1587                self.execute_statement(&session, i, st)?;
1588            }
1589        }
1590        pool.close_all();
1591
1592        self.name_result_sets()?;
1593        if let Some(dir) = &self.opts.output_json {
1594            let dir = self
1595                .project
1596                .root
1597                .join(dir)
1598                .join(&self.report.name)
1599                .join(self.b.dir_name());
1600            std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
1601            for result in &self.produced {
1602                crate::rows::write_spool(
1603                    &result.spool,
1604                    &dir.join(format!("{}.json", result.name)),
1605                    result.rows,
1606                )
1607                .map_err(|e| e.to_string())?;
1608            }
1609        }
1610
1611        // 5. Format every file output into target/run/; one whose `when:` is false is skipped.
1612        let (file_outs, message_outs): (Vec<usize>, Vec<usize>) =
1613            (0..self.outs.len()).partition(|&i| !self.b.outputs[i].is_message());
1614        for &oi in &file_outs {
1615            if !self.passes_when(oi)? {
1616                continue;
1617            }
1618            let filename = self.file_names(oi);
1619            self.format(oi, &filename).map_err(|e| e.or(Code::FormatFailed))?;
1620        }
1621
1622        // Schema drift, before delivery.
1623        if self.opts.preview.is_none() {
1624            self.drift = self.schema_drift();
1625            if !self.drift.is_empty() {
1626                if self.opts.accept_schema_change {
1627                    self.ui.warn(&format!(
1628                        "  schema changed since the last successful run (accepted): {}",
1629                        self.drift.join("; ")
1630                    ));
1631                } else {
1632                    return Err(format!(
1633                        "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",
1634                        self.drift.join("; "),
1635                        rel(&self.project.root, &self.run_dir).display()
1636                    ).into());
1637                }
1638            }
1639        }
1640
1641        // 6. Deliver the file outputs, then render and deliver each message, so a message can
1642        // link to what was delivered. A failed output doesn't stop the next one.
1643        let preview = self.opts.preview.is_some() || self.opts.no_deliver;
1644        let mut failures = Vec::new();
1645        for oi in file_outs.into_iter().chain(message_outs) {
1646            if self.outs[oi].status == Some(OutputStatus::Skipped) {
1647                continue;
1648            }
1649            if self.b.outputs[oi].is_message() {
1650                match self.render_message(oi) {
1651                    Ok(true) => {}
1652                    Ok(false) => continue,
1653                    Err(e) => {
1654                        let e = e.or(Code::RenderFailed);
1655                        let e = Fail::new(e.code.clone(), format!("{}: {e}", self.b.outputs[oi].label(oi)));
1656                        self.outs[oi].status = Some(OutputStatus::Failed);
1657                        self.outs[oi].error = Some(e.message.clone());
1658                        failures.push(e);
1659                        continue;
1660                    }
1661                }
1662            }
1663            if preview {
1664                self.outs[oi].delivery_note = Some(if self.opts.no_deliver {
1665                    "--no-deliver: output kept locally".into()
1666                } else {
1667                    "preview: not delivered".into()
1668                });
1669                self.outs[oi].status = Some(OutputStatus::Kept);
1670            } else if let Err(e) = self.deliver(oi) {
1671                failures.push(e.or(Code::DeliveryFailed));
1672            }
1673        }
1674        if preview {
1675            let dir = rel(&self.project.root, &self.run_dir);
1676            self.ui.step(
1677                Level::Info,
1678                "Preview",
1679                &format!("not delivered; output stays in {}", dir.display()),
1680                None,
1681            );
1682        }
1683        if let Some(first) = failures.first() {
1684            let messages: Vec<&str> = failures.iter().map(|f| f.message.as_str()).collect();
1685            return Err(Fail::new(first.code.clone(), messages.join("; ")));
1686        }
1687
1688        // 7. Snapshot the schema for the next drift check (not from a run for an earlier instant
1689        // than the current one).
1690        if self.opts.preview.is_none()
1691            && self.becomes_current
1692            && let Err(error) = self.write_snapshot()
1693        {
1694            self.ui.warn(&format!(
1695                "delivery completed, but the schema snapshot could not be written: {error}; the next run may compare against an older schema"
1696            ));
1697        }
1698        Ok(())
1699    }
1700
1701    /// Output `oi`'s queries' results, `rows` capped at `max_rows`.
1702    fn results_for(&mut self, oi: usize, max_rows: u64) -> Result<Vec<(String, Arc<QueryResult>)>, Fail> {
1703        let o = &self.b.outputs[oi];
1704        let mut out = Vec::new();
1705        for p in self.produced.iter().filter(|p| o.feeds(&p.query)) {
1706            let r = QueryResult::read(&p.spool, p.rows, max_rows, self.calendar)
1707                .map_err(|e| format!("can't read the result of `{}`: {e}", p.query))?;
1708            out.push((p.query.clone(), Arc::new(r)));
1709        }
1710        Ok(out)
1711    }
1712
1713    /// What output `oi`'s templates read besides the usual context: `results` and `outputs`
1714    /// (every other named output: `location`, `files` and `status`).
1715    fn template_context(&self, oi: usize, results: &[(String, Arc<QueryResult>)]) -> minijinja::Value {
1716        use minijinja::Value;
1717        let results: BTreeMap<String, Value> = results
1718            .iter()
1719            .map(|(q, r)| (q.clone(), r.clone().value()))
1720            .collect();
1721        let mut outputs: BTreeMap<String, Value> = BTreeMap::new();
1722        for (j, (o, run)) in self.b.outputs.iter().zip(&self.outs).enumerate() {
1723            let Some(name) = &o.name else { continue };
1724            if j == oi {
1725                continue;
1726            }
1727            let locations: Vec<&str> = run
1728                .deliveries
1729                .iter()
1730                .filter_map(|d| d.location.as_deref())
1731                .collect();
1732            let files: Vec<String> = run
1733                .files
1734                .iter()
1735                .map(|(f, _)| record_path(self.project, f).to_string_lossy().to_string())
1736                .collect();
1737            let status = run.status.map_or("pending", OutputStatus::as_str);
1738            outputs.insert(
1739                name.clone(),
1740                Value::from_serialize(json!({
1741                    "location": locations.join(", "),
1742                    "files": files,
1743                    "status": status,
1744                })),
1745            );
1746        }
1747        Value::from(BTreeMap::from([
1748            ("results".to_string(), Value::from(results)),
1749            ("outputs".to_string(), Value::from(outputs)),
1750        ]))
1751    }
1752
1753    /// Evaluate output `oi`'s `when:`. False: the output is skipped (recorded, not a failure).
1754    fn passes_when(&mut self, oi: usize) -> Result<bool, Fail> {
1755        if self.b.outputs[oi].when.is_none() {
1756            return Ok(true);
1757        }
1758        let results = self.results_for(oi, crate::message::DEFAULT_MAX_ROWS)?;
1759        self.when_holds(oi, &results)
1760    }
1761
1762    /// Output `oi`'s `when:` over `results` (the same ones its message reads).
1763    fn when_holds(&mut self, oi: usize, results: &[(String, Arc<QueryResult>)]) -> Result<bool, Fail> {
1764        let Some(when) = self.b.outputs[oi].when.clone() else {
1765            return Ok(true);
1766        };
1767        let ctx = self.template_context(oi, results);
1768        let renderer = self.renderer(None)?;
1769        let label = self.b.outputs[oi].label(oi);
1770        let src = format!("{{% if {when} %}}true{{% endif %}}");
1771        let passed = renderer
1772            .render_with(&self.report.file, &src, ctx, false)
1773            .map_err(|e| format!("{label}: `when` `{when}`: {}", e.message))?
1774            == "true";
1775        self.outs[oi].when = Some(passed);
1776        if !passed {
1777            self.skip(oi, &format!("`when` is false ({when})"));
1778        }
1779        Ok(passed)
1780    }
1781
1782    fn skip(&mut self, oi: usize, why: &str) {
1783        let label = self.b.outputs[oi].label(oi);
1784        self.outs[oi].status = Some(OutputStatus::Skipped);
1785        self.outs[oi].delivery_note = Some(format!("skipped: {why}"));
1786        self.ui
1787            .step(Level::Info, "Skipped", &format!("{label}: {why}"), None);
1788    }
1789
1790    /// Render message output `oi` and write its `.md` file. `Ok(false)`: skipped (`when:` false,
1791    /// or the text rendered empty).
1792    fn render_message(&mut self, oi: usize) -> Result<bool, Fail> {
1793        let o = &self.b.outputs[oi];
1794        let opt = |k: &str| o.options.get(k).and_then(Json::as_str).map(str::to_string);
1795        let (text_src, file_src, title_src) = (opt("text"), opt("file"), opt("title"));
1796        let max_rows = o
1797            .options
1798            .get("max_rows")
1799            .and_then(Json::as_u64)
1800            .unwrap_or(crate::message::DEFAULT_MAX_ROWS);
1801        let results = self.results_for(oi, max_rows)?;
1802        if !self.when_holds(oi, &results)? {
1803            return Ok(false);
1804        }
1805        for (q, r) in results.iter().filter(|(_, r)| r.capped()) {
1806            self.ui.warn(&format!(
1807                "  results.{q}.rows holds the first {} of {} rows (`max_rows`); aggregate in SQL, or raise `max_rows`",
1808                thousands(r.rows.len() as u64),
1809                thousands(r.row_count)
1810            ));
1811        }
1812        let ctx = self.template_context(oi, &results);
1813        let renderer = self.renderer(None)?;
1814        let file = self.report.file.clone();
1815        let text = match (text_src, file_src) {
1816            (Some(t), _) => renderer
1817                .render_with(&file, &t, ctx.clone(), true)
1818                .map_err(|e| e.message)?,
1819            (None, Some(f)) => {
1820                let root = &self.project.root;
1821                let path = crate::project::find_template(root, &f)
1822                    .ok_or_else(|| format!("message file `{f}` doesn't exist"))?;
1823                let src = std::fs::read_to_string(&path).map_err(|e| format!("{}: {e}", path.display()))?;
1824                let shown = rel(root, &path);
1825                renderer
1826                    .render_with(&shown, &src, ctx.clone(), true)
1827                    .map_err(|e| e.to_string())?
1828            }
1829            (None, None) => crate::message::default_text(&results, self.locale()),
1830        };
1831        let text = text.trim().to_string();
1832        if text.is_empty() {
1833            self.skip(oi, "the message is empty");
1834            return Ok(false);
1835        }
1836        let title = match title_src {
1837            Some(t) => renderer
1838                .render_with(&file, &t, ctx, false)
1839                .map_err(|e| e.message)?,
1840            None => format!("{}: {}", self.report.name, self.date),
1841        };
1842        let title = title.split_whitespace().collect::<Vec<_>>().join(" ");
1843        let name = self.file_names(oi);
1844        let path = self.run_dir.join(&name);
1845        if let Some(prev) = (0..oi).find(|&j| self.outs[j].files.iter().any(|(f, _)| f == &path)) {
1846            return Err(format!(
1847                "it writes {name}, as {} does; give them different `name:`s",
1848                self.b.outputs[prev].label(prev)
1849            )
1850            .into());
1851        }
1852        std::fs::create_dir_all(&self.run_dir).map_err(|e| e.to_string())?;
1853        std::fs::write(&path, crate::message::file_body(&title, &text)).map_err(|e| e.to_string())?;
1854        self.outs[oi].files.push((path, None));
1855        if self.opts.preview.is_some() {
1856            let mut notes = Vec::new();
1857            if let Some(n) = self.opts.preview {
1858                notes.push(format!(
1859                    "numbers come from a sample of at most {} rows per query (--preview)",
1860                    thousands(n)
1861                ));
1862            }
1863            if let Some(w) = &self.b.outputs[oi].when {
1864                notes.push(format!("`when` passed ({w})"));
1865            }
1866            let length = text.chars().count();
1867            notes.push(format!("{} characters", thousands(length as u64)));
1868            notes.extend(self.destination_limits(oi, length));
1869            self.ui.message(&ShownMessage {
1870                output: self.b.outputs[oi].name.clone(),
1871                title: title.clone(),
1872                text: text.clone(),
1873                notes,
1874            });
1875        } else {
1876            self.ui.step(
1877                Level::Info,
1878                "Message",
1879                &crate::message::excerpt(&title, &text, 100),
1880                None,
1881            );
1882        }
1883        self.outs[oi].message = Some((title, text));
1884        Ok(true)
1885    }
1886
1887    /// For `--preview`: how each destination of message output `oi` takes it, and its limit.
1888    fn destination_limits(&self, oi: usize, length: usize) -> Vec<String> {
1889        let mut out = Vec::new();
1890        let Some(pool) = &self.pool else { return out };
1891        for d in &self.outs[oi].dests {
1892            let Some(t) = self.project.profiles.target(Role::Destination, &d.profile) else {
1893                out.push(format!("{}: delivers nowhere on this target", d.profile));
1894                continue;
1895            };
1896            let described = pool.connections.described(Role::Destination, &t.kind);
1897            let takes = described
1898                .as_ref()
1899                .is_some_and(|x| x.capabilities.iter().any(|c| c == CAP_MESSAGE));
1900            let limit = described.and_then(|x| x.message_limit);
1901            out.push(match (takes, limit) {
1902                (false, _) => format!("{} ({}): delivers the .md file", d.profile, t.kind),
1903                (true, Some(l)) if length as u64 > l => format!(
1904                    "{} ({}): {} of {} characters: over the limit, so it will be cut short",
1905                    d.profile,
1906                    t.kind,
1907                    thousands(length as u64),
1908                    thousands(l)
1909                ),
1910                (true, Some(l)) => format!(
1911                    "{} ({}): {} of {} characters",
1912                    d.profile,
1913                    t.kind,
1914                    thousands(length as u64),
1915                    thousands(l)
1916                ),
1917                (true, None) => format!("{} ({}): posts the message", d.profile, t.kind),
1918            });
1919        }
1920        out
1921    }
1922
1923    /// The Binding's locale for the number filters.
1924    fn locale(&self) -> crate::numbers::Locale {
1925        self.b
1926            .locale
1927            .as_deref()
1928            .and_then(|l| crate::numbers::Locale::parse(l).ok())
1929            .unwrap_or_default()
1930    }
1931
1932    /// Render one query file into its statements, writing it to `target/compiled/`.
1933    fn render_query(
1934        &mut self,
1935        renderer: &Renderer,
1936        q: &QueryEntry,
1937        connection: &str,
1938        two_tabs: &mut Option<String>,
1939    ) -> Result<Vec<Statement>, Fail> {
1940        let mut out = Vec::new();
1941        let src = std::fs::read_to_string(self.project.root.join(&q.path))
1942            .map_err(|e| format!("{}: {e}", q.path.display()))?;
1943        let _ = renderer.take_sources();
1944        let sql = renderer
1945            .render(&q.path, &src)
1946            .map_err(|e: RenderError| e.to_string())?;
1947        // The manifest's `depends_on` and the query's connection come from the parse pass, so
1948        // a `source()` it couldn't see would make both wrong.
1949        let known = self
1950            .parsed
1951            .query(&q.query)
1952            .map(|p| p.sources.clone())
1953            .unwrap_or_default();
1954        if let Some((s, t)) = renderer
1955            .take_sources()
1956            .into_iter()
1957            .find(|(s, t)| !known.contains(&format!("{s}.{t}")))
1958        {
1959            return Err(format!(
1960                "{}: `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",
1961                q.path.display(),
1962                q.query
1963            ).into());
1964        }
1965        let sensitive = crate::secrets::contains_secret(&sql);
1966        std::fs::write(
1967            self.compiled_dir.join(format!("{}.sql", q.query)),
1968            crate::secrets::mask(&sql).as_bytes(),
1969        )
1970        .map_err(|e| e.to_string())?;
1971        // Hash the exact rendered SQL before masking display artifacts. Integrations can
1972        // identify executed SQL without receiving any secret embedded in it.
1973        {
1974            use sha2::Digest;
1975            let digest = sha2::Sha256::digest(sql.as_bytes());
1976            let hash = digest
1977                .iter()
1978                .map(|byte| format!("{byte:02x}"))
1979                .collect::<String>();
1980            std::fs::write(self.compiled_dir.join(format!("{}.sha256", q.query)), hash)
1981                .map_err(|e| e.to_string())?;
1982        }
1983        self.ui
1984            .step(Level::Debug, "Rendered", &q.path.display().to_string(), None);
1985        // 2. Split. One file makes at most one tab, from its last statement, so any earlier
1986        // SELECT would be lost: that's an error, not a guess.
1987        let parts = sqlsplit::split(&sql);
1988        if q.tab && parts.is_empty() {
1989            return Err(format!(
1990                "{}: `{}` makes a tab but has no statements; add a query, or `tab: false` if it's meant to be empty",
1991                q.path.display(),
1992                q.query
1993            ).into());
1994        }
1995        let n = parts.len();
1996        for (i, st) in parts.into_iter().enumerate() {
1997            let body = sqlsplit::strip_leading_comments(&st.text);
1998            let line = st.line + st.text[..st.text.len() - body.len()].matches('\n').count();
1999            let kind = sqlsplit::classify(&st.text);
2000            let last = i + 1 == n;
2001            if q.tab && !last && kind == StatementKind::Read && two_tabs.is_none() {
2002                *two_tabs = Some(format!(
2003                    "{}:{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:`",
2004                    q.path.display(),
2005                    q.query
2006                ));
2007            }
2008            out.push(Statement {
2009                query: q.query.clone(),
2010                connection: connection.to_string(),
2011                file: q.path.clone(),
2012                line,
2013                kind,
2014                text: st.text,
2015                sensitive,
2016                tab: q.tab && last,
2017            });
2018        }
2019        Ok(out)
2020    }
2021
2022    /// Fail unless `connection`'s plugin can hold one session across `n` statements.
2023    fn check_sessions(&self, pool: &Pool, connection: &str, n: usize) -> Result<(), Fail> {
2024        let session = pool.session(connection)?;
2025        let mut s = session.lock().unwrap();
2026        if !s
2027            .get()
2028            .map_err(|e| Fail::new(Code::ConnectionFailed, e))?
2029            .has(CAP_SESSIONS)
2030        {
2031            let kind = pool.kind(connection).unwrap_or_default();
2032            return Err(format!(
2033                "this Binding runs {n} statements on connection `{connection}`, but the `{kind}` source plugin can't hold one session across them; nothing was run"
2034            ).into());
2035        }
2036        Ok(())
2037    }
2038
2039    /// Run statement `i`, spooling its result if it makes a tab.
2040    fn execute_statement(
2041        &mut self,
2042        session: &Arc<Mutex<Session>>,
2043        i: usize,
2044        st: &Statement,
2045    ) -> Result<(), Fail> {
2046        self.not_cancelled()?;
2047        let sql_log = self.ui.sql_log();
2048        sql_log(
2049            &format!("{}:{}", st.file.display(), st.line),
2050            &protected_sql(&st.text, st.sensitive),
2051        );
2052        let spool_path = self.run_dir.join(".spool").join(format!("{i}.arrow"));
2053        let mut writer: Option<FileWriter<File>> = None;
2054        let t = Instant::now();
2055        let exec = {
2056            let mut s = session.lock().unwrap();
2057            let (p, _log_scope) = s
2058                .statement(st.sensitive)
2059                .map_err(|e| Fail::new(Code::ConnectionFailed, e))?;
2060            p.execute(&st.text, self.opts.preview, |schema, batch| {
2061                if writer.is_none() {
2062                    let f = File::create(&spool_path).map_err(|e| e.to_string())?;
2063                    writer = Some(FileWriter::try_new(f, schema).map_err(|e| e.to_string())?);
2064                }
2065                writer.as_mut().unwrap().write(&batch).map_err(|e| e.to_string())
2066            })
2067            .map_err(|e| {
2068                let message = format!(
2069                    "{}:{}: {}",
2070                    st.file.display(),
2071                    st.line,
2072                    masked_source_error(&e, st.sensitive)
2073                );
2074                Fail::from_plugin(&e, message).or(Code::QueryFailed)
2075            })?
2076        };
2077        let what = match &exec {
2078            Execution::Result { rows, .. } => {
2079                format!("{} row{}", thousands(*rows), if *rows == 1 { "" } else { "s" })
2080            }
2081            Execution::NoResult {
2082                rows_affected: Some(n),
2083            } => format!("no result set ({n} affected)"),
2084            Execution::NoResult { .. } => "no result set".to_string(),
2085        };
2086        let at = format!("{}:{}", st.file.display(), st.line);
2087        self.ui.step(
2088            Level::Debug,
2089            "Executed",
2090            &format!("{at}  {what}"),
2091            Some(t.elapsed()),
2092        );
2093        // The YAML decides the tabs; the result only has to match it. A tab with no rows
2094        // still gets its column names.
2095        let (schema, rows) = match (st.tab, exec) {
2096            (true, Execution::Result { schema, rows }) => (schema, rows),
2097            (true, Execution::NoResult { .. }) => {
2098                return Err(Fail::new(
2099                    Code::NoResultSet,
2100                    format!(
2101                        "{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",
2102                        st.query
2103                    ),
2104                ));
2105            }
2106            (false, _) => {
2107                drop(writer);
2108                let _ = std::fs::remove_file(&spool_path);
2109                return Ok(());
2110            }
2111        };
2112        {
2113            let mut w = match writer {
2114                Some(w) => w,
2115                None => FileWriter::try_new(File::create(&spool_path).map_err(|e| e.to_string())?, &schema)
2116                    .map_err(|e| e.to_string())?,
2117            };
2118            w.finish().map_err(|e| e.to_string())?;
2119            self.produced.push(Produced {
2120                query: st.query.clone(),
2121                connection: st.connection.clone(),
2122                schema,
2123                rows,
2124                spool: spool_path,
2125                name: String::new(),
2126                anchor: None,
2127                header: None,
2128                columns: Default::default(),
2129                autofit: None,
2130                style: None,
2131            });
2132        }
2133        Ok(())
2134    }
2135
2136    /// Each destination's rendered `profile`, `path` and options; `destination.*` is that
2137    /// destination while its values render.
2138    fn render_destinations(&mut self) -> Result<(), Fail> {
2139        if self.opts.no_deliver {
2140            return Ok(());
2141        }
2142        if self.b.destinations().next().is_none() {
2143            return Ok(());
2144        }
2145        let renderer = self.renderer(None)?;
2146        let connections = self.pool.as_ref().expect("pool").connections.clone();
2147        // Destinations are numbered across every output, as the parse pass renders them.
2148        let numbered: Vec<(usize, usize, &crate::project::Destination)> = self
2149            .b
2150            .outputs
2151            .iter()
2152            .enumerate()
2153            .flat_map(|(oi, o)| o.destinations.iter().map(move |d| (oi, d)))
2154            .enumerate()
2155            .map(|(i, (oi, d))| (i, oi, d))
2156            .collect();
2157        for (i, oi, d) in numbered {
2158            let profile = self
2159                .parsed
2160                .destinations
2161                .get(i)
2162                .cloned()
2163                .flatten()
2164                .ok_or_else(|| format!("destination `{}` has no profile", d.profile))?;
2165            // Nothing is delivered, so there's no path or options to render.
2166            if matches!(
2167                self.project.profiles.entry(Role::Destination, &profile),
2168                Entry::Nowhere
2169            ) {
2170                self.outs[oi].dests.push(RenderedDest {
2171                    profile,
2172                    path: None,
2173                    options: JsonMap::new(),
2174                    attach: Vec::new(),
2175                });
2176                continue;
2177            }
2178            renderer.set_destination(connections.profile(&profile, Some("destination")).ok());
2179            let rendered = (|| {
2180                let path = match &d.path {
2181                    Some(p) => Some(
2182                        renderer
2183                            .render(&self.report.file, p)
2184                            .map_err(|e| format!("output path: {e}"))?,
2185                    ),
2186                    None => None,
2187                };
2188                let mut options = JsonMap::new();
2189                for (k, v) in &d.options {
2190                    let v = render_json(&renderer, &self.report.file, v)
2191                        .map_err(|e| format!("destination `{profile}` option `{k}`: {e}"))?;
2192                    options.insert(k.clone(), v);
2193                }
2194                Ok::<_, Fail>(RenderedDest {
2195                    profile: profile.clone(),
2196                    path,
2197                    options,
2198                    attach: d.attach.clone(),
2199                })
2200            })();
2201            renderer.set_destination(None);
2202            self.outs[oi].dests.push(rendered?);
2203        }
2204        Ok(())
2205    }
2206
2207    /// Tab names: `tab_name`, else the query's basename.
2208    fn name_result_sets(&mut self) -> Result<(), Fail> {
2209        let entries: BTreeMap<&str, &QueryEntry> =
2210            self.b.queries.iter().map(|q| (q.query.as_str(), q)).collect();
2211        for p in &mut self.produced {
2212            let q = entries[p.query.as_str()];
2213            p.name = q.tab_name.clone().unwrap_or_else(|| q.query.clone());
2214            p.anchor = q.anchor.clone();
2215            p.header = q.header;
2216            p.columns = q.columns.clone();
2217            p.autofit = q.autofit;
2218            p.style = q.style.clone();
2219        }
2220        for o in self.b.outputs.iter().filter(|o| o.format == "xlsx") {
2221            let mut seen: BTreeMap<String, String> = BTreeMap::new();
2222            for p in self.produced.iter().filter(|p| o.feeds(&p.query)) {
2223                check_sheet_name(&p.name)?;
2224                if let Some(prev) = seen.insert(p.name.to_lowercase(), p.query.clone()) {
2225                    return Err(format!(
2226                        "sheet name `{}` is used twice (by `{prev}` and `{}`); Excel sheet names must be unique",
2227                        p.name, p.query
2228                    ).into());
2229                }
2230            }
2231        }
2232        Ok(())
2233    }
2234
2235    /// What a real run of this Binding would use and produce, for `dre compile` and `validate`.
2236    fn plan(&mut self) -> BindingPlan {
2237        let compiled = self
2238            .b
2239            .queries
2240            .iter()
2241            .map(|q| {
2242                rel(
2243                    &self.project.root,
2244                    &self.compiled_dir.join(format!("{}.sql", q.query)),
2245                )
2246            })
2247            .collect();
2248        let mut outputs = Vec::new();
2249        for oi in 0..self.outs.len() {
2250            let file = self.file_names(oi);
2251            let o = &self.b.outputs[oi];
2252            let destinations = self.outs[oi]
2253                .dests
2254                .iter()
2255                .map(|d| {
2256                    let out = self.project.profiles.target(Role::Destination, &d.profile);
2257                    PlannedDestination {
2258                        profile: d.profile.clone(),
2259                        kind: out.map(|o| o.kind.clone()),
2260                        target: Some(self.project.profiles.target_of(Role::Destination, &d.profile)),
2261                        path: d.path.clone(),
2262                        delivers: out.is_some(),
2263                    }
2264                })
2265                .collect();
2266            outputs.push(PlannedOutput {
2267                name: o.name.clone(),
2268                format: o.format.clone(),
2269                output: rel(&self.project.root, &self.run_dir.join(file)),
2270                destinations,
2271            });
2272        }
2273        let first = outputs.first().cloned();
2274        let pool = self.pool.clone().expect("pool");
2275        let queries = self
2276            .parsed
2277            .queries
2278            .iter()
2279            .map(|q| {
2280                let connection = q.connection.clone().unwrap_or_default();
2281                PlannedQuery {
2282                    query: q.query.clone(),
2283                    kind: pool.kind(&connection).unwrap_or_default(),
2284                    target: self.project.profiles.target_of(Role::Connection, &connection),
2285                    connection,
2286                    sources: q.sources.clone(),
2287                }
2288            })
2289            .collect();
2290        BindingPlan {
2291            report: self.report.name.clone(),
2292            set: self.b.set.clone(),
2293            compiled,
2294            profile: self.parsed.inherited.clone(),
2295            queries,
2296            target: self.target.clone(),
2297            format: first.as_ref().map(|o| o.format.clone()).unwrap_or_default(),
2298            output: first.as_ref().map(|o| o.output.clone()).unwrap_or_default(),
2299            destinations: first.map(|o| o.destinations).unwrap_or_default(),
2300            outputs,
2301            schedules: self.b.schedules.clone(),
2302        }
2303    }
2304
2305    /// Apply `--output-path`/`--output-name` to every destination of the first output that has a
2306    /// path, and return output `oi`'s local file name: `--output-name` (first output only), else
2307    /// the first destination path's file name, else `<name>.<ext>` (the output's `name:`, else
2308    /// the report's).
2309    fn file_names(&mut self, oi: usize) -> String {
2310        let first = oi == 0;
2311        let (output_path, output_name) = if first {
2312            (self.opts.output_path.clone(), self.opts.output_name.clone())
2313        } else {
2314            (None, None)
2315        };
2316        for d in self.outs[oi].dests.iter_mut().filter(|d| d.path.is_some()) {
2317            if let Some(p) = &output_path {
2318                d.path = Some(p.clone());
2319            }
2320            if let (Some(n), Some(r)) = (&output_name, d.path.as_mut()) {
2321                *r = match r.rfind(['/', '\\']) {
2322                    Some(i) => format!("{}{n}", &r[..=i]),
2323                    None => n.clone(),
2324                };
2325            }
2326        }
2327        let o = &self.b.outputs[oi];
2328        let ext = match &o.extension {
2329            Some(e) => e.as_str(),
2330            None => extension(&o.format),
2331        };
2332        let stem = o.name.clone().unwrap_or_else(|| self.report.name.clone());
2333        let from_remote = self.outs[oi]
2334            .dests
2335            .iter()
2336            .find_map(|d| d.path.as_deref())
2337            .and_then(|r| r.rsplit(['/', '\\']).next())
2338            .filter(|s| !s.is_empty())
2339            .map(str::to_string);
2340        output_name.or(from_remote).unwrap_or_else(|| match ext {
2341            "" => stem,
2342            ext => format!("{stem}.{ext}"),
2343        })
2344    }
2345
2346    /// Write output `oi`'s files from the result sets of its queries.
2347    fn format(&mut self, oi: usize, filename: &str) -> Result<(), Fail> {
2348        self.not_cancelled()?;
2349        std::fs::create_dir_all(&self.run_dir).map_err(|e| e.to_string())?;
2350        let b = self.b;
2351        let out = &b.outputs[oi];
2352        let mine: Vec<usize> = (0..self.produced.len())
2353            .filter(|&i| out.feeds(&self.produced[i].query))
2354            .collect();
2355        if mine.is_empty() {
2356            let what = if self.outs.len() > 1 {
2357                format!("{} gets no tab", out.label(oi))
2358            } else {
2359                "no query makes a tab (every query has `tab: false`)".to_string()
2360            };
2361            self.ui.warn(&format!("  {what}; no output file was written"));
2362            return Ok(());
2363        }
2364        let multi = out.format == "xlsx" || out.template.is_some() || !is_single_table(&out.format);
2365        let groups: Vec<(String, Vec<usize>)> = if multi || mine.len() == 1 {
2366            vec![(filename.to_string(), mine)]
2367        } else {
2368            let (stem, ext) = split_ext(filename);
2369            let mut g = Vec::new();
2370            for i in mine {
2371                let r = &self.produced[i];
2372                if r.name.contains(['/', '\\']) {
2373                    return Err(format!("`{}` can't be used in a file name", r.name).into());
2374                }
2375                g.push((format!("{stem}_{}{ext}", r.name), vec![i]));
2376            }
2377            g
2378        };
2379        // Two outputs writing one file would overwrite each other.
2380        for (name, _) in &groups {
2381            let path = self.run_dir.join(name);
2382            if let Some(prev) = (0..oi).find(|&j| self.outs[j].files.iter().any(|(f, _)| f == &path)) {
2383                return Err(format!(
2384                    "{} and {} both write {name}; give them different `name:`s or destination paths",
2385                    self.b.outputs[prev].label(prev),
2386                    out.label(oi)
2387                )
2388                .into());
2389            }
2390        }
2391        let plugin = find_plugin(self.project, PluginKind::Format, &out.format)
2392            .map_err(|e| Fail::new(Code::FormatFailed, e.to_string()))?;
2393        let mut p = plugin
2394            .start(self.ui.plugin_log(), Some(&self.project.root))
2395            .map_err(|e| e.to_string())?;
2396        let template = match &out.template {
2397            Some(t) => Some(self.template_payload(t)?),
2398            None => None,
2399        };
2400        for (name, idxs) in groups {
2401            let t = Instant::now();
2402            let path = self.run_dir.join(&name);
2403            let metas: Vec<ResultSetMeta> = idxs
2404                .iter()
2405                .map(|&i| {
2406                    let r = &self.produced[i];
2407                    ResultSetMeta {
2408                        name: r.name.clone(),
2409                        query: r.query.clone(),
2410                        result_index: 1,
2411                        anchor: r.anchor.clone(),
2412                        header: r.header,
2413                        columns: r.columns.clone(),
2414                        autofit: r.autofit,
2415                        style: r.style.clone(),
2416                    }
2417                })
2418                .collect();
2419            p.write_begin(
2420                &path.to_string_lossy(),
2421                &out.format,
2422                out.options.clone(),
2423                metas,
2424                template.clone(),
2425            )
2426            .map_err(|e| e.to_string())?;
2427            for &i in &idxs {
2428                let r = &self.produced[i];
2429                let reader = FileReader::try_new(File::open(&r.spool).map_err(|e| e.to_string())?, None)
2430                    .map_err(|e| e.to_string())?;
2431                let schema = r.schema.clone();
2432                // Stream batch by batch so memory stays flat.
2433                let mut any = false;
2434                for b in reader {
2435                    let b = b.map_err(|e| e.to_string())?;
2436                    any = true;
2437                    p.send_batch(&b)
2438                        .map_err(|e| format!("{} format: {e}", out.format))?;
2439                }
2440                if !any {
2441                    p.send_batch(&RecordBatch::new_empty(schema))
2442                        .map_err(|e| format!("{} format: {e}", out.format))?;
2443                }
2444                p.send(&dre_protocol::msg::Request::ResultSetEnd {})
2445                    .map_err(|e| format!("{} format: {e}", out.format))?;
2446            }
2447            let (files, warnings) = p
2448                .write_finish_with_warnings()
2449                .map_err(|e| Fail::from_plugin(&e, format!("{} format: {e}", out.format)))?;
2450            for w in warnings {
2451                self.ui.warn(&format!("  {} format: {w}", out.format));
2452            }
2453            for f in files {
2454                let f = PathBuf::from(f);
2455                let size = std::fs::metadata(&f).map(|m| m.len()).unwrap_or(0);
2456                let shown = rel(&self.project.root, &f);
2457                self.ui.step(
2458                    Level::Debug,
2459                    "Wrote",
2460                    &format!("{} ({})", shown.display(), human_bytes(size)),
2461                    Some(t.elapsed()),
2462                );
2463                self.outs[oi].files.push((f, None));
2464            }
2465        }
2466        let _ = p.close();
2467        Ok(())
2468    }
2469
2470    fn template_payload(&mut self, t: &crate::project::Template) -> Result<Json, Fail> {
2471        let file = crate::project::find_template(&self.project.root, &t.file)
2472            .ok_or_else(|| format!("template file `{}` doesn't exist", t.file))?;
2473        let renderer = self.renderer(None)?;
2474        let mut values = JsonMap::new();
2475        for b in &t.bindings {
2476            if let (Some(cell), Some(v)) = (&b.cell, &b.value) {
2477                let rendered = renderer
2478                    .render(&self.report.file, v)
2479                    .map_err(|e| format!("template value: {e}"))?;
2480                values.insert(format!("{}!{}", b.sheet, cell), Json::String(rendered));
2481            }
2482        }
2483        Ok(json!({"file": file.to_string_lossy(), "bindings": t.bindings, "values": values}))
2484    }
2485
2486    /// Deliver output `oi` to each of its destinations in order. A failure is recorded and the
2487    /// rest are still attempted; the output fails if any did.
2488    fn deliver(&mut self, oi: usize) -> Result<(), Fail> {
2489        let result = self.deliver_output(oi);
2490        let o = &mut self.outs[oi];
2491        o.status = Some(match &result {
2492            Err(_) => OutputStatus::Failed,
2493            Ok(()) if o.deliveries.iter().any(|d| d.status == "delivered") => OutputStatus::Delivered,
2494            Ok(()) => OutputStatus::Kept,
2495        });
2496        o.error = result.as_ref().err().map(|f| f.message.clone());
2497        result
2498    }
2499
2500    fn deliver_output(&mut self, oi: usize) -> Result<(), Fail> {
2501        if self.outs[oi].dests.is_empty() {
2502            self.outs[oi].delivery_note = Some("no destination declared; output stays in target/".into());
2503            let dir = rel(&self.project.root, &self.run_dir);
2504            self.ui.step(
2505                Level::Debug,
2506                "Kept",
2507                &format!("no destination declared: output stays in {}", dir.display()),
2508                None,
2509            );
2510            return Ok(());
2511        }
2512        if self.outs[oi].files.is_empty() {
2513            return Ok(());
2514        }
2515        let dests = std::mem::take(&mut self.outs[oi].dests);
2516        let mut failures = Vec::new();
2517        let mut nowhere = Vec::new();
2518        for d in &dests {
2519            let profiles = &self.project.profiles;
2520            let target = profiles.target_of(Role::Destination, &d.profile);
2521            // `None` for an entry that delivers nowhere.
2522            let kind = profiles
2523                .target(Role::Destination, &d.profile)
2524                .map(|o| o.kind.clone());
2525            self.attempts = 1;
2526            let (status, location, error) =
2527                match self.deliver_one(oi, d).map_err(|e| e.or(Code::DeliveryFailed)) {
2528                    Ok(Some(loc)) => ("delivered", Some(loc), None),
2529                    Ok(None) => {
2530                        nowhere.push(nowhere_note(&d.profile, &target));
2531                        ("not_delivered", None, None)
2532                    }
2533                    Err(e) => {
2534                        failures.push(e.clone());
2535                        ("failed", None, Some(e))
2536                    }
2537                };
2538            let record = Delivery {
2539                profile: d.profile.clone(),
2540                kind,
2541                target,
2542                status,
2543                location,
2544                attempts: Some(self.attempts).filter(|n| *n > 1),
2545                error_code: error.as_ref().map(|e| e.code.clone()),
2546                error: error.map(|e| e.message),
2547            };
2548            self.outs[oi].deliveries.push(record);
2549        }
2550        let o = &mut self.outs[oi];
2551        o.dests = dests;
2552        if o.files.iter().all(|(_, d)| d.is_none()) && !nowhere.is_empty() {
2553            o.delivery_note = Some(nowhere.join("; "));
2554        }
2555        match failures.len() {
2556            0 => Ok(()),
2557            1 if o.dests.len() == 1 => Err(failures.remove(0)),
2558            n => Err(Fail::new(
2559                failures[0].code.clone(),
2560                format!(
2561                    "{n} of {} destinations failed: {}",
2562                    o.dests.len(),
2563                    failures
2564                        .iter()
2565                        .map(|f| f.message.as_str())
2566                        .collect::<Vec<_>>()
2567                        .join("; ")
2568                ),
2569            )),
2570        }
2571    }
2572
2573    /// Deliver every file to one destination. `Ok(None)`: its entry for this run is
2574    /// `deliver: false`, so nothing was sent.
2575    fn deliver_one(&mut self, oi: usize, d: &RenderedDest) -> Result<Option<String>, Fail> {
2576        // An output is never delivered once the run is cancelled.
2577        self.not_cancelled()?;
2578        let profiles = &self.project.profiles;
2579        let out = match profiles.entry(Role::Destination, &d.profile) {
2580            Entry::Use(o) => o,
2581            Entry::Nowhere => {
2582                let note = nowhere_note(&d.profile, &profiles.target_of(Role::Destination, &d.profile));
2583                self.ui.step(Level::Info, "Kept", &note, None);
2584                return Ok(None);
2585            }
2586            // Both are checked before the run starts.
2587            Entry::Missing => {
2588                return Err(Fail::new(
2589                    Code::MissingTargetEntry,
2590                    profiles.missing_entry(Role::Destination, &d.profile),
2591                ));
2592            }
2593            Entry::Unknown => {
2594                return Err(format!("destination profile `{}` isn't in profiles.yml", d.profile).into());
2595            }
2596        };
2597        let kind = out.kind.clone();
2598        let connection = render_connection(out)?;
2599        let files = &self.outs[oi].files;
2600        let targets: Vec<DeliveryFile> = files
2601            .iter()
2602            .map(|(f, _)| {
2603                let name = f.file_name().unwrap().to_string_lossy().to_string();
2604                let r = d.path.as_deref().map(|r| {
2605                    if files.len() == 1 {
2606                        r.to_string()
2607                    } else {
2608                        match r.rfind(['/', '\\']) {
2609                            Some(i) => format!("{}{name}", &r[..=i]),
2610                            None => name.clone(),
2611                        }
2612                    }
2613                });
2614                DeliveryFile {
2615                    local_path: f.to_string_lossy().to_string(),
2616                    remote_path: r,
2617                }
2618            })
2619            .collect();
2620        // Bindings running at once never deliver to the same file.
2621        for t in &targets {
2622            if let Some(r) = &t.remote_path
2623                && let Some(other) = self.store.claim(&d.profile, r, &self.label())
2624            {
2625                return Err(Fail::new(
2626                    Code::DeliveryFailed,
2627                    format!(
2628                        "`{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)",
2629                        d.profile
2630                    ),
2631                ));
2632            }
2633        }
2634        let mut locations = Vec::new();
2635        if kind == LOCAL_TYPE {
2636            let errors = crate::options::local_option_errors(&d.options);
2637            if !errors.is_empty() {
2638                return Err(format!("destination `{}`: {}", d.profile, errors.join("; ")).into());
2639            }
2640            // `if_exists`, `atomic` (default true: written as `.<name>.dre-part`, then renamed)
2641            // and `temp_dir`.
2642            let (rules, _) = dre_protocol::delivery::Rules::from_settings(
2643                dre_protocol::delivery::Rules {
2644                    retries: 0,
2645                    ..Default::default()
2646                },
2647                &JsonMap::new(),
2648                &d.options,
2649                &[],
2650            )?;
2651            for (i, f) in targets.iter().enumerate() {
2652                let t = Instant::now();
2653                let r = f
2654                    .remote_path
2655                    .as_ref()
2656                    .ok_or("the local destination needs `output.destination.path`")?;
2657                let dst = self.project.root.join(r).to_string_lossy().to_string();
2658                let delivered = dre_protocol::delivery::deliver(
2659                    &mut dre_protocol::delivery::LocalStore,
2660                    Path::new(&f.local_path),
2661                    &dst,
2662                    &rules,
2663                )
2664                .map_err(|e| {
2665                    let message = format!("delivery to {dst} failed: {e}; the output is still in target/");
2666                    if e.exists {
2667                        // `if_exists: error`, coded as a plugin's would be.
2668                        let code = crate::codes::ErrorCode::Plugin {
2669                            code: "local/file-exists".into(),
2670                            kind: crate::codes::Kind::Delivery,
2671                        };
2672                        Fail::new(code, message)
2673                    } else {
2674                        Fail::new(Code::DeliveryFailed, message)
2675                    }
2676                })?;
2677                let loc = delivered.path;
2678                self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
2679                self.outs[oi].files[i].1.get_or_insert_with(|| loc.clone());
2680                locations.push(loc);
2681            }
2682            return Ok(Some(locations.join(", ")));
2683        }
2684        let failed = |e: dre_protocol::host::HostError| {
2685            Fail::from_plugin(
2686                &e,
2687                format!("delivery through `{kind}` failed: {e}; the output is still in target/"),
2688            )
2689        };
2690        let plugin = find_plugin(self.project, PluginKind::Destination, &kind)
2691            .map_err(|e| Fail::new(Code::DeliveryFailed, e.to_string()))?;
2692        let mut p = plugin
2693            .start(self.ui.plugin_log(), Some(&self.project.root))
2694            .map_err(|e| e.to_string())?;
2695        let is_message = self.b.outputs[oi].is_message();
2696        // A message goes to a destination that takes messages as a message; to any other, as
2697        // its `.md` file.
2698        if is_message
2699            && p.has(CAP_MESSAGE)
2700            && let Some((title, text)) = self.outs[oi].message.clone()
2701        {
2702            let t = Instant::now();
2703            let message = dre_protocol::msg::Message {
2704                title,
2705                text,
2706                html: None,
2707                path: targets[0].local_path.clone(),
2708            };
2709            let attach: Vec<DeliveryFile> = d
2710                .attach
2711                .iter()
2712                .filter_map(|name| {
2713                    self.b
2714                        .outputs
2715                        .iter()
2716                        .position(|o| o.name.as_deref() == Some(name))
2717                })
2718                .flat_map(|j| self.outs[j].files.iter())
2719                .map(|(f, _)| DeliveryFile {
2720                    local_path: f.to_string_lossy().to_string(),
2721                    remote_path: None,
2722                })
2723                .collect();
2724            let loc = p
2725                .deliver_message(&message, &attach, connection, d.options.clone())
2726                .map_err(failed)?;
2727            self.attempts = p.last_attempts();
2728            self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
2729            self.outs[oi].files[0].1.get_or_insert_with(|| loc.clone());
2730            let _ = p.close();
2731            return Ok(Some(loc));
2732        }
2733        if !is_message && p.has(CAP_MESSAGE_ONLY) {
2734            let _ = p.close();
2735            return Err(format!(
2736                "`{kind}` only takes messages, but this output is `{}`; deliver the file elsewhere and link it from a message",
2737                self.b.outputs[oi].format
2738            ).into());
2739        }
2740        let batches: Vec<Vec<usize>> = if targets.len() > 1 && p.has(CAP_MULTI_FILE) {
2741            vec![(0..targets.len()).collect()]
2742        } else {
2743            (0..targets.len()).map(|i| vec![i]).collect()
2744        };
2745        for batch in batches {
2746            let t = Instant::now();
2747            let files: Vec<DeliveryFile> = batch.iter().map(|&i| targets[i].clone()).collect();
2748            let loc = p
2749                .deliver_files(&files, connection.clone(), d.options.clone())
2750                .map_err(failed)?;
2751            self.attempts = self.attempts.max(p.last_attempts());
2752            self.ui.step(Level::Debug, "Delivered", &loc, Some(t.elapsed()));
2753            for i in batch {
2754                self.outs[oi].files[i].1.get_or_insert_with(|| loc.clone());
2755            }
2756            locations.push(loc);
2757        }
2758        let _ = p.close();
2759        Ok(Some(locations.join(", ")))
2760    }
2761
2762    fn snapshot(&self) -> Json {
2763        let sets: Vec<Json> = self
2764            .produced
2765            .iter()
2766            .map(|p| {
2767                json!({
2768                    "name": p.name,
2769                    "columns": p.schema.fields().iter().map(|f| json!({
2770                        "name": f.name(),
2771                        "type": f.data_type().to_string(),
2772                        "type_shape": crate::schema::type_shape(f.data_type()),
2773                    })).collect::<Vec<_>>(),
2774                })
2775            })
2776            .collect();
2777        json!({ "result_sets": sets })
2778    }
2779
2780    fn schema_drift(&mut self) -> Vec<String> {
2781        let path = self.schema_dir.join("last_success.json");
2782        let mut current = self.snapshot();
2783        let Ok(prev) = std::fs::read_to_string(&path) else {
2784            crate::schema::redact(&mut current);
2785            self.protected_snapshot = Some(current);
2786            return Vec::new();
2787        };
2788        let Ok(mut prev) = serde_json::from_str::<Json>(&prev) else {
2789            crate::schema::redact(&mut current);
2790            self.protected_snapshot = Some(current);
2791            return Vec::new();
2792        };
2793        let original = prev.clone();
2794        crate::schema::redact(&mut prev);
2795        crate::schema::redact(&mut current);
2796        if crate::secrets::enabled() {
2797            crate::schema::align_protection(&mut prev, &mut current);
2798        }
2799        if prev != original {
2800            let protected =
2801                serde_json::to_string_pretty(&prev).expect("serializing a serde_json::Value cannot fail");
2802            if let Err(error) = crate::fs::write_atomic(&path, protected.as_bytes()) {
2803                self.ui.warn(&format!(
2804                    "can't protect the existing schema snapshot at {}: {error}; continuing without rewriting it",
2805                    path.display()
2806                ));
2807            }
2808        }
2809        let drift = crate::schema::drift(&prev, &current);
2810        self.protected_snapshot = Some(current);
2811        drift
2812    }
2813
2814    fn write_snapshot(&self) -> std::io::Result<()> {
2815        std::fs::create_dir_all(&self.schema_dir)?;
2816        let snapshot = self.protected_snapshot.clone().unwrap_or_else(|| {
2817            let mut snapshot = self.snapshot();
2818            crate::schema::redact(&mut snapshot);
2819            snapshot
2820        });
2821        crate::fs::write_atomic(
2822            &self.schema_dir.join("last_success.json"),
2823            serde_json::to_string_pretty(&snapshot)?.as_bytes(),
2824        )
2825    }
2826
2827    fn write_results(&self, status: &Status, error: Option<&Fail>) -> std::io::Result<()> {
2828        std::fs::create_dir_all(&self.run_dir)?;
2829        // `outputs` lists every file (as before 0.3); `output_results` has one entry per output.
2830        let mut outputs: Vec<run_results::OutputFile> = Vec::new();
2831        let mut output_results: Vec<run_results::OutputResult> = Vec::new();
2832        for (o, run) in self.b.outputs.iter().zip(&self.outs) {
2833            let files: Vec<run_results::OutputFileRef> = run
2834                .files
2835                .iter()
2836                .map(|(f, delivered)| run_results::OutputFileRef {
2837                    path: record_path(self.project, f),
2838                    size: std::fs::metadata(f).map(|m| m.len()).unwrap_or(0),
2839                    delivered_to: delivered.clone(),
2840                })
2841                .collect();
2842            outputs.extend(files.iter().map(|f| run_results::OutputFile {
2843                path: f.path.clone(),
2844                size: f.size,
2845                delivered_to: f.delivered_to.clone(),
2846                output: o.name.clone(),
2847            }));
2848            let queries: Vec<String> = self
2849                .b
2850                .queries
2851                .iter()
2852                .filter(|q| q.tab && o.feeds(&q.query))
2853                .map(|q| q.query.clone())
2854                .collect();
2855            let status = match (run.status, status) {
2856                (Some(s), _) => s,
2857                (None, Status::Error) => OutputStatus::Failed,
2858                (None, _) => OutputStatus::Kept,
2859            }
2860            .as_str();
2861            output_results.push(run_results::OutputResult {
2862                name: o.name.clone(),
2863                format: o.format.clone(),
2864                queries,
2865                status,
2866                error: run.error.clone(),
2867                files,
2868                delivery: run.delivery_note.clone(),
2869                deliveries: run.deliveries.clone(),
2870                when: run.when,
2871                message: run
2872                    .message
2873                    .clone()
2874                    .map(|(title, text)| run_results::Message { title, text }),
2875            });
2876        }
2877        // In delivery order: file outputs, then messages.
2878        let order = (0..self.outs.len())
2879            .filter(|&i| !self.b.outputs[i].is_message())
2880            .chain((0..self.outs.len()).filter(|&i| self.b.outputs[i].is_message()));
2881        let deliveries: Vec<Delivery> = order
2882            .flat_map(|i| self.outs[i].deliveries.iter().cloned())
2883            .collect();
2884        let notes: Vec<&str> = self
2885            .outs
2886            .iter()
2887            .filter_map(|o| o.delivery_note.as_deref())
2888            .collect();
2889        let delivery = (!notes.is_empty()).then(|| notes.join("; "));
2890        let results = RunResults {
2891            schema_version: run_results::SCHEMA_VERSION,
2892            run_id: self.run_id.clone(),
2893            report: self.report.name.clone(),
2894            set: self.b.set.clone(),
2895            binding: self.b.dir_name().to_string(),
2896            managed: self.report.managed,
2897            profile: self.parsed.inherited.clone(),
2898            connections: self
2899                .parsed
2900                .connections()
2901                .into_iter()
2902                .map(str::to_string)
2903                .collect(),
2904            target: self.target.clone(),
2905            schedule: self.opts.schedule.clone(),
2906            schedule_vars: self.schedule_vars.clone(),
2907            vars: self.rendered_vars.clone(),
2908            run_date: self.date.to_string(),
2909            scheduled_at: self.opts.scheduled_at.map(rfc3339),
2910            timezone: self.calendar.tz.name().to_string(),
2911            params: self.opts.params(self.date),
2912            status: *status,
2913            error: error.map(|e| e.message.clone()),
2914            error_code: error.map(|e| e.code.clone()),
2915            error_kind: error.map(|e| e.code.kind()),
2916            preview: self.opts.preview.is_some(),
2917            row_limit: self.opts.preview,
2918            started_at: self.started_at.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
2919            duration_ms: self.started.elapsed().as_millis() as u64,
2920            result_sets: self
2921                .produced
2922                .iter()
2923                .map(|p| run_results::ResultSet {
2924                    name: p.name.clone(),
2925                    query: p.query.clone(),
2926                    connection: p.connection.clone(),
2927                    rows: p.rows,
2928                    columns: p.schema.fields().iter().map(|f| f.name().clone()).collect(),
2929                })
2930                .collect(),
2931            outputs,
2932            output_results,
2933            delivery,
2934            deliveries,
2935            schema_drift: self.drift.clone(),
2936            target_path: self.project.target_dir.clone(),
2937            settings: self.project.settings.clone(),
2938            manifest_checksum: self.opts.manifest_checksum.clone(),
2939        };
2940        self.store.write_results(&self.run_dir, &results)
2941    }
2942
2943    /// `dre validate --live`: execute temp creates, `check` everything else, each on its
2944    /// connection; then check declared source columns against the database.
2945    fn live_check(&mut self, statements: &[Statement]) -> Result<(), Fail> {
2946        let pool = self.pool.clone().expect("pool");
2947        let mut failures = Vec::new();
2948        let mut unexecuted_setup: BTreeMap<String, (PathBuf, usize)> = BTreeMap::new();
2949        let mut uncheckable: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
2950        let sql_log = self.ui.sql_log();
2951        for st in statements {
2952            if uncheckable.contains(&st.connection) {
2953                continue;
2954            }
2955            let session = pool.session(&st.connection)?;
2956            let mut s = session.lock().unwrap();
2957            if !s.get()?.has(dre_protocol::CAP_CHECK) {
2958                let kind = pool.kind(&st.connection).unwrap_or_default();
2959                self.ui.warn(&format!(
2960                    "  not checkable: the `{kind}` source plugin (connection `{}`) can't check statements without running them",
2961                    st.connection
2962                ));
2963                uncheckable.insert(st.connection.clone());
2964                continue;
2965            }
2966            let verb = if st.kind == StatementKind::TempCreate {
2967                ""
2968            } else {
2969                " (check)"
2970            };
2971            sql_log(
2972                &format!("{}:{}{verb}", st.file.display(), st.line),
2973                &protected_sql(&st.text, st.sensitive),
2974            );
2975            if st.kind == StatementKind::TempCreate {
2976                let result = {
2977                    let (process, _log_scope) = s.statement(st.sensitive)?;
2978                    process.execute(&st.text, None, |_, _| Ok(()))
2979                };
2980                if let Err(e) = result {
2981                    failures.push(format!(
2982                        "{}:{}: {}",
2983                        st.file.display(),
2984                        st.line,
2985                        masked_source_error(&e, st.sensitive)
2986                    ));
2987                }
2988                continue;
2989            }
2990            let result = {
2991                let (process, _log_scope) = s.statement(st.sensitive)?;
2992                process.check(&st.text)
2993            };
2994            if let Err(e) = result {
2995                let mut msg = format!(
2996                    "{}:{}: {}",
2997                    st.file.display(),
2998                    st.line,
2999                    masked_source_error(&e, st.sensitive)
3000                );
3001                if let Some((f, l)) = unexecuted_setup.get(&st.connection) {
3002                    msg.push_str(&format!(
3003                        " (may be a false positive: the setup statement at {}:{l} wasn't executed during the check)",
3004                        f.display()
3005                    ));
3006                }
3007                failures.push(msg);
3008            }
3009            if st.kind == StatementKind::Other {
3010                unexecuted_setup.insert(st.connection.clone(), (st.file.clone(), st.line));
3011            }
3012        }
3013        self.check_source_columns(&pool, &mut failures);
3014        if failures.is_empty() {
3015            Ok(())
3016        } else {
3017            Err(Fail::new(Code::QueryFailed, failures.join("\n    ")))
3018        }
3019    }
3020
3021    /// Declared source `columns` against the real table, once per source table, connection and
3022    /// target in this process: each must exist, and match its `data_type` (loosely) where given.
3023    fn check_source_columns(&mut self, pool: &Arc<Pool>, failures: &mut Vec<String>) {
3024        /// `(project, source.table, connection, target)`.
3025        type Checked = std::collections::BTreeSet<(PathBuf, String, String, String)>;
3026        static CHECKED: std::sync::LazyLock<Mutex<Checked>> = std::sync::LazyLock::new(Mutex::default);
3027        for q in self.parsed.queries.clone() {
3028            let Some(conn) = q.connection.clone() else {
3029                continue;
3030            };
3031            for key in &q.sources {
3032                let Some(r) = self.parsed.sources.get(key).cloned() else {
3033                    continue;
3034                };
3035                let Some(declared) = self
3036                    .project
3037                    .sources
3038                    .get(&r.source)
3039                    .and_then(|s| s.table(&r.table))
3040                    .map(|t| t.columns.clone())
3041                    .filter(|c| !c.is_empty())
3042                else {
3043                    continue;
3044                };
3045                let id = (
3046                    self.project.root.clone(),
3047                    key.clone(),
3048                    conn.clone(),
3049                    self.project.profiles.target_of(Role::Connection, &conn),
3050                );
3051                if !CHECKED.lock().unwrap().insert(id) {
3052                    continue;
3053                }
3054                let quote = if r.quoting.any() {
3055                    let kind = pool.kind(&conn).unwrap_or_default();
3056                    match pool.connections.identifier_quote(&kind) {
3057                        Ok(q) => q,
3058                        Err(e) => {
3059                            failures.push(format!("source `{key}`: can't quote its name: {e}"));
3060                            continue;
3061                        }
3062                    }
3063                } else {
3064                    None
3065                };
3066                let rel = r.relation(quote.as_deref());
3067                let runner = PoolRunner {
3068                    pool: pool.clone(),
3069                    default: Some(conn.clone()),
3070                    log: self.ui.sql_log(),
3071                };
3072                let actual = match runner.columns(&format!("select * from {rel} where 1=0"), None) {
3073                    Ok(c) => c,
3074                    Err(e) => {
3075                        failures.push(format!("source `{key}` ({rel} on connection `{conn}`): {e}"));
3076                        continue;
3077                    }
3078                };
3079                for c in &declared {
3080                    let Some(a) = actual.iter().find(|a| a.name.eq_ignore_ascii_case(&c.name)) else {
3081                        failures.push(format!(
3082                            "source `{key}`: declared column `{}` isn't in {rel} on connection `{conn}` (it has: {})",
3083                            c.name,
3084                            actual.iter().map(|a| a.name.as_str()).collect::<Vec<_>>().join(", ")
3085                        ));
3086                        continue;
3087                    };
3088                    let Some(dt) = &c.data_type else { continue };
3089                    match crate::coltypes::matches(dt, &a.data_type) {
3090                        Some(true) => {}
3091                        Some(false) => failures.push(format!(
3092                            "source `{key}`: column `{}` is declared `{dt}`, but {rel} on connection `{conn}` returns {}",
3093                            c.name, a.data_type
3094                        )),
3095                        None => self.ui.warn(&format!(
3096                            "  source `{key}`: column `{}`'s `data_type: {dt}` isn't a type DRE can compare, so only its presence was checked",
3097                            c.name
3098                        )),
3099                    }
3100                }
3101            }
3102        }
3103    }
3104}
3105
3106/// The source session, started on first use (rendering may need it for `run_query()`).
3107struct Session {
3108    path: Result<crate::plugins::Located, String>,
3109    connection: Result<JsonMap<String, Json>, String>,
3110    cwd: PathBuf,
3111    log: LogSink,
3112    /// Whether plugin diagnostics for the active statement must be hidden rather than exactly
3113    /// masked. Connection diagnostics always use exact-value masking.
3114    opaque_log: Arc<AtomicBool>,
3115    /// Unmanaged reports run read-only unless they create temp objects.
3116    unmanaged: bool,
3117    read_only: bool,
3118    /// A lookup was loaded into a temp table, so the session must stay open as it is.
3119    loaded: bool,
3120    proc_: Option<PluginProcess>,
3121}
3122
3123impl Session {
3124    fn new(
3125        path: Result<crate::plugins::Located, String>,
3126        connection: Result<JsonMap<String, Json>, String>,
3127        cwd: PathBuf,
3128        log: LogSink,
3129        unmanaged: bool,
3130    ) -> Session {
3131        let opaque_log = Arc::new(AtomicBool::new(false));
3132        let protect = Arc::clone(&opaque_log);
3133        let downstream = log;
3134        let log: LogSink = Arc::new(move |plugin, line| {
3135            let line = crate::secrets::mask_source_error(line, protect.load(Ordering::SeqCst));
3136            downstream(plugin, &line);
3137        });
3138        Session {
3139            path,
3140            connection,
3141            cwd,
3142            log,
3143            opaque_log,
3144            unmanaged,
3145            read_only: unmanaged,
3146            loaded: false,
3147            proc_: None,
3148        }
3149    }
3150
3151    fn want_read_only(&mut self, ro: bool) {
3152        if self.read_only != ro {
3153            // Reopen if a session was already started in the other mode.
3154            self.close();
3155            self.read_only = ro;
3156        }
3157    }
3158
3159    fn get(&mut self) -> Result<&mut PluginProcess, String> {
3160        if self.proc_.is_none() {
3161            let plugin = self.path.clone()?;
3162            let connection = self.connection.clone()?;
3163            let mut p = plugin
3164                .start(self.log.clone(), Some(&self.cwd))
3165                .map_err(|e| masked_source_error(&e, false))?;
3166            let ro = self.unmanaged && self.read_only && p.has(CAP_READ_ONLY);
3167            p.open(connection, ro).map_err(|e| {
3168                let protected = masked_source_error(&e, false);
3169                format!("can't open the source connection: {protected}")
3170            })?;
3171            self.proc_ = Some(p);
3172        }
3173        Ok(self.proc_.as_mut().unwrap())
3174    }
3175
3176    fn statement(&mut self, sensitive: bool) -> Result<(&mut PluginProcess, OpaqueLogScope), String> {
3177        let opaque_log = Arc::clone(&self.opaque_log);
3178        let process = self.get()?;
3179        let scope = OpaqueLogScope::new(opaque_log, sensitive);
3180        Ok((process, scope))
3181    }
3182
3183    fn close(&mut self) {
3184        if let Some(p) = self.proc_.take() {
3185            let _ = p.close();
3186        }
3187    }
3188}
3189
3190struct OpaqueLogScope(Arc<AtomicBool>);
3191
3192impl OpaqueLogScope {
3193    fn new(active: Arc<AtomicBool>, sensitive: bool) -> Self {
3194        active.store(sensitive, Ordering::SeqCst);
3195        Self(active)
3196    }
3197}
3198
3199impl Drop for OpaqueLogScope {
3200    fn drop(&mut self) {
3201        self.0.store(false, Ordering::SeqCst);
3202    }
3203}
3204
3205/// What a run logs, and records, for a destination whose entry is `deliver: false`.
3206fn nowhere_note(profile: &str, target: &str) -> String {
3207    format!(
3208        "destination `{profile}`: `{target}` delivers nowhere (`deliver: false`); output stays in target/"
3209    )
3210}
3211
3212/// The Binding's connections: each one's session, opened when a query first needs it and held
3213/// until the report ends.
3214struct Pool {
3215    connections: Arc<ProfileConnections>,
3216    cwd: PathBuf,
3217    log: LogSink,
3218    unmanaged: bool,
3219    sessions: Mutex<BTreeMap<String, Arc<Mutex<Session>>>>,
3220}
3221
3222impl Pool {
3223    fn new(connections: Arc<ProfileConnections>, cwd: PathBuf, log: LogSink, unmanaged: bool) -> Pool {
3224        Pool {
3225            connections,
3226            cwd,
3227            log,
3228            unmanaged,
3229            sessions: Mutex::default(),
3230        }
3231    }
3232
3233    /// A connection's settings for this run.
3234    fn target(&self, name: &str) -> Result<ProfileTarget, String> {
3235        let profiles = &self.connections.profiles;
3236        match profiles.entry(Role::Connection, name) {
3237            Entry::Use(o) => Ok(o.clone()),
3238            Entry::Unknown => Err(format!(
3239                "connection `{name}` isn't under `{}:` in {}",
3240                profiles.section_key(Role::Connection),
3241                profiles.path.display()
3242            )),
3243            // A connection never has `deliver: false`.
3244            Entry::Missing | Entry::Nowhere => Err(profiles.missing_entry(Role::Connection, name)),
3245        }
3246    }
3247
3248    /// A connection's plugin type.
3249    fn kind(&self, name: &str) -> Result<String, String> {
3250        Ok(self.target(name)?.kind)
3251    }
3252
3253    fn session(&self, name: &str) -> Result<Arc<Mutex<Session>>, String> {
3254        if let Some(s) = self.sessions.lock().unwrap().get(name) {
3255            return Ok(s.clone());
3256        }
3257        let output = self.target(name)?;
3258        let id = crate::project::PluginId::new(PluginKind::Source, output.kind.clone());
3259        let path = crate::plugins::locate_in(&self.connections.root, &self.connections.plugins, &id)
3260            .map_err(String::from);
3261        let session = Arc::new(Mutex::new(Session::new(
3262            path,
3263            render_connection(&output),
3264            self.cwd.clone(),
3265            self.log.clone(),
3266            self.unmanaged,
3267        )));
3268        self.sessions
3269            .lock()
3270            .unwrap()
3271            .insert(name.to_string(), session.clone());
3272        Ok(session)
3273    }
3274
3275    fn close_all(&self) {
3276        for s in self.sessions.lock().unwrap().values() {
3277            if let Ok(mut s) = s.lock() {
3278                s.close();
3279            }
3280        }
3281    }
3282}
3283
3284/// `run_query()`, `columns()` and lookups for one renderer: on the named connection's session,
3285/// else on the renderer's own (`default`).
3286struct PoolRunner {
3287    pool: Arc<Pool>,
3288    default: Option<String>,
3289    log: LogSink,
3290}
3291
3292impl PoolRunner {
3293    fn session(&self, profile: Option<&str>) -> Result<Arc<Mutex<Session>>, String> {
3294        let name = profile.or(self.default.as_deref()).ok_or(
3295            "no connection here: pass `profile=`, or give the report a `profile:` (this isn't a query's SQL)",
3296        )?;
3297        self.pool.session(name)
3298    }
3299}
3300
3301impl QueryRunner for PoolRunner {
3302    fn run_query(&self, sql: &str, max_rows: u64, profile: Option<&str>) -> Result<QueryRows, String> {
3303        let session = self.session(profile)?;
3304        let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
3305        let sensitive = crate::secrets::contains_secret(sql);
3306        // An unmanaged report may only read, and run_query() runs before the file's own
3307        // statements are checked, so it gets the same rule up front.
3308        if s.unmanaged
3309            && let Some(bad) = sqlsplit::split(sql)
3310                .into_iter()
3311                .find(|st| sqlsplit::classify(&st.text) != StatementKind::Read)
3312        {
3313            return Err(format!(
3314                "run_query() in an unmanaged report may only read, but got `{}`; give the report a YAML to declare it",
3315                masked_sql_summary(&bad.text, 60, sensitive)
3316            ));
3317        }
3318        (self.log)("run_query()", &protected_sql(sql, sensitive));
3319        let (p, _log_scope) = s.statement(sensitive)?;
3320        let mut out = QueryRows::default();
3321        let mut too_many = false;
3322        let exec = p
3323            .execute(sql, Some(max_rows + 1), |schema, batch| {
3324                if out.columns.is_empty() {
3325                    out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
3326                }
3327                if out.rows.len() as u64 + batch.num_rows() as u64 > max_rows {
3328                    too_many = true;
3329                    return Ok(());
3330                }
3331                out.rows.extend(crate::values::batch_rows(&batch));
3332                Ok(())
3333            })
3334            .map_err(|e| format!("run_query() failed: {}", masked_source_error(&e, sensitive)))?;
3335        if too_many {
3336            return Err(format!(
3337                "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"
3338            ));
3339        }
3340        if let Execution::Result { schema, .. } = exec
3341            && out.columns.is_empty()
3342        {
3343            out.columns = schema.fields().iter().map(|f| f.name().clone()).collect();
3344        }
3345        Ok(out)
3346    }
3347
3348    fn columns(&self, sql: &str, profile: Option<&str>) -> Result<Vec<Column>, String> {
3349        let session = self.session(profile)?;
3350        let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
3351        let sensitive = crate::secrets::contains_secret(sql);
3352        (self.log)("columns()", &protected_sql(sql, sensitive));
3353        let (p, _log_scope) = s.statement(sensitive)?;
3354        let mut schema: Option<SchemaRef> = None;
3355        let exec = p
3356            .execute(sql, Some(1), |sch, _| {
3357                schema.get_or_insert_with(|| sch.clone());
3358                Ok(())
3359            })
3360            .map_err(|e| masked_source_error(&e, sensitive))?;
3361        if let Execution::Result { schema: sch, .. } = exec {
3362            schema.get_or_insert(sch);
3363        }
3364        let schema = schema.ok_or("the query returned no result set")?;
3365        Ok(schema
3366            .fields()
3367            .iter()
3368            .map(|f| Column {
3369                name: f.name().clone(),
3370                data_type: f.data_type().to_string(),
3371            })
3372            .collect())
3373    }
3374
3375    fn load(&self, name: &str, table: &Table) -> Result<Option<(String, Option<String>)>, String> {
3376        let session = self.session(None)?;
3377        let mut s = session.lock().map_err(|_| "session lock poisoned".to_string())?;
3378        // A temp table is allowed even in an unmanaged report; it needs a writable session.
3379        s.want_read_only(false);
3380        if !s.get()?.has(CAP_LOAD) {
3381            return Ok(None);
3382        }
3383        let batch = table.to_batch()?;
3384        let schema = batch.schema();
3385        (self.log)(
3386            &format!("lookup `{name}`"),
3387            &format!(
3388                "-- {} rows loaded through the plugin's `load` request",
3389                batch.num_rows()
3390            ),
3391        );
3392        let r = s.get()?.load(name, &schema, [batch]).map_err(|e| e.to_string())?;
3393        s.loaded = true;
3394        Ok(Some(r))
3395    }
3396}
3397
3398/// Render every string inside a destination option value.
3399fn render_json(renderer: &Renderer, file: &Path, v: &Json) -> Result<Json, RenderError> {
3400    Ok(match v {
3401        Json::String(s) => Json::String(renderer.render(file, s)?),
3402        Json::Array(a) => Json::Array(
3403            a.iter()
3404                .map(|v| render_json(renderer, file, v))
3405                .collect::<Result<_, _>>()?,
3406        ),
3407        Json::Object(o) => Json::Object(
3408            o.iter()
3409                .map(|(k, v)| Ok((k.clone(), render_json(renderer, file, v)?)))
3410                .collect::<Result<_, RenderError>>()?,
3411        ),
3412        other => other.clone(),
3413    })
3414}
3415
3416/// Render `env_var()` (and only that) inside a profile output's string fields.
3417pub(crate) fn render_connection(output: &ProfileTarget) -> Result<JsonMap<String, Json>, String> {
3418    let mut env = minijinja::Environment::new();
3419    env.set_undefined_behavior(minijinja::UndefinedBehavior::Strict);
3420    env.add_function("env_var", |name: String, default: Option<String>| -> Result<String, minijinja::Error> {
3421        std::env::var(&name).ok().or(default).ok_or_else(|| {
3422            minijinja::Error::new(
3423                minijinja::ErrorKind::UndefinedError,
3424                format!("`env_var('{name}')`: environment variable `{name}` is not set and no default is given"),
3425            )
3426        })
3427    });
3428    fn walk(env: &minijinja::Environment<'_>, v: &Json) -> Result<Json, String> {
3429        Ok(match v {
3430            Json::String(s) if crate::preflight::is_templated(s) => Json::String(
3431                env.render_str(s, ())
3432                    .map_err(|e| format!("profiles.yml: {}", e.detail().unwrap_or("render error")))?,
3433            ),
3434            Json::Array(a) => Json::Array(a.iter().map(|x| walk(env, x)).collect::<Result<_, _>>()?),
3435            Json::Object(o) => Json::Object(
3436                o.iter()
3437                    .map(|(k, x)| Ok((k.clone(), walk(env, x)?)))
3438                    .collect::<Result<_, String>>()?,
3439            ),
3440            other => other.clone(),
3441        })
3442    }
3443    match walk(&env, &Json::Object(output.fields.clone()))? {
3444        Json::Object(o) => Ok(o),
3445        _ => unreachable!(),
3446    }
3447}
3448
3449/// Locate the plugin for a declared type among the project's packages, honouring `dre.lock`
3450/// pins and declared constraints.
3451pub fn find_plugin(
3452    project: &Project,
3453    kind: PluginKind,
3454    name: &str,
3455) -> Result<crate::plugins::Located, crate::plugins::LocateError> {
3456    crate::plugins::locate(project, &crate::project::PluginId::new(kind, name))
3457}
3458
3459/// `connection`, `destination` and `profile()` from `profiles.yml`, for the run's target. A
3460/// field is secret when the plugin's `describe` says so; when the plugin can't be asked, when its
3461/// name looks like one; and always when its value comes from a `DRE_SECRET_*` variable.
3462struct ProfileConnections {
3463    /// Each profile's entry for the run comes from here.
3464    profiles: Profiles,
3465    root: PathBuf,
3466    plugins: Vec<crate::project::PluginRequirement>,
3467    log: LogSink,
3468}
3469
3470/// What `describe` said, by `(project, role, type)`, `None` when it couldn't be asked: each
3471/// plugin is asked once per process, not once per Binding.
3472static DESCRIBED: std::sync::LazyLock<Mutex<BTreeMap<DescribeKey, Option<Described>>>> =
3473    std::sync::LazyLock::new(Mutex::default);
3474
3475/// `(project root, role, plugin type)`.
3476type DescribeKey = (PathBuf, Role, String);
3477
3478#[derive(Clone)]
3479struct Described {
3480    secrets: Vec<String>,
3481    identifier_quote: Option<String>,
3482    capabilities: Vec<String>,
3483    /// A message destination's length limit.
3484    message_limit: Option<u64>,
3485}
3486
3487const SECRET_NAME_WORDS: &[&str] = &["password", "secret", "token", "key", "credential"];
3488
3489impl ProfileConnections {
3490    fn view(&self, role: Role, name: &str) -> Result<Connection, String> {
3491        let target = self.profiles.target_of(role, name);
3492        let out: ProfileTarget = match self.profiles.entry(role, name) {
3493            Entry::Use(o) => o.clone(),
3494            Entry::Nowhere => {
3495                return Err(format!(
3496                    "destination `{name}`: `{target}` delivers nowhere (`deliver: false`), so it has no settings"
3497                ));
3498            }
3499            Entry::Missing => return Err(self.profiles.missing_entry(role, name)),
3500            Entry::Unknown => {
3501                return Err(format!(
3502                    "no {} profile `{name}` in {}",
3503                    role.as_str(),
3504                    self.profiles.path.display()
3505                ));
3506            }
3507        };
3508        let mut secrets: Vec<String> = out
3509            .fields
3510            .iter()
3511            .filter(|(_, v)| v.as_str().is_some_and(|s| s.contains(crate::secrets::PREFIX)))
3512            .map(|(k, _)| k.clone())
3513            .collect();
3514        match self.described(role, &out.kind) {
3515            Some(d) => secrets.extend(d.secrets),
3516            None => secrets.extend(
3517                out.fields
3518                    .keys()
3519                    .filter(|k| {
3520                        let k = k.to_lowercase();
3521                        SECRET_NAME_WORDS.iter().any(|w| k.contains(w))
3522                    })
3523                    .cloned(),
3524            ),
3525        }
3526        let fields = render_connection(&out)?;
3527        Ok(Connection {
3528            profile: name.to_string(),
3529            target,
3530            kind: out.kind.clone(),
3531            fields,
3532            secrets,
3533        })
3534    }
3535
3536    /// The plugin's `describe` reply, asked once per plugin type.
3537    fn described(&self, role: Role, kind: &str) -> Option<Described> {
3538        let key = (self.root.clone(), role, kind.to_string());
3539        if let Some(v) = DESCRIBED.lock().unwrap().get(&key) {
3540            return v.clone();
3541        }
3542        let plugin_kind = match role {
3543            Role::Connection => PluginKind::Source,
3544            Role::Destination => PluginKind::Destination,
3545        };
3546        let asked = (|| {
3547            if kind == LOCAL_TYPE {
3548                return Some(Described {
3549                    secrets: Vec::new(),
3550                    identifier_quote: None,
3551                    capabilities: Vec::new(),
3552                    message_limit: None,
3553                });
3554            }
3555            let id = crate::project::PluginId::new(plugin_kind, kind);
3556            let plugin = crate::plugins::locate_in(&self.root, &self.plugins, &id).ok()?;
3557            let mut p = plugin.start(self.log.clone(), Some(&self.root)).ok()?;
3558            let capabilities = p.info().capabilities.clone();
3559            let d = p.description().ok();
3560            let _ = p.close();
3561            let d = d?;
3562            Some(Described {
3563                secrets: d
3564                    .connection_fields
3565                    .into_iter()
3566                    .filter(|f| f.secret)
3567                    .map(|f| f.name)
3568                    .collect(),
3569                identifier_quote: d.identifier_quote,
3570                capabilities,
3571                message_limit: d.message_limit,
3572            })
3573        })();
3574        DESCRIBED.lock().unwrap().insert(key, asked.clone());
3575        asked
3576    }
3577}
3578
3579impl Connections for ProfileConnections {
3580    fn profile(&self, name: &str, role: Option<&str>) -> Result<Connection, String> {
3581        let role = match role {
3582            Some("connection") => Role::Connection,
3583            Some(_) => Role::Destination,
3584            None => {
3585                let c = self.profiles.get(Role::Connection, name).is_some();
3586                let d = self.profiles.get(Role::Destination, name).is_some();
3587                match (c, d) {
3588                    (true, true) => {
3589                        return Err(format!(
3590                            "`{name}` is both a connection and a destination profile; say which with `profile('{name}', role='connection')` or `role='destination'`"
3591                        ));
3592                    }
3593                    (true, false) => Role::Connection,
3594                    _ => Role::Destination,
3595                }
3596            }
3597        };
3598        self.view(role, name)
3599    }
3600
3601    fn identifier_quote(&self, kind: &str) -> Result<Option<String>, String> {
3602        match self.described(Role::Connection, kind) {
3603            Some(d) => Ok(d.identifier_quote),
3604            None => Err(format!(
3605                "the `{kind}` source plugin couldn't be asked for its identifier quote character"
3606            )),
3607        }
3608    }
3609}
3610
3611fn check_sheet_name(n: &str) -> Result<(), String> {
3612    if n.is_empty() || n.chars().count() > 31 {
3613        return Err(format!("sheet name `{n}` must be 1 to 31 characters long"));
3614    }
3615    if let Some(c) = n.chars().find(|c| "[]:*?/\\".contains(*c)) {
3616        return Err(format!(
3617            "sheet name `{n}` contains `{c}`, which Excel doesn't allow in sheet names"
3618        ));
3619    }
3620    if n.starts_with('\'') || n.ends_with('\'') {
3621        return Err(format!("sheet name `{n}` can't start or end with an apostrophe"));
3622    }
3623    if n.eq_ignore_ascii_case("history") {
3624        return Err("`History` is reserved by Excel and can't be a sheet name".into());
3625    }
3626    Ok(())
3627}
3628
3629fn is_single_table(format: &str) -> bool {
3630    matches!(format, "csv" | "delimited" | "fixed_width" | "parquet")
3631}
3632
3633fn extension(format: &str) -> &str {
3634    match format {
3635        "delimited" | "fixed_width" => "txt",
3636        crate::project::MESSAGE_FORMAT => "md",
3637        f => f,
3638    }
3639}
3640
3641fn split_ext(name: &str) -> (&str, &str) {
3642    match name.rfind('.') {
3643        Some(i) if i > 0 => (&name[..i], &name[i..]),
3644        _ => (name, ""),
3645    }
3646}
3647
3648fn rel(root: &Path, p: &Path) -> PathBuf {
3649    crate::slash(p.strip_prefix(root).unwrap_or(p))
3650}
3651
3652/// An instant as DRE records it: RFC 3339 in UTC, to the second.
3653pub fn rfc3339(t: chrono::DateTime<chrono::Utc>) -> String {
3654    t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
3655}
3656
3657/// How a generated file is recorded in `run_results.json`: relative to the project root when
3658/// it's inside it, else relative to the target path, so moving the target folder doesn't
3659/// change what's recorded.
3660fn record_path(project: &Project, p: &Path) -> PathBuf {
3661    match p.strip_prefix(&project.root) {
3662        Ok(r) => crate::slash(r),
3663        Err(_) => rel(&project.target_dir, p),
3664    }
3665}
3666
3667impl BindingRun<'_> {
3668    /// What this Binding produced, for its end-of-run line.
3669    fn summary_line(&self) -> String {
3670        let rows: u64 = self.produced.iter().map(|p| p.rows).sum();
3671        let mut parts = Vec::new();
3672        if !self.produced.is_empty() {
3673            let n = self.produced.len();
3674            parts.push(format!(
3675                "{n} result set{}, {} row{}",
3676                if n == 1 { "" } else { "s" },
3677                thousands(rows),
3678                if rows == 1 { "" } else { "s" }
3679            ));
3680        }
3681        let names: Vec<String> = self
3682            .files()
3683            .map(|(f, _)| f.file_name().unwrap_or_default().to_string_lossy().to_string())
3684            .collect();
3685        if !names.is_empty() {
3686            parts.push(format!("→ {}", names.join(", ")));
3687        }
3688        // Every destination's location, not just the first one each file reached.
3689        let delivered: Vec<&str> = self
3690            .outs
3691            .iter()
3692            .flat_map(|o| o.deliveries.iter())
3693            .filter_map(|d| d.location.as_deref())
3694            .collect();
3695        let notes: Vec<&str> = self
3696            .outs
3697            .iter()
3698            .filter(|o| !o.dests.is_empty())
3699            .filter_map(|o| o.delivery_note.as_deref())
3700            .collect();
3701        if !delivered.is_empty() {
3702            parts.push(format!("→ {}", delivered.join(", ")));
3703        } else if !notes.is_empty() && self.opts.preview.is_none() {
3704            parts.push(format!("({})", notes.join("; ")));
3705        }
3706        parts.join(" ")
3707    }
3708
3709    /// Every file the Binding's outputs wrote, and where each was first delivered.
3710    fn files(&self) -> impl Iterator<Item = &(PathBuf, Option<String>)> {
3711        self.outs.iter().flat_map(|o| o.files.iter())
3712    }
3713}
3714
3715/// `1234567` → `1,234,567`.
3716pub fn thousands(n: u64) -> String {
3717    let s = n.to_string();
3718    let mut out = String::new();
3719    for (i, c) in s.chars().enumerate() {
3720        if i > 0 && (s.len() - i).is_multiple_of(3) {
3721            out.push(',');
3722        }
3723        out.push(c);
3724    }
3725    out
3726}
3727
3728fn human_bytes(n: u64) -> String {
3729    match n {
3730        n if n < 1024 => format!("{n} B"),
3731        n if n < 1024 * 1024 => format!("{:.1} KB", n as f64 / 1024.0),
3732        n => format!("{:.1} MB", n as f64 / (1024.0 * 1024.0)),
3733    }
3734}
3735
3736/// `dre clean`: remove the target folder `t`.
3737pub fn clean(t: &Path) -> std::io::Result<bool> {
3738    if t.exists() {
3739        std::fs::remove_dir_all(t)?;
3740        Ok(true)
3741    } else {
3742        Ok(false)
3743    }
3744}