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)]
15pub struct Envelope<T> {
16    #[serde(default, skip_serializing_if = "Option::is_none")]
17    pub id: Option<u64>,
18    #[serde(flatten)]
19    pub body: T,
20}
21
22impl<T> Envelope<T> {
23    pub fn new(id: Option<u64>, body: T) -> Envelope<T> {
24        Envelope { id, body }
25    }
26}
27
28/// Core → plugin.
29#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
30#[serde(tag = "type", rename_all = "snake_case")]
31pub enum Request {
32    /// Always first. The plugin picks a version in both ranges or replies `version_mismatch`.
33    Hello {
34        min_version: u32,
35        max_version: u32,
36        core_version: String,
37        /// The plugin core wants served, for an executable that provides several. A package
38        /// that doesn't provide it replies `error`; without it, the package serves its first.
39        #[serde(default, skip_serializing_if = "Option::is_none")]
40        plugin: Option<PluginId>,
41    },
42    /// Describe the connection fields a profile output for this plugin takes (used by `dre init`)
43    /// and the options a report's config block for it takes.
44    Describe {},
45    /// Check one config block of options (a format's `output:` keys, a destination entry's keys)
46    /// without doing anything. Replies `validated`.
47    Validate {
48        options: Map<String, Value>,
49    },
50    /// Source or destination (protocol 1): check a profile entry's connection settings without
51    /// connecting: the declared fields' generic checks, then the plugin's own rules. Keys in
52    /// `unresolved` (an unset `env_var()`) arrive as `null` and count as set, unchecked. Replies
53    /// `validated`, with `warnings` for unknown keys.
54    ValidateConnection {
55        connection: Map<String, Value>,
56        #[serde(default, skip_serializing_if = "Vec::is_empty")]
57        unresolved: Vec<String>,
58    },
59    /// Source: open a session. All later `execute`/`check` requests run on it.
60    Open {
61        connection: Map<String, Value>,
62        read_only: bool,
63    },
64    /// Source: run one statement. Replies `result` (then Arrow frames, then `result_end`) or
65    /// `no_result`.
66    Execute {
67        sql: String,
68        #[serde(default, skip_serializing_if = "Option::is_none")]
69        row_limit: Option<u64>,
70    },
71    /// Source: verify one statement without executing it. Replies `ok` or `error`.
72    Check {
73        sql: String,
74    },
75    /// Format: write result sets to `path`. Followed, per result set, by Arrow frames (at least one,
76    /// carrying the schema) and a `result_set_end`, then `finish`. Replies `written`.
77    Write {
78        path: String,
79        format: String,
80        options: Map<String, Value>,
81        result_sets: Vec<ResultSetMeta>,
82        #[serde(default, skip_serializing_if = "Option::is_none")]
83        template: Option<Value>,
84    },
85    ResultSetEnd {},
86    Finish {},
87    /// Destination: deliver a local file, or with `files` (only to a plugin advertising
88    /// `multi_file`) every file of one output at once. Exactly one of `local_path`/`files` is set,
89    /// unless `message` is: then `files` (possibly empty) holds the files to attach to it, and is
90    /// sent only to a plugin advertising `message`. `options` are the destination entry's plugin
91    /// options, rendered by core. Replies `delivered`.
92    Deliver {
93        #[serde(default, skip_serializing_if = "Option::is_none")]
94        local_path: Option<String>,
95        #[serde(default, skip_serializing_if = "Option::is_none")]
96        remote_path: Option<String>,
97        #[serde(default, skip_serializing_if = "Vec::is_empty")]
98        files: Vec<DeliveryFile>,
99        connection: Map<String, Value>,
100        #[serde(default)]
101        options: Map<String, Value>,
102        #[serde(default, skip_serializing_if = "Option::is_none")]
103        message: Option<Message>,
104    },
105    /// Source: load rows into a temporary table on the session, named after `name`. Followed by
106    /// Arrow frames (at least one, carrying the schema) and a `result_set_end`. Replies `loaded`.
107    /// Only sent to plugins advertising `load`.
108    Load {
109        name: String,
110    },
111    /// End the conversation; the plugin exits 0.
112    Close {},
113    /// Protocol 1: stop the request whose id the envelope carries. It has no reply of its own:
114    /// the request replies `error` with kind `cancelled`, or its usual reply if it finished
115    /// first. A cancel for a request that isn't running is ignored.
116    Cancel {},
117}
118
119/// Plugin → core.
120#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
121#[serde(tag = "type", rename_all = "snake_case")]
122pub enum Response {
123    Hello {
124        protocol_version: u32,
125        kind: Kind,
126        name: String,
127        version: String,
128        #[serde(default)]
129        capabilities: Vec<String>,
130        /// Every plugin the executable serves; empty for one that serves only `kind`/`name`.
131        #[serde(default, skip_serializing_if = "Vec::is_empty")]
132        provides: Vec<PluginId>,
133    },
134    VersionMismatch {
135        min_version: u32,
136        max_version: u32,
137    },
138    Describe {
139        connection_fields: Vec<ConnectionField>,
140        #[serde(default, skip_serializing_if = "Vec::is_empty")]
141        option_fields: Vec<OptionField>,
142        /// Source: the character the database quotes identifiers with (`"` for DuckDB and
143        /// Postgres, a backtick for Databricks). Core uses it to quote names a project asks to
144        /// quote (a source's `quoting:`). Every source sets it; other kinds leave it out.
145        #[serde(default, skip_serializing_if = "Option::is_none")]
146        identifier_quote: Option<String>,
147        /// Destination advertising `message`: the most characters a message may have in the
148        /// service, after translation. Core shows it next to a message's length in a preview.
149        #[serde(default, skip_serializing_if = "Option::is_none")]
150        message_limit: Option<u64>,
151    },
152    /// Every problem with the options, each a sentence naming the key; empty when they're fine.
153    Validated {
154        errors: Vec<String>,
155        /// Problems that don't stop a run (an unknown key); `validate_connection` only.
156        #[serde(default, skip_serializing_if = "Vec::is_empty")]
157        warnings: Vec<String>,
158    },
159    Ok {},
160    /// A result set follows as Arrow frames, ended by `result_end`.
161    Result {
162        columns: Vec<String>,
163    },
164    ResultEnd {
165        rows: u64,
166    },
167    NoResult {
168        #[serde(default, skip_serializing_if = "Option::is_none")]
169        rows_affected: Option<u64>,
170    },
171    Written {
172        files: Vec<String>,
173        /// Things the person should know about the files (a value written as text, say).
174        #[serde(default, skip_serializing_if = "Vec::is_empty")]
175        warnings: Vec<String>,
176    },
177    Delivered {
178        location: String,
179        /// How many tries the delivery took, when it was more than one (protocol 1).
180        #[serde(default, skip_serializing_if = "Option::is_none")]
181        attempts: Option<u32>,
182    },
183    /// `relation` is what SQL uses to read the loaded rows. `warning`, when set, is shown to the
184    /// user (e.g. the database has no bulk path, so a load this size is slow).
185    Loaded {
186        relation: String,
187        rows: u64,
188        #[serde(default, skip_serializing_if = "Option::is_none")]
189        warning: Option<String>,
190    },
191    /// The request failed. `kind` is one of [`ERROR_KINDS`]; `code` names the problem,
192    /// namespaced by the plugin (`sftp/host-key-mismatch`). Both are optional (protocol 0 has
193    /// neither): without a `code`, core reports the failure under its own code for the step;
194    /// a `code` without a `kind` counts as `internal`.
195    Error {
196        message: String,
197        #[serde(default, skip_serializing_if = "Option::is_none")]
198        kind: Option<String>,
199        #[serde(default, skip_serializing_if = "Option::is_none")]
200        code: Option<String>,
201    },
202    /// Protocol 1: a log line, while a request runs (with its id) or between requests.
203    Log {
204        level: LogLevel,
205        message: String,
206        /// Structured context (`{"attempt": 2}`); never secrets.
207        #[serde(default, skip_serializing_if = "Map::is_empty")]
208        fields: Map<String, Value>,
209    },
210    /// Protocol 1, optional: how far a long request has got. Core shows it when present.
211    Progress {
212        #[serde(default, skip_serializing_if = "Option::is_none")]
213        message: Option<String>,
214        #[serde(default, skip_serializing_if = "Option::is_none")]
215        done: Option<u64>,
216        #[serde(default, skip_serializing_if = "Option::is_none")]
217        total: Option<u64>,
218    },
219}
220
221impl Response {
222    /// An `error` reply with only a message.
223    pub fn error(message: impl Into<String>) -> Response {
224        Response::Error {
225            message: message.into(),
226            kind: None,
227            code: None,
228        }
229    }
230}
231
232/// The kinds an `error` reply may give, from DRE's error-code registry.
233pub const ERROR_KINDS: &[&str] = &[
234    "config",
235    "plugin",
236    "refused",
237    "connection",
238    "auth",
239    "query",
240    "delivery",
241    "internal",
242    "cancelled",
243    "timed_out",
244];
245
246/// How important a `log` message is. Core shows `info` and above to the person, and keeps
247/// `debug` and `trace` for `-v` and the log file.
248#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
249#[serde(rename_all = "lowercase")]
250pub enum LogLevel {
251    Error,
252    Warn,
253    Info,
254    Debug,
255    Trace,
256}
257
258/// A rendered message, as `deliver` carries it.
259#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
260pub struct Message {
261    /// One line, plain text.
262    pub title: String,
263    /// The portable Markdown subset ([`crate::markdown`]).
264    pub text: String,
265    /// An HTML body, when one was made; core leaves it out today. A plugin that can show HTML
266    /// (email) uses it when present, else converts `text`.
267    #[serde(default, skip_serializing_if = "Option::is_none")]
268    pub html: Option<String>,
269    /// The `.md` file core wrote: `# title`, then the text.
270    pub path: String,
271}
272
273/// One file of a multi-file `deliver`.
274#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
275pub struct DeliveryFile {
276    pub local_path: String,
277    #[serde(default, skip_serializing_if = "Option::is_none")]
278    pub remote_path: Option<String>,
279}
280
281/// Per result set, what a format plugin needs to lay it out.
282#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
283pub struct ResultSetMeta {
284    /// Sheet name (xlsx) or file suffix (single-table formats).
285    pub name: String,
286    /// The query (`.sql` basename) that produced it.
287    pub query: String,
288    /// 1-based index among that query's result sets.
289    pub result_index: usize,
290    #[serde(default, skip_serializing_if = "Option::is_none")]
291    pub anchor: Option<String>,
292    #[serde(default, skip_serializing_if = "Option::is_none")]
293    pub header: Option<bool>,
294    /// The query entry's `columns:` map: per result column, how to show it.
295    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
296    pub columns: std::collections::BTreeMap<String, ColumnOptions>,
297    /// The query entry's `autofit` (xlsx): size this tab's columns from their content, over the
298    /// output's `autofit`.
299    #[serde(default, skip_serializing_if = "Option::is_none")]
300    pub autofit: Option<bool>,
301    /// The query entry's `style` (xlsx), over the output's.
302    #[serde(default, skip_serializing_if = "Option::is_none")]
303    pub style: Option<crate::style::SheetStyle>,
304}
305
306/// One entry of a `columns:` map.
307#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
308pub struct ColumnOptions {
309    /// An Excel number format code.
310    #[serde(default, skip_serializing_if = "Option::is_none")]
311    pub format: Option<String>,
312    /// A row formula (`={qty}*{price}`), written in place of the column's value, which becomes
313    /// its cached result.
314    #[serde(default, skip_serializing_if = "Option::is_none")]
315    pub formula: Option<String>,
316    /// A totals row entry under the data: `sum`, `average`, `count`, `min`, `max`, or a formula
317    /// over whole columns (`=SUM({amount:*})/COUNT({qty:*})`).
318    #[serde(default, skip_serializing_if = "Option::is_none")]
319    pub total: Option<String>,
320    /// The column's width (xlsx), over the tab's and output's `autofit`.
321    #[serde(default, skip_serializing_if = "Option::is_none")]
322    pub width: Option<ColumnWidth>,
323    /// How the column's data cells look (xlsx), over the tab's and output's `style`.
324    #[serde(default, skip_serializing_if = "Option::is_none")]
325    pub style: Option<crate::style::CellStyle>,
326}
327
328/// A column width: sized from the content, or a number of characters.
329#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
330#[serde(untagged)]
331pub enum ColumnWidth {
332    Chars(f64),
333    Auto(AutoWidth),
334}
335
336/// The word `auto`.
337#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
338#[serde(rename_all = "lowercase")]
339pub enum AutoWidth {
340    Auto,
341}
342
343#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
344pub struct ConnectionField {
345    pub name: String,
346    #[serde(default)]
347    pub description: String,
348    #[serde(default)]
349    pub required: bool,
350    /// Offered as an `env_var()` reference by default in `dre init`.
351    #[serde(default)]
352    pub secret: bool,
353    #[serde(default, skip_serializing_if = "Option::is_none")]
354    pub default: Option<Value>,
355    /// A source type whose profile has the same field: `dre init` offers the value entered for
356    /// that source as this field's default (e.g. one Databricks host for source and destination).
357    #[serde(default, skip_serializing_if = "Option::is_none")]
358    pub same_as_source: Option<String>,
359    /// Set by hand in `profiles.yml`: `dre init` doesn't ask for it. For alternatives to a
360    /// prompted field (a key given as text instead of a file) and nested blocks (`ssh:`).
361    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
362    pub manual: bool,
363    /// What the value must be, checked by `validate_connection` (and `dre init` as you type).
364    #[serde(default, skip_serializing_if = "Option::is_none")]
365    pub kind: Option<FieldKind>,
366    /// The only values it may take (`dre init` offers them as a choice).
367    #[serde(default, skip_serializing_if = "Vec::is_empty")]
368    pub choices: Vec<String>,
369}
370
371/// What a connection field's value must be.
372#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
373#[serde(rename_all = "snake_case")]
374pub enum FieldKind {
375    String,
376    /// A whole number, or a string of one (`port: "5432"`).
377    Integer,
378    Boolean,
379    /// A file that must exist; a leading `~/` is the home directory.
380    Path,
381    /// `30s`, `2m`, `1h`, or seconds.
382    Duration,
383    /// A block of settings (`ssh:`), checked by the plugin's own rules.
384    Map,
385}
386
387impl ConnectionField {
388    pub fn new(name: &str, description: &str) -> Self {
389        ConnectionField {
390            name: name.into(),
391            description: description.into(),
392            required: false,
393            secret: false,
394            default: None,
395            same_as_source: None,
396            manual: false,
397            kind: None,
398            choices: Vec::new(),
399        }
400    }
401    pub fn kind(mut self, kind: FieldKind) -> Self {
402        self.kind = Some(kind);
403        self
404    }
405    pub fn choices(mut self, choices: &[&str]) -> Self {
406        self.choices = choices.iter().map(|c| c.to_string()).collect();
407        self.kind.get_or_insert(FieldKind::String);
408        self
409    }
410    pub fn required(mut self) -> Self {
411        self.required = true;
412        self
413    }
414    pub fn secret(mut self) -> Self {
415        self.secret = true;
416        self
417    }
418    pub fn default(mut self, v: impl Into<Value>) -> Self {
419        self.default = Some(v.into());
420        self
421    }
422    pub fn same_as_source(mut self, source_type: &str) -> Self {
423        self.same_as_source = Some(source_type.into());
424        self
425    }
426    pub fn manual(mut self) -> Self {
427        self.manual = true;
428        self
429    }
430}
dre_protocol/msg.rs

dre_protocol/
msg.rs
