dre_protocol/
lib.rs

1//! The DRE plugin protocol, version 0.
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; stderr
6//! is the plugin's log channel. 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.
14//! - [`sessions`]: OAuth sessions plugins keep in `~/.dre/oauth_sessions.json`.
15
16pub mod conformance;
17pub mod frame;
18pub mod host;
19pub mod msg;
20pub mod options;
21pub mod plugin;
22pub mod sessions;
23pub mod util;
24
25/// This crate's version (its own, not DRE's); the fixture plugins report it.
26pub const CRATE_VERSION: &str = env!("CARGO_PKG_VERSION");
27
28/// Protocol versions this build speaks.
29pub const MIN_VERSION: u32 = 0;
30pub const MAX_VERSION: u32 = 0;
31
32/// Capability: can hold one session (and its temp objects) across requests.
33pub const CAP_SESSIONS: &str = "sessions";
34/// Capability: can open a connection read-only.
35pub const CAP_READ_ONLY: &str = "read_only";
36/// Capability: can verify a statement without executing it.
37pub const CAP_CHECK: &str = "check";
38/// Capability (destinations): takes every file of one output in a single `deliver`.
39pub const CAP_MULTI_FILE: &str = "multi_file";
40/// Source: loads rows into a temporary table on the session (`load`), for large lookups.
41pub const CAP_LOAD: &str = "load";
42/// Answers `validate` (checks a config block of options). The Rust SDK always advertises it.
43pub const CAP_VALIDATE: &str = "validate";
44
45#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize)]
46#[serde(rename_all = "lowercase")]
47pub enum Kind {
48    Source,
49    Format,
50    Destination,
51}
52
53impl Kind {
54    pub fn as_str(self) -> &'static str {
55        match self {
56            Kind::Source => "source",
57            Kind::Format => "format",
58            Kind::Destination => "destination",
59        }
60    }
61
62    /// Parse a kind, singular or plural (`source`, `sources`).
63    pub fn parse(s: &str) -> Option<Kind> {
64        match s {
65            "source" | "sources" => Some(Kind::Source),
66            "format" | "formats" => Some(Kind::Format),
67            "destination" | "destinations" => Some(Kind::Destination),
68            _ => None,
69        }
70    }
71}
72
73impl std::fmt::Display for Kind {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        f.write_str(self.as_str())
76    }
77}
78
79/// One plugin: its kind and name. Written `<kind>/<name>` (`destination/s3`), in messages as
80/// well as in `dre.lock`.
81#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
82pub struct PluginId {
83    pub kind: Kind,
84    pub name: String,
85}
86
87impl PluginId {
88    pub fn new(kind: Kind, name: impl Into<String>) -> PluginId {
89        PluginId {
90            kind,
91            name: name.into(),
92        }
93    }
94}
95
96impl std::fmt::Display for PluginId {
97    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
98        write!(f, "{}/{}", self.kind, self.name)
99    }
100}
101
102impl std::str::FromStr for PluginId {
103    type Err = String;
104
105    fn from_str(s: &str) -> Result<PluginId, String> {
106        let bad = || format!("`{s}` isn't `<kind>/<name>` (e.g. `destination/s3`)");
107        let (k, n) = s.split_once('/').ok_or_else(bad)?;
108        let kind = Kind::parse(k).ok_or_else(bad)?;
109        valid_name(n).then(|| PluginId::new(kind, n)).ok_or_else(bad)
110    }
111}
112
113impl serde::Serialize for PluginId {
114    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
115        s.collect_str(self)
116    }
117}
118
119impl<'de> serde::Deserialize<'de> for PluginId {
120    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
121        String::deserialize(d)?.parse().map_err(serde::de::Error::custom)
122    }
123}
124
125/// Whether `name` is a valid plugin or package name: `[a-z0-9_]+`.
126pub fn valid_name(name: &str) -> bool {
127    !name.is_empty()
128        && name
129            .bytes()
130            .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
131}
132
133/// The executable file name for a package on this platform: `dre-plugin-<package>`.
134pub fn package_executable_name(package: &str) -> String {
135    format!("dre-plugin-{package}{}", std::env::consts::EXE_SUFFIX)
136}
137
138/// Parse `dre-plugin-<package>[.exe]` into the package name.
139pub fn parse_package_executable_name(file: &str) -> Option<String> {
140    let stem = file.strip_suffix(std::env::consts::EXE_SUFFIX).unwrap_or(file);
141    if !std::env::consts::EXE_SUFFIX.is_empty() && stem == file {
142        return None;
143    }
144    let name = stem.strip_prefix("dre-plugin-")?;
145    valid_name(name).then(|| name.to_string())
146}
147
148/// The executable file name for a single plugin on this platform.
149pub fn executable_name(kind: Kind, name: &str) -> String {
150    format!("dre-{}-{name}{}", kind.as_str(), std::env::consts::EXE_SUFFIX)
151}
152
153/// Parse `dre-<kind>-<name>[.exe]` into its kind and name. Names are `[a-z0-9_]+`.
154pub fn parse_executable_name(file: &str) -> Option<(Kind, String)> {
155    let stem = file.strip_suffix(std::env::consts::EXE_SUFFIX).unwrap_or(file);
156    if !std::env::consts::EXE_SUFFIX.is_empty() && stem == file {
157        return None;
158    }
159    let rest = stem.strip_prefix("dre-")?;
160    let (kind, name) = rest.split_once('-')?;
161    let kind = Kind::parse(kind)?;
162    valid_name(name).then(|| (kind, name.to_string()))
163}
164
165#[cfg(test)]
166mod tests {
167    use super::*;
168
169    #[test]
170    fn parses_plugin_executable_names() {
171        let exe = |s: &str| format!("{s}{}", std::env::consts::EXE_SUFFIX);
172        assert_eq!(
173            parse_executable_name(&exe("dre-source-duckdb")),
174            Some((Kind::Source, "duckdb".into()))
175        );
176        assert_eq!(
177            parse_executable_name(&exe("dre-destination-azure_blob")),
178            Some((Kind::Destination, "azure_blob".into()))
179        );
180        assert_eq!(parse_executable_name(&exe("dre-format-csv.d")), None);
181        assert_eq!(parse_executable_name(&exe("dre-widget-x")), None);
182        assert_eq!(parse_executable_name(&exe("dre")), None);
183        assert_eq!(parse_executable_name(&exe("dre-plugin-csv")), None);
184    }
185
186    #[test]
187    fn parses_package_executable_names() {
188        let exe = |s: &str| format!("{s}{}", std::env::consts::EXE_SUFFIX);
189        assert_eq!(
190            parse_package_executable_name(&exe("dre-plugin-object_store")).as_deref(),
191            Some("object_store")
192        );
193        assert_eq!(parse_package_executable_name(&exe("dre-source-duckdb")), None);
194        assert_eq!(parse_package_executable_name(&exe("dre-plugin-")), None);
195    }
196
197    #[test]
198    fn plugin_ids_read_and_write_as_kind_slash_name() {
199        let id: PluginId = "destination/s3".parse().unwrap();
200        assert_eq!(id, PluginId::new(Kind::Destination, "s3"));
201        assert_eq!(id.to_string(), "destination/s3");
202        assert_eq!(serde_json::to_string(&id).unwrap(), "\"destination/s3\"");
203        assert!("s3".parse::<PluginId>().is_err());
204        assert!("widget/s3".parse::<PluginId>().is_err());
205    }
206}
dre_protocol/lib.rs

dre_protocol/
lib.rs
