dre_protocol/
lib.rs

1//! The DRE plugin protocol, version 1.
2//!
3//! Plugins ship in packages: one executable, `dre-plugin-<package>`, serving every plugin the
4//! package provides (a single plugin may also be named `dre-<kind>-<name>`). It talks to DRE core
5//! over stdin/stdout in length-prefixed frames. Each frame is a JSON control message or an Arrow IPC stream; plugins
6//! log through `log` messages, and stderr carries what can't (panics, native drivers). See `docs/protocol.md` in the repository for the public contract.
7//!
8//! - [`frame`]: the wire format.
9//! - [`msg`]: every control message.
10//! - [`host`]: the core side: spawning a plugin, the handshake, requests.
11//! - [`plugin`]: the SDK plugin authors use to serve requests.
12//! - [`options`]: the options a plugin declares, and how they're checked.
13//! - [`conformance`]: checks any plugin binary against the protocol, from the language-neutral
14//!   fixture cases in `conformance/`.
15//! - [`schema`]: the JSON Schema of every message, published as `conformance/schema.json`.
16//! - [`sessions`]: OAuth sessions plugins keep in `~/.dre/oauth_sessions.json`.
17//! - [`markdown`]: the portable Markdown subset of messages, and its translations.
18//! - [`delivery`]: the delivery and network rules every file destination shares.
19
20// A public function that can panic says when in a `# Panics` section (CI denies warnings).
21#![warn(clippy::missing_panics_doc)]
22
23/// The `log` crate, re-exported: plugins log with `dre_protocol::log::info!` and friends, which
24/// the SDK sends to core as `log` messages.
25pub use log;
26
27pub mod conformance;
28pub mod connection;
29pub mod delivery;
30pub mod frame;
31#[cfg(target_os = "linux")]
32mod fuse_fds;
33pub mod host;
34pub mod markdown;
35pub mod msg;
36pub mod options;
37pub mod plugin;
38pub mod schema;
39pub mod sessions;
40pub mod style;
41pub mod util;
42
43/// This crate's version (its own, not DRE's); the fixture plugins report it.
44pub const CRATE_VERSION: &str = env!("CARGO_PKG_VERSION");
45
46/// Protocol versions the plugin SDK speaks: a plugin built on it also works with an older core.
47pub const MIN_VERSION: u32 = 0;
48pub const MAX_VERSION: u32 = 1;
49/// The oldest protocol version core accepts from a plugin.
50pub const CORE_MIN_VERSION: u32 = 1;
51
52/// Capability: can hold one session (and its temp objects) across requests.
53pub const CAP_SESSIONS: &str = "sessions";
54/// Capability: can open a connection read-only.
55pub const CAP_READ_ONLY: &str = "read_only";
56/// Capability: can verify a statement without executing it.
57pub const CAP_CHECK: &str = "check";
58/// Capability (destinations): takes every file of one output in a single `deliver`.
59pub const CAP_MULTI_FILE: &str = "multi_file";
60/// Capability (destinations): takes a message (`deliver` with `message`), not only files.
61pub const CAP_MESSAGE: &str = "message";
62/// Capability (destinations): takes only messages; core never sends it a file output.
63pub const CAP_MESSAGE_ONLY: &str = "message_only";
64/// Source: loads rows into a temporary table on the session (`load`), for large lookups.
65pub const CAP_LOAD: &str = "load";
66/// Source: discovers accessible relations through the connection's native metadata.
67pub const CAP_CATALOG: &str = "catalog";
68/// Answers `validate` (checks a config block of options). The Rust SDK always advertises it.
69pub const CAP_VALIDATE: &str = "validate";
70
71/// Every capability a plugin may advertise, as the published schema lists them.
72pub const CAPABILITIES: &[&str] = &[
73    CAP_SESSIONS,
74    CAP_READ_ONLY,
75    CAP_CHECK,
76    CAP_MULTI_FILE,
77    CAP_MESSAGE,
78    CAP_MESSAGE_ONLY,
79    CAP_LOAD,
80    CAP_CATALOG,
81    CAP_VALIDATE,
82];
83
84#[derive(
85    Debug,
86    Clone,
87    Copy,
88    PartialEq,
89    Eq,
90    PartialOrd,
91    Ord,
92    Hash,
93    serde::Serialize,
94    serde::Deserialize,
95    schemars::JsonSchema,
96)]
97#[serde(rename_all = "lowercase")]
98pub enum Kind {
99    Source,
100    Format,
101    Destination,
102}
103
104impl Kind {
105    pub fn as_str(self) -> &'static str {
106        match self {
107            Kind::Source => "source",
108            Kind::Format => "format",
109            Kind::Destination => "destination",
110        }
111    }
112
113    /// Parse a kind, singular or plural (`source`, `sources`).
114    pub fn parse(s: &str) -> Option<Kind> {
115        match s {
116            "source" | "sources" => Some(Kind::Source),
117            "format" | "formats" => Some(Kind::Format),
118            "destination" | "destinations" => Some(Kind::Destination),
119            _ => None,
120        }
121    }
122}
123
124impl std::fmt::Display for Kind {
125    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
126        f.write_str(self.as_str())
127    }
128}
129
130/// One plugin: its kind and name. Written `<kind>/<name>` (`destination/s3`), in messages as
131/// well as in `dre.lock`.
132#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
133pub struct PluginId {
134    pub kind: Kind,
135    pub name: String,
136}
137
138impl PluginId {
139    pub fn new(kind: Kind, name: impl Into<String>) -> PluginId {
140        PluginId {
141            kind,
142            name: name.into(),
143        }
144    }
145}
146
147impl std::fmt::Display for PluginId {
148    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
149        write!(f, "{}/{}", self.kind, self.name)
150    }
151}
152
153impl std::str::FromStr for PluginId {
154    type Err = String;
155
156    fn from_str(s: &str) -> Result<PluginId, String> {
157        let bad = || format!("`{s}` isn't `<kind>/<name>` (e.g. `destination/s3`)");
158        let (k, n) = s.split_once('/').ok_or_else(bad)?;
159        let kind = Kind::parse(k).ok_or_else(bad)?;
160        valid_name(n).then(|| PluginId::new(kind, n)).ok_or_else(bad)
161    }
162}
163
164impl serde::Serialize for PluginId {
165    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
166        s.collect_str(self)
167    }
168}
169
170impl schemars::JsonSchema for PluginId {
171    fn schema_name() -> std::borrow::Cow<'static, str> {
172        "PluginId".into()
173    }
174
175    fn json_schema(_: &mut schemars::SchemaGenerator) -> schemars::Schema {
176        schemars::json_schema!({
177            "description": "One plugin, written `<kind>/<name>` (`destination/s3`).",
178            "type": "string",
179            "pattern": "^(source|format|destination)/[a-z0-9_]+$"
180        })
181    }
182}
183
184impl<'de> serde::Deserialize<'de> for PluginId {
185    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
186        String::deserialize(d)?.parse().map_err(serde::de::Error::custom)
187    }
188}
189
190/// Whether `name` is a valid plugin or package name: `[a-z0-9_]+`.
191pub fn valid_name(name: &str) -> bool {
192    !name.is_empty()
193        && name
194            .bytes()
195            .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
196}
197
198/// The executable file name for a package on this platform: `dre-plugin-<package>`.
199pub fn package_executable_name(package: &str) -> String {
200    format!("dre-plugin-{package}{}", std::env::consts::EXE_SUFFIX)
201}
202
203/// Parse `dre-plugin-<package>[.exe]` into the package name.
204pub fn parse_package_executable_name(file: &str) -> Option<String> {
205    let stem = file.strip_suffix(".exe").unwrap_or(file);
206    let name = stem.strip_prefix("dre-plugin-")?;
207    valid_name(name).then(|| name.to_string())
208}
209
210/// The executable file name for a single plugin on this platform.
211pub fn executable_name(kind: Kind, name: &str) -> String {
212    format!("dre-{}-{name}{}", kind.as_str(), std::env::consts::EXE_SUFFIX)
213}
214
215/// Parse `dre-<kind>-<name>[.exe]` into its kind and name. Names are `[a-z0-9_]+`.
216pub fn parse_executable_name(file: &str) -> Option<(Kind, String)> {
217    let stem = file.strip_suffix(".exe").unwrap_or(file);
218    let rest = stem.strip_prefix("dre-")?;
219    let (kind, name) = rest.split_once('-')?;
220    let kind = Kind::parse(kind)?;
221    valid_name(name).then(|| (kind, name.to_string()))
222}
223
224#[cfg(test)]
225mod tests {
226    use super::*;
227
228    #[test]
229    fn parses_plugin_executable_names() {
230        let exe = |s: &str| format!("{s}{}", std::env::consts::EXE_SUFFIX);
231        assert_eq!(
232            parse_executable_name(&exe("dre-source-duckdb")),
233            Some((Kind::Source, "duckdb".into()))
234        );
235        assert_eq!(
236            parse_executable_name(&exe("dre-destination-azure_blob")),
237            Some((Kind::Destination, "azure_blob".into()))
238        );
239        assert_eq!(parse_executable_name(&exe("dre-format-csv.d")), None);
240        assert_eq!(parse_executable_name("dre-source-fixture").unwrap().1, "fixture");
241        assert_eq!(
242            parse_executable_name("dre-source-fixture.exe").unwrap().1,
243            "fixture"
244        );
245        assert_eq!(parse_executable_name(&exe("dre-widget-x")), None);
246        assert_eq!(parse_executable_name(&exe("dre")), None);
247        assert_eq!(parse_executable_name(&exe("dre-plugin-csv")), None);
248    }
249
250    #[test]
251    fn parses_package_executable_names() {
252        let exe = |s: &str| format!("{s}{}", std::env::consts::EXE_SUFFIX);
253        assert_eq!(
254            parse_package_executable_name(&exe("dre-plugin-object_store")).as_deref(),
255            Some("object_store")
256        );
257        assert_eq!(parse_package_executable_name(&exe("dre-source-duckdb")), None);
258        assert_eq!(parse_package_executable_name(&exe("dre-plugin-")), None);
259        assert_eq!(
260            parse_package_executable_name("dre-plugin-fixture").as_deref(),
261            Some("fixture")
262        );
263        assert_eq!(
264            parse_package_executable_name("dre-plugin-fixture.exe").as_deref(),
265            Some("fixture")
266        );
267    }
268
269    #[test]
270    fn plugin_ids_read_and_write_as_kind_slash_name() {
271        let id: PluginId = "destination/s3".parse().unwrap();
272        assert_eq!(id, PluginId::new(Kind::Destination, "s3"));
273        assert_eq!(id.to_string(), "destination/s3");
274        assert_eq!(serde_json::to_string(&id).unwrap(), "\"destination/s3\"");
275        assert!("s3".parse::<PluginId>().is_err());
276        assert!("widget/s3".parse::<PluginId>().is_err());
277    }
278}
dre_protocol/lib.rs

dre_protocol/
lib.rs
