dre_protocol/
msg.rs

1//! Control messages. Every JSON frame is an object with a `type` field. From protocol 1, every
2//! message after the handshake also carries the `id` of the request it belongs to ([`Envelope`]).
3
4use serde::{Deserialize, Serialize};
5use serde_json::{Map, Value};
6
7pub use crate::options::OptionField;
8use crate::{Kind, PluginId};
9
10/// A message with the id of the request it belongs to. Core numbers its requests; every reply,
11/// and every message streamed as part of a request (`result_set_end`, `finish`, `result`,
12/// `result_end`, `log`, `progress`), carries that request's id. `hello` has none, and neither
13/// does anything a protocol 0 peer sends.
14#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
15// In the schema, `Request` and `Response` are each message with its envelope.
16#[schemars(rename = "{T}")]
17pub struct Envelope<T> {
18    #[serde(default, skip_serializing_if = "Option::is_none")]
19    pub id: Option<u64>,
20    #[serde(flatten)]
21    pub body: T,
22}
23
24impl<T> Envelope<T> {
25    pub fn new(id: Option<u64>, body: T) -> Envelope<T> {
26        Envelope { id, body }
27    }
28}
29
30/// Core → plugin.
31#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
32#[serde(tag = "type", rename_all = "snake_case")]
33pub enum Request {
34    /// Always first. The plugin picks a version in both ranges or replies `version_mismatch`.
35    Hello {
36        min_version: u32,
37        max_version: u32,
38        core_version: String,
39        /// The plugin core wants served, for an executable that provides several. A package
40        /// that doesn't provide it replies `error`; without it, the package serves its first.
41        #[serde(default, skip_serializing_if = "Option::is_none")]
42        plugin: Option<PluginId>,
43    },
44    /// Describe the connection fields a profile output for this plugin takes (used by `dre init`)
45    /// and the options a report's config block for it takes.
46    Describe {},
47    /// Check one config block of options (a format's `output:` keys, a destination entry's keys)
48    /// without doing anything. Replies `validated`.
49    Validate {
50        options: Map<String, Value>,
51        /// Format: what the plugin needs to check the output's template too. For the html and
52        /// pdf formats, `{"file": <absolute path or null>, "project_root", "queries": [...]}`.
53        /// Optional; older plugins ignore it.
54        #[serde(default, skip_serializing_if = "Option::is_none")]
55        template: Option<Value>,
56    },
57    /// Source or destination (protocol 1): check a profile entry's connection settings without
58    /// connecting: the declared fields' generic checks, then the plugin's own rules. Keys in
59    /// `unresolved` (an unset `env_var()`) arrive as `null` and count as set, unchecked. Replies
60    /// `validated`, with `warnings` for unknown keys.
61    ValidateConnection {
62        connection: Map<String, Value>,
63        #[serde(default, skip_serializing_if = "Vec::is_empty")]
64        unresolved: Vec<String>,
65    },
66    /// Source: open a session. All later `execute`/`check` requests run on it.
67    Open {
68        connection: Map<String, Value>,
69        read_only: bool,
70        /// Core's `dre query`: the session must not reach files, programs or the network
71        /// beyond the database (protocol 1, optional; older plugins ignore it).
72        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
73        sandbox: bool,
74    },
75    /// Source: run one statement. Replies `result` (then Arrow frames, then `result_end`) or
76    /// `no_result`.
77    Execute {
78        sql: String,
79        #[serde(default, skip_serializing_if = "Option::is_none")]
80        row_limit: Option<u64>,
81    },
82    /// Source: verify one statement without executing it. Replies `ok` or `error`.
83    Check {
84        sql: String,
85    },
86    /// Source advertising `catalog`: discover accessible relations through native metadata.
87    Catalog {
88        #[serde(default)]
89        tables: Vec<String>,
90        #[serde(default)]
91        schemas: Vec<String>,
92    },
93    /// Format: write result sets to `path`. Followed, per result set, by Arrow frames (at least one,
94    /// carrying the schema) and a `result_set_end`, then `finish`. Replies `written`.
95    Write {
96        path: String,
97        format: String,
98        options: Map<String, Value>,
99        result_sets: Vec<ResultSetMeta>,
100        /// Core can answer `need_fonts` during this write. Absent on older cores.
101        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
102        font_requests: bool,
103        #[serde(default, skip_serializing_if = "Option::is_none")]
104        template: Option<Value>,
105    },
106    /// Reply to a font request within a write; retains the write request id. `repeated` is set
107    /// when core already gave this font's `error` earlier in the run, so the person has been
108    /// told once and the format needn't warn again.
109    Fonts {
110        paths: Vec<String>,
111        error: Option<String>,
112        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
113        repeated: bool,
114    },
115    ResultSetEnd {},
116    Finish {},
117    /// Destination: deliver a local file, or with `files` (only to a plugin advertising
118    /// `multi_file`) every file of one output at once. Exactly one of `local_path`/`files` is set,
119    /// unless `message` is: then `files` (possibly empty) holds the files to attach to it, and is
120    /// sent only to a plugin advertising `message`. `options` are the destination entry's plugin
121    /// options, rendered by core. Replies `delivered`.
122    Deliver {
123        #[serde(default, skip_serializing_if = "Option::is_none")]
124        local_path: Option<String>,
125        #[serde(default, skip_serializing_if = "Option::is_none")]
126        remote_path: Option<String>,
127        #[serde(default, skip_serializing_if = "Vec::is_empty")]
128        files: Vec<DeliveryFile>,
129        connection: Map<String, Value>,
130        #[serde(default)]
131        options: Map<String, Value>,
132        #[serde(default, skip_serializing_if = "Option::is_none")]
133        message: Option<Message>,
134    },
135    /// Source: load rows into a temporary table on the session, named after `name`. Followed by
136    /// Arrow frames (at least one, carrying the schema) and a `result_set_end`. Replies `loaded`.
137    /// Only sent to plugins advertising `load`.
138    Load {
139        name: String,
140    },
141    /// End the conversation; the plugin exits 0.
142    Close {},
143    /// Protocol 1: stop the request whose id the envelope carries. It has no reply of its own:
144    /// the request replies `error` with kind `cancelled`, or its usual reply if it finished
145    /// first. A cancel for a request that isn't running is ignored.
146    Cancel {},
147}
148
149/// Plugin → core.
150#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
151#[serde(tag = "type", rename_all = "snake_case")]
152pub enum Response {
153    Hello {
154        protocol_version: u32,
155        kind: Kind,
156        name: String,
157        version: String,
158        #[serde(default)]
159        capabilities: Vec<String>,
160        /// Every plugin the executable serves; empty for one that serves only `kind`/`name`.
161        #[serde(default, skip_serializing_if = "Vec::is_empty")]
162        provides: Vec<PluginId>,
163    },
164    VersionMismatch {
165        min_version: u32,
166        max_version: u32,
167    },
168    Describe {
169        connection_fields: Vec<ConnectionField>,
170        #[serde(default, skip_serializing_if = "Vec::is_empty")]
171        option_fields: Vec<OptionField>,
172        /// Source: the character the database quotes identifiers with (`"` for DuckDB and
173        /// Postgres, a backtick for Databricks). Core uses it to quote names a project asks to
174        /// quote (a source's `quoting:`). Every source sets it; other kinds leave it out.
175        #[serde(default, skip_serializing_if = "Option::is_none")]
176        identifier_quote: Option<String>,
177        /// Source: how core spells a string holding `'` or `\` when it writes one into SQL (an
178        /// inlined lookup): absent or `chr` for `chr(39)`/`chr(92)`, `backslash` for `\'`/`\\`
179        /// inside the quotes (a database without `chr()`, such as ClickHouse).
180        #[serde(default, skip_serializing_if = "Option::is_none")]
181        string_literal: Option<String>,
182        /// Destination advertising `message`: the most characters a message may have in the
183        /// service, after translation. Core shows it next to a message's length in a preview.
184        #[serde(default, skip_serializing_if = "Option::is_none")]
185        message_limit: Option<u64>,
186        /// Source: a connection field naming a file only one process may write at a time
187        /// (DuckDB's `path`). Core runs one Binding at a time on a connection that sets it,
188        /// whatever the entry's `threads:`.
189        #[serde(default, skip_serializing_if = "Option::is_none")]
190        single_writer: Option<SingleWriter>,
191    },
192    /// Every problem with the options, each a sentence naming the key; empty when they're fine.
193    Validated {
194        errors: Vec<String>,
195        /// Problems that don't stop a run (an unknown key); `validate_connection` only.
196        #[serde(default, skip_serializing_if = "Vec::is_empty")]
197        warnings: Vec<String>,
198    },
199    Ok {},
200    /// A result set follows as Arrow frames, ended by `result_end`.
201    Result {
202        columns: Vec<String>,
203    },
204    ResultEnd {
205        rows: u64,
206    },
207    NoResult {
208        #[serde(default, skip_serializing_if = "Option::is_none")]
209        rows_affected: Option<u64>,
210    },
211    /// A source's catalogue, using the existing `dre/catalog/v1` document shape.
212    Catalog {
213        catalog: Value,
214    },
215    /// A format needs a pinned registry font before completing its write.
216    NeedFonts {
217        font: FontRequest,
218    },
219    Written {
220        files: Vec<String>,
221        /// Things the person should know about the files (a value written as text, say).
222        #[serde(default, skip_serializing_if = "Vec::is_empty")]
223        warnings: Vec<String>,
224    },
225    Delivered {
226        location: String,
227        /// How many tries the delivery took, when it was more than one (protocol 1).
228        #[serde(default, skip_serializing_if = "Option::is_none")]
229        attempts: Option<u32>,
230    },
231    /// `relation` is what SQL uses to read the loaded rows. `warning`, when set, is shown to the
232    /// user (e.g. the database has no bulk path, so a load this size is slow).
233    Loaded {
234        relation: String,
235        rows: u64,
236        #[serde(default, skip_serializing_if = "Option::is_none")]
237        warning: Option<String>,
238    },
239    /// The request failed. `kind` is one of [`ERROR_KINDS`]; `code` names the problem,
240    /// namespaced by the plugin (`sftp/host-key-mismatch`). Both are optional (protocol 0 has
241    /// neither): without a `code`, core reports the failure under its own code for the step;
242    /// a `code` without a `kind` counts as `internal`.
243    Error {
244        message: String,
245        #[serde(default, skip_serializing_if = "Option::is_none")]
246        kind: Option<String>,
247        #[serde(default, skip_serializing_if = "Option::is_none")]
248        code: Option<String>,
249    },
250    /// Protocol 1: a log line, while a request runs (with its id) or between requests.
251    Log {
252        level: LogLevel,
253        message: String,
254        /// Structured context (`{"attempt": 2}`); never secrets.
255        #[serde(default, skip_serializing_if = "Map::is_empty")]
256        fields: Map<String, Value>,
257    },
258    /// Protocol 1, optional: how far a long request has got. Core shows it when present.
259    Progress {
260        #[serde(default, skip_serializing_if = "Option::is_none")]
261        message: Option<String>,
262        #[serde(default, skip_serializing_if = "Option::is_none")]
263        done: Option<u64>,
264        #[serde(default, skip_serializing_if = "Option::is_none")]
265        total: Option<u64>,
266    },
267}
268
269impl Response {
270    /// An `error` reply with only a message.
271    pub fn error(message: impl Into<String>) -> Response {
272        Response::Error {
273            message: message.into(),
274            kind: None,
275            code: None,
276        }
277    }
278}
279
280/// The kinds an `error` reply may give, from DRE's error-code registry.
281pub const ERROR_KINDS: &[&str] = &[
282    "config",
283    "plugin",
284    "refused",
285    "connection",
286    "auth",
287    "query",
288    "delivery",
289    "internal",
290    "cancelled",
291    "timed_out",
292];
293
294/// How important a `log` message is. Core shows `info` and above to the person, and keeps
295/// `debug` and `trace` for `-v` and the log file.
296#[derive(
297    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, schemars::JsonSchema,
298)]
299#[serde(rename_all = "lowercase")]
300pub enum LogLevel {
301    Error,
302    Warn,
303    Info,
304    Debug,
305    Trace,
306}
307
308/// A rendered message, as `deliver` carries it.
309#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
310pub struct Message {
311    /// One line, plain text.
312    pub title: String,
313    /// The portable Markdown subset ([`crate::markdown`]).
314    pub text: String,
315    /// An HTML body, when one was made; core leaves it out today. A plugin that can show HTML
316    /// (email) uses it when present, else converts `text`.
317    #[serde(default, skip_serializing_if = "Option::is_none")]
318    pub html: Option<String>,
319    /// The `.md` file core wrote: `# title`, then the text.
320    pub path: String,
321}
322
323/// One file of a multi-file `deliver`.
324#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
325pub struct DeliveryFile {
326    pub local_path: String,
327    #[serde(default, skip_serializing_if = "Option::is_none")]
328    pub remote_path: Option<String>,
329}
330
331/// Per result set, what a format plugin needs to lay it out.
332#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
333pub struct ResultSetMeta {
334    /// Sheet name (xlsx) or file suffix (single-table formats).
335    pub name: String,
336    /// The query (`.sql` basename) that produced it.
337    pub query: String,
338    /// 1-based index among that query's result sets.
339    pub result_index: usize,
340    #[serde(default, skip_serializing_if = "Option::is_none")]
341    pub anchor: Option<String>,
342    #[serde(default, skip_serializing_if = "Option::is_none")]
343    pub header: Option<bool>,
344    /// The query entry's `columns:` map: per result column, how to show it.
345    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
346    pub columns: std::collections::BTreeMap<String, ColumnOptions>,
347    /// The query entry's `autofit` (xlsx): size this tab's columns from their content, over the
348    /// output's `autofit`.
349    #[serde(default, skip_serializing_if = "Option::is_none")]
350    pub autofit: Option<bool>,
351    /// The query entry's `style` (xlsx), over the output's.
352    #[serde(default, skip_serializing_if = "Option::is_none")]
353    pub style: Option<crate::style::SheetStyle>,
354}
355
356/// One entry of a `columns:` map.
357#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
358pub struct ColumnOptions {
359    /// An Excel number format code.
360    #[serde(default, skip_serializing_if = "Option::is_none")]
361    pub format: Option<String>,
362    /// A row formula (`={qty}*{price}`), written in place of the column's value, which becomes
363    /// its cached result.
364    #[serde(default, skip_serializing_if = "Option::is_none")]
365    pub formula: Option<String>,
366    /// A totals row entry under the data: `sum`, `average`, `count`, `min`, `max`, or a formula
367    /// over whole columns (`=SUM({amount:*})/COUNT({qty:*})`).
368    #[serde(default, skip_serializing_if = "Option::is_none")]
369    pub total: Option<String>,
370    /// The column's width (xlsx), over the tab's and output's `autofit`.
371    #[serde(default, skip_serializing_if = "Option::is_none")]
372    pub width: Option<ColumnWidth>,
373    /// How the column's data cells look (xlsx), over the tab's and output's `style`.
374    #[serde(default, skip_serializing_if = "Option::is_none")]
375    pub style: Option<crate::style::CellStyle>,
376}
377
378/// A column width: sized from the content, or a number of characters.
379#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
380#[serde(untagged)]
381pub enum ColumnWidth {
382    Chars(f64),
383    Auto(AutoWidth),
384}
385
386/// The word `auto`.
387#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
388#[serde(rename_all = "lowercase")]
389pub enum AutoWidth {
390    Auto,
391}
392
393/// A source's connection field naming a file only one process may write at a time.
394#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
395pub struct SingleWriter {
396    /// The connection field (`path`).
397    pub field: String,
398    /// Values of the field that name no file, so connections may share them (`:memory:`).
399    #[serde(default, skip_serializing_if = "Vec::is_empty")]
400    pub except: Vec<String>,
401}
402
403impl SingleWriter {
404    /// Whether `connection` sets the field to a file: a string other than the exceptions.
405    pub fn applies(&self, connection: &Map<String, Value>) -> bool {
406        connection
407            .get(&self.field)
408            .and_then(Value::as_str)
409            .is_some_and(|v| !self.except.iter().any(|e| e == v))
410    }
411}
412
413#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
414pub struct ConnectionField {
415    pub name: String,
416    #[serde(default)]
417    pub description: String,
418    #[serde(default)]
419    pub required: bool,
420    /// Offered as an `env_var()` reference by default in `dre init`.
421    #[serde(default)]
422    pub secret: bool,
423    #[serde(default, skip_serializing_if = "Option::is_none")]
424    pub default: Option<Value>,
425    /// A source type whose profile has the same field: `dre init` offers the value entered for
426    /// that source as this field's default (e.g. one Databricks host for source and destination).
427    #[serde(default, skip_serializing_if = "Option::is_none")]
428    pub same_as_source: Option<String>,
429    /// Set by hand in `profiles.yml`: `dre init` doesn't ask for it. For alternatives to a
430    /// prompted field (a key given as text instead of a file) and nested blocks (`ssh:`).
431    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
432    pub manual: bool,
433    /// What the value must be, checked by `validate_connection` (and `dre init` as you type).
434    #[serde(default, skip_serializing_if = "Option::is_none")]
435    pub kind: Option<FieldKind>,
436    /// The only values it may take (`dre init` offers them as a choice).
437    #[serde(default, skip_serializing_if = "Vec::is_empty")]
438    pub choices: Vec<String>,
439}
440
441/// What a connection field's value must be.
442#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)]
443#[serde(rename_all = "snake_case")]
444pub enum FieldKind {
445    String,
446    /// A whole number, or a string of one (`port: "5432"`).
447    Integer,
448    Boolean,
449    /// A file that must exist; a leading `~/` is the home directory.
450    Path,
451    /// `30s`, `2m`, `1h`, or seconds.
452    Duration,
453    /// A block of settings (`ssh:`), checked by the plugin's own rules.
454    Map,
455}
456
457impl ConnectionField {
458    pub fn new(name: &str, description: &str) -> Self {
459        ConnectionField {
460            name: name.into(),
461            description: description.into(),
462            required: false,
463            secret: false,
464            default: None,
465            same_as_source: None,
466            manual: false,
467            kind: None,
468            choices: Vec::new(),
469        }
470    }
471    #[must_use]
472    pub fn kind(mut self, kind: FieldKind) -> Self {
473        self.kind = Some(kind);
474        self
475    }
476    #[must_use]
477    pub fn choices(mut self, choices: &[&str]) -> Self {
478        self.choices = choices.iter().map(|c| c.to_string()).collect();
479        self.kind.get_or_insert(FieldKind::String);
480        self
481    }
482    #[must_use]
483    pub fn required(mut self) -> Self {
484        self.required = true;
485        self
486    }
487    #[must_use]
488    pub fn secret(mut self) -> Self {
489        self.secret = true;
490        self
491    }
492    #[must_use]
493    pub fn default(mut self, v: impl Into<Value>) -> Self {
494        self.default = Some(v.into());
495        self
496    }
497    #[must_use]
498    pub fn same_as_source(mut self, source_type: &str) -> Self {
499        self.same_as_source = Some(source_type.into());
500        self
501    }
502    #[must_use]
503    pub fn manual(mut self) -> Self {
504        self.manual = true;
505        self
506    }
507}
508
509#[cfg(test)]
510mod font_tests {
511    use super::*;
512
513    #[test]
514    fn old_write_requests_do_not_enable_font_requests() {
515        let request: Request = serde_json::from_value(serde_json::json!({
516            "type": "write", "path": "out.pdf", "format": "pdf", "options": {}, "result_sets": []
517        }))
518        .unwrap();
519        assert!(matches!(
520            request,
521            Request::Write {
522                font_requests: false,
523                ..
524            }
525        ));
526        assert!(
527            serde_json::to_value(request)
528                .unwrap()
529                .get("font_requests")
530                .is_none()
531        );
532    }
533}
534
535/// Why core couldn't supply a requested font: `message` for the person, and whether core already
536/// gave it for this font earlier in the run (`repeated`).
537#[derive(Debug, Clone, PartialEq, Eq)]
538pub struct FontError {
539    pub message: String,
540    pub repeated: bool,
541}
542
543impl From<String> for FontError {
544    fn from(message: String) -> FontError {
545        FontError {
546            message,
547            repeated: false,
548        }
549    }
550}
551
552impl From<&str> for FontError {
553    fn from(message: &str) -> FontError {
554        message.to_string().into()
555    }
556}
557
558impl std::fmt::Display for FontError {
559    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
560        f.write_str(&self.message)
561    }
562}
563
564/// One immutable font package requested by a format plugin.
565#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
566pub struct FontRequest {
567    pub name: String,
568    pub version: String,
569}
dre_protocol/msg.rs

dre_protocol/
msg.rs
