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