Publish theme snapshots through IPC from explicit live state

crates/ferese-ipc/src/theme.rs

@@ -4,11 +4,47 @@ use std::os::unix::net::UnixStream;
 use std::path::PathBuf;
 use std::time::Duration;
 
-use ferese_config::theme::Snapshot;
+use ferese_theme_model::{Mode, ResolvedTheme, families::Family};
+use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 
 use crate::{Request, Response, VERSION, read_frame, write_frame};
 
+pub const SCHEMA_VERSION: u32 = 2;
+
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
+pub struct Snapshot {
+    pub version: u32,
+    pub revision: u64,
+    pub mode: Mode,
+    pub theme: ResolvedTheme,
+    pub presented: ResolvedTheme,
+    pub warnings: Vec<String>,
+    pub error: Option<String>,
+    pub families: Vec<Family>,
+    #[serde(default)]
+    pub fallback_note: Option<String>,
+}
+
+impl Snapshot {
+    /// Decode older messages with the caller's catalog. Explicit empty catalogs stay empty.
+    pub fn decode(mut value: Value, fallback_families: impl FnOnce() -> Vec<Family>) -> Result<Self, String> {
+        if let Some(object) = value.as_object_mut()
+            && !object.contains_key("families")
+        {
+            object.insert(
+                "families".into(),
+                serde_json::to_value(fallback_families()).map_err(|e| e.to_string())?,
+            );
+        }
+        let snapshot: Self = serde_json::from_value(value).map_err(|e| e.to_string())?;
+        if snapshot.version != SCHEMA_VERSION {
+            return Err("Unsupported theme snapshot version".into());
+        }
+        Ok(snapshot)
+    }
+}
+
 pub fn socket_path() -> io::Result<PathBuf> {
     std::env::var_os("XDG_RUNTIME_DIR")
         .filter(|value| !value.is_empty())
@@ -53,23 +89,17 @@ impl Connection {
         response.result.ok_or_else(|| "Missing theme snapshot".into())
     }
 
-    pub fn get(&mut self) -> Result<Snapshot, String> {
-        let snapshot: Snapshot =
-            serde_json::from_value(self.call("theme-get", json!({}))?).map_err(|e| e.to_string())?;
-        if snapshot.version != ferese_config::theme::SCHEMA_VERSION {
-            return Err("Unsupported theme snapshot version".into());
-        }
-        Ok(snapshot)
+    pub fn get(&mut self, fallback_families: impl FnOnce() -> Vec<Family>) -> Result<Snapshot, String> {
+        Snapshot::decode(self.call("theme-get", json!({}))?, fallback_families)
     }
 
-    pub fn watch(&mut self, revision: u64) -> Result<Snapshot, String> {
+    pub fn watch(
+        &mut self,
+        revision: u64,
+        fallback_families: impl FnOnce() -> Vec<Family>,
+    ) -> Result<Snapshot, String> {
         self.stream.set_read_timeout(None).map_err(|e| e.to_string())?;
-        let snapshot: Snapshot =
-            serde_json::from_value(self.call("theme-watch", json!({"since": revision}))?).map_err(|e| e.to_string())?;
-        if snapshot.version != ferese_config::theme::SCHEMA_VERSION {
-            return Err("Unsupported theme snapshot version".into());
-        }
-        Ok(snapshot)
+        Snapshot::decode(self.call("theme-watch", json!({"since": revision}))?, fallback_families)
     }
 }
 
@@ -81,9 +111,51 @@ impl Drop for Cancellation {
     }
 }
 
-pub fn current() -> Snapshot {
-    Connection::connect()
-        .ok()
-        .and_then(|mut connection| connection.get().ok())
-        .unwrap_or_default()
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::mpsc;
+
+    #[test]
+    fn cancelling_a_blocked_watch_releases_the_client_and_server() {
+        let (stream, mut server) = UnixStream::pair().unwrap();
+        let mut connection = Connection { stream, id: 0 };
+        let cancellation = connection.cancellation().unwrap();
+        let (send, receive) = mpsc::channel();
+        let worker = std::thread::spawn(move || {
+            send.send(connection.watch(7, Vec::new)).unwrap();
+        });
+        server.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+        let request: Request = read_frame(&mut server).unwrap();
+        assert_eq!(request.command, "theme-watch");
+        assert_eq!(request.args, json!({"since": 7}));
+        drop(cancellation);
+        assert!(receive.recv_timeout(Duration::from_secs(2)).unwrap().is_err());
+        assert!(read_frame::<Request>(&mut server).is_err());
+        worker.join().unwrap();
+    }
+
+    #[test]
+    fn response_identity_is_checked_before_decoding_theme_values() {
+        for wrong_version in [false, true] {
+            let (stream, mut server) = UnixStream::pair().unwrap();
+            server.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+            stream.set_read_timeout(Some(Duration::from_secs(2))).unwrap();
+            let worker = std::thread::spawn(move || {
+                let request: Request = read_frame(&mut server).unwrap();
+                let mut response = Response::success(request.id, json!({}));
+                if wrong_version {
+                    response.version += 1;
+                } else {
+                    response.id += 1;
+                }
+                write_frame(&mut server, &response).unwrap();
+            });
+            let error = Connection { stream, id: 0 }
+                .get(|| panic!("must not decode an unrelated response"))
+                .unwrap_err();
+            assert_eq!(error, "Unexpected theme IPC response");
+            worker.join().unwrap();
+        }
+    }
 }