[PATCH pve-cluster 03/12] rust: notify: add the wire protocol

[Hannes Laimer <h.laimer@proxmox.com>]

Requests, replies and events are all JSON objects, one per line. So a
client needs a single parser, and the stream stays readable as text.
A request is a command with its arguments, and a reply carries either
a result or an error. An event carries a mutation's type, the path,
the rename target and the captured values per subscription name.

Signed-off-by: Hannes Laimer <h.laimer@proxmox.com>
---
 src/rust/pmxcfs-notify/src/lib.rs      |   8 ++
 src/rust/pmxcfs-notify/src/protocol.rs | 185 +++++++++++++++++++++++++
 2 files changed, 193 insertions(+)
 create mode 100644 src/rust/pmxcfs-notify/src/protocol.rs

diff --git a/src/rust/pmxcfs-notify/src/lib.rs b/src/rust/pmxcfs-notify/src/lib.rs
index 889fe50..d35d3c5 100644
--- a/src/rust/pmxcfs-notify/src/lib.rs
+++ b/src/rust/pmxcfs-notify/src/lib.rs
@@ -1,5 +1,12 @@
 //! Change notification socket for pmxcfs.
 //!
+//! Clients connect to a Unix stream socket and exchange newline delimited JSON objects. A request
+//! is `{"command": "...", "args": {...}}` and is answered with `{"ok": <json>}` or
+//! `{"error": "..."}`. Once a client has sent `subscribe` with named path templates, every
+//! matching memdb mutation is pushed as `{"event": {...}}` with the event type, the path, the
+//! rename target, and per subscription name the parameter sets captured from the path and the
+//! rename target.
+//!
 //! A template is a memdb path whose components may each hold one `{name}` placeholder for all or
 //! part of that component. The last component may instead be `{name...}` for the rest of the path.
 //! `nodes/{node}/qemu-server/{vmid}.conf` captures node and vmid, and `{path...}` matches every
@@ -7,6 +14,7 @@
 
 #![allow(dead_code)]
 
+mod protocol;
 mod template;
 
 /// Events and templates name memdb paths without a leading slash.
