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