diff --git a/src/rust/pmxcfs-notify/src/protocol.rs b/src/rust/pmxcfs-notify/src/protocol.rs
new file mode 100644
index 0000000..d28c9dc
--- /dev/null
+++ b/src/rust/pmxcfs-notify/src/protocol.rs
@@ -0,0 +1,185 @@
+use std::collections::BTreeMap;
+
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+use crate::template::{Captures, Template};
+
+pub(crate) const PROTOCOL_VERSION: u32 = 1;
+
+/// Keyed by subscription name, with one capture set per side of the event that matched.
+pub(crate) type Params<'a> = BTreeMap<&'a str, Vec<Captures<'a>>>;
+
+#[non_exhaustive]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "lowercase")]
+pub enum EventKind {
+    Create,
+    Write,
+    Mtime,
+    Rename,
+    Delete,
+    Mkdir,
+    Resync,
+}
+
+#[derive(Serialize)]
+pub(crate) struct Event<'a> {
+    #[serde(rename = "type")]
+    pub kind: EventKind,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub path: Option<&'a str>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub to: Option<&'a str>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub params: Option<Params<'a>>,
+}
+
+/// Everything the server sends. The variant name is the single key of the JSON object on the wire.
+#[derive(Serialize)]
+#[serde(rename_all = "lowercase")]
+pub(crate) enum Message<'a> {
+    Ok(Value),
+    Error(String),
+    Event(Event<'a>),
+}
+
+impl Message<'static> {
+    pub(crate) fn resync() -> Self {
+        Message::Event(Event {
+            kind: EventKind::Resync,
+            path: None,
+            to: None,
+            params: None,
+        })
+    }
+}
+
+impl Message<'_> {
+    pub(crate) fn line(&self) -> String {
+        let mut line = serde_json::to_string(self).expect("message serialization cannot fail");
+        line.push('\n');
+        line
+    }
+}
+
+#[derive(Debug, PartialEq, Eq, Deserialize)]
+#[serde(tag = "command", content = "args", rename_all = "lowercase")]
+pub(crate) enum Command {
+    Hello,
+    Subscribe {
+        #[serde(default)]
+        patterns: BTreeMap<String, Template>,
+    },
+    Unsubscribe,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn parse(line: &str) -> Result<Command, serde_json::Error> {
+        serde_json::from_str(line)
+    }
+
+    fn template(text: &str) -> Template {
+        text.parse().unwrap()
+    }
+
+    #[test]
+    fn parses_commands() {
+        assert_eq!(parse(r#"{"command":"hello"}"#).unwrap(), Command::Hello);
+        assert_eq!(
+            parse(
+                r#"{"command":"subscribe","args":{"patterns":{"guest":"nodes/{node}/qemu-server/{vmid}.conf","dc":"/datacenter.cfg"}}}"#
+            )
+            .unwrap(),
+            Command::Subscribe {
+                patterns: BTreeMap::from([
+                    ("dc".to_owned(), template("datacenter.cfg")),
+                    (
+                        "guest".to_owned(),
+                        template("nodes/{node}/qemu-server/{vmid}.conf")
+                    ),
+                ]),
+            }
+        );
+        assert_eq!(
+            parse(r#"{"command":"subscribe","args":{}}"#).unwrap(),
+            Command::Subscribe {
+                patterns: BTreeMap::new(),
+            }
+        );
+        assert_eq!(
+            parse(r#"{"command":"unsubscribe"}"#).unwrap(),
+            Command::Unsubscribe
+        );
+    }
+
+    #[test]
+    fn rejects_bad_requests() {
+        assert!(parse("not json").is_err());
+        assert!(parse(r#"{"args":{}}"#).is_err());
+        assert!(parse(r#"{"command":"subscribe"}"#).is_err());
+        assert!(parse(r#"{"command":"subscribe","args":{"patterns":["a.cfg"]}}"#).is_err());
+        assert!(parse(r#"{"command":"subscribe","args":{"patterns":{"x":1}}}"#).is_err());
+        let err =
+            parse(r#"{"command":"subscribe","args":{"patterns":{"x":"{a}{b}"}}}"#).unwrap_err();
+        assert!(
+            err.to_string()
+                .starts_with("only one placeholder per component in '{a}{b}'"),
+            "{err}"
+        );
+        let err = parse(r#"{"command":"bogus"}"#).unwrap_err();
+        assert!(
+            err.to_string().starts_with("unknown variant `bogus`"),
+            "{err}"
+        );
+    }
+
+    #[test]
+    fn formats_events() {
+        let params = Params::from([(
+            "guest",
+            vec![Captures::from([("node", "n1"), ("vmid", "100")])],
+        )]);
+        let write = Message::Event(Event {
+            kind: EventKind::Write,
+            path: Some("nodes/n1/qemu-server/100.conf"),
+            to: None,
+            params: Some(params),
+        });
+        assert_eq!(
+            write.line(),
+            "{\"event\":{\"type\":\"write\",\"path\":\"nodes/n1/qemu-server/100.conf\",\"params\":{\"guest\":[{\"node\":\"n1\",\"vmid\":\"100\"}]}}}\n"
+        );
+        let params = Params::from([("dc", vec![Captures::new()])]);
+        let rename = Message::Event(Event {
+            kind: EventKind::Rename,
+            path: Some("a"),
+            to: Some("datacenter.cfg"),
+            params: Some(params),
+        });
+        assert_eq!(
+            rename.line(),
+            "{\"event\":{\"type\":\"rename\",\"path\":\"a\",\"to\":\"datacenter.cfg\",\"params\":{\"dc\":[{}]}}}\n"
+        );
+        assert_eq!(
+            Message::resync().line(),
+            "{\"event\":{\"type\":\"resync\"}}\n"
+        );
+    }
+
+    #[test]
+    fn formats_replies() {
+        assert_eq!(Message::Ok(Value::Null).line(), "{\"ok\":null}\n");
+        assert_eq!(
+            Message::Ok(serde_json::json!({"protocol": 1})).line(),
+            "{\"ok\":{\"protocol\":1}}\n"
+        );
+        assert_eq!(
+            Message::Error("multi\nline".to_owned()).line(),
+            "{\"error\":\"multi\\nline\"}\n"
+        );
+    }
+}
-- 
2.47.3