fix(ui): detect out of sync storage

rust/agama-lib/src/auth.rs

@@ -45,7 +45,7 @@ const USER_TOKEN_PATH: &str = ".local/agama/token";
 const AGAMA_TOKEN_FILE: &str = "/run/agama/token";
 
 use std::{
-    fmt::Display,
+    fmt,
     fs::{self, File},
     io::{self, BufRead, BufReader, Write},
     os::unix::fs::OpenOptionsExt,
@@ -184,8 +184,8 @@ impl AuthToken {
     }
 }
 
-impl Display for AuthToken {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+impl fmt::Display for AuthToken {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "{}", self.0)
     }
 }
@@ -223,6 +223,16 @@ impl ClientId {
     pub fn new() -> Self {
         ClientId(Uuid::new_v4())
     }
+
+    pub fn new_from_uuid(uuid: Uuid) -> Self {
+        ClientId(uuid)
+    }
+}
+
+impl fmt::Display for ClientId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
 }
 
 #[cfg(test)]

rust/agama-lib/src/http/event.rs

@@ -103,6 +103,7 @@ pub enum EventPayload {
         #[serde(flatten)]
         change: NetworkChange,
     },
+    StorageChanged,
     // TODO: it should include the full software proposal or, at least,
     // all the relevant changes.
     SoftwareProposalChanged {

rust/agama-lib/src/manager.rs

@@ -23,6 +23,7 @@
 pub mod http_client;
 pub use http_client::ManagerHTTPClient;
 use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
 
 use crate::error::ServiceError;
 use crate::proxies::ServiceStatusProxy;
@@ -141,15 +142,21 @@ impl<'a> ManagerClient<'a> {
     }
 
     /// Starts the probing process.
-    pub async fn probe(&self) -> Result<(), ServiceError> {
+    pub async fn probe(&self, client_id: String) -> Result<(), ServiceError> {
         self.wait().await?;
-        Ok(self.manager_proxy.probe().await?)
+        Ok(self
+            .manager_proxy
+            .probe(HashMap::from([("client_id", &client_id.into())]))
+            .await?)
     }
 
     /// Starts the reprobing process.
-    pub async fn reprobe(&self) -> Result<(), ServiceError> {
+    pub async fn reprobe(&self, client_id: String) -> Result<(), ServiceError> {
         self.wait().await?;
-        Ok(self.manager_proxy.reprobe().await?)
+        Ok(self
+            .manager_proxy
+            .reprobe(HashMap::from([("client_id", &client_id.into())]))
+            .await?)
     }
 
     /// Starts the installation.

rust/agama-lib/src/proxies/manager1.rs

@@ -39,10 +39,16 @@ pub trait Manager1 {
     fn finish(&self, method: &str) -> zbus::Result<bool>;
 
     /// Probe method
-    fn probe(&self) -> zbus::Result<()>;
+    fn probe(
+        &self,
+        data: std::collections::HashMap<&str, &zbus::zvariant::Value<'_>>,
+    ) -> zbus::Result<()>;
 
     /// Reprobe method
-    fn reprobe(&self) -> zbus::Result<()>;
+    fn reprobe(
+        &self,
+        data: std::collections::HashMap<&str, &zbus::zvariant::Value<'_>>,
+    ) -> zbus::Result<()>;
 
     /// BusyServices property
     #[zbus(property)]

rust/agama-lib/src/storage/client.rs

@@ -150,31 +150,50 @@ impl<'a> StorageClient<'a> {
     }
 
     /// Runs the probing process
-    pub async fn probe(&self) -> Result<(), ServiceError> {
-        Ok(self.storage_proxy.probe().await?)
+    pub async fn probe(&self, client_id: String) -> Result<(), ServiceError> {
+        Ok(self
+            .storage_proxy
+            .probe(HashMap::from([("client_id", &client_id.into())]))
+            .await?)
     }
 
     /// Runs the reprobing process
-    pub async fn reprobe(&self) -> Result<(), ServiceError> {
-        Ok(self.storage_proxy.reprobe().await?)
+    pub async fn reprobe(&self, client_id: String) -> Result<(), ServiceError> {
+        Ok(self
+            .storage_proxy
+            .reprobe(HashMap::from([("client_id", &client_id.into())]))
+            .await?)
     }
 
     /// Runs the reactivation process
-    pub async fn reactivate(&self) -> Result<(), ServiceError> {
-        Ok(self.storage_proxy.reactivate().await?)
+    pub async fn reactivate(&self, client_id: String) -> Result<(), ServiceError> {
+        Ok(self
+            .storage_proxy
+            .reactivate(HashMap::from([("client_id", &client_id.into())]))
+            .await?)
     }
 
     /// Set the storage config according to the JSON schema
-    pub async fn set_config(&self, settings: StorageSettings) -> Result<u32, ServiceError> {
+    pub async fn set_config(
+        &self,
+        settings: StorageSettings,
+        client_id: String,
+    ) -> Result<u32, ServiceError> {
         Ok(self
             .storage_proxy
-            .set_config(serde_json::to_string(&settings)?.as_str())
+            .set_config(
+                serde_json::to_string(&settings)?.as_str(),
+                HashMap::from([("client_id", &client_id.into())]),
+            )
             .await?)
     }
 
     /// Reset the storage config to the default value
-    pub async fn reset_config(&self) -> Result<u32, ServiceError> {
-        Ok(self.storage_proxy.reset_config().await?)
+    pub async fn reset_config(&self, client_id: String) -> Result<u32, ServiceError> {
+        Ok(self
+            .storage_proxy
+            .reset_config(HashMap::from([("client_id", &client_id.into())]))
+            .await?)
     }
 
     /// Get the storage config according to the JSON schema
@@ -185,10 +204,17 @@ impl<'a> StorageClient<'a> {
     }
 
     /// Set the storage config model according to the JSON schema
-    pub async fn set_config_model(&self, model: Box<RawValue>) -> Result<u32, ServiceError> {
+    pub async fn set_config_model(
+        &self,
+        model: Box<RawValue>,
+        client_id: String,
+    ) -> Result<u32, ServiceError> {
         Ok(self
             .storage_proxy
-            .set_config_model(serde_json::to_string(&model).unwrap().as_str())
+            .set_config_model(
+                serde_json::to_string(&model).unwrap().as_str(),
+                HashMap::from([("client_id", &client_id.into())]),
+            )
             .await?)
     }
 

rust/agama-lib/src/storage/proxies/storage1.rs

@@ -47,33 +47,57 @@ use zbus::proxy;
     assume_defaults = true
 )]
 pub trait Storage1 {
+    /// Storage configured signal
+    #[zbus(signal)]
+    fn configured(&self, client_id: &str) -> zbus::Result<()>;
+
     /// Finish method
     fn finish(&self) -> zbus::Result<()>;
 
     /// Install method
     fn install(&self) -> zbus::Result<()>;
 
     /// Probe method
-    fn probe(&self) -> zbus::Result<()>;
+    fn probe(
+        &self,
+        data: std::collections::HashMap<&str, &zbus::zvariant::Value<'_>>,
+    ) -> zbus::Result<()>;
 
     /// Reprobe method
-    fn reprobe(&self) -> zbus::Result<()>;
+    fn reprobe(
+        &self,
+        data: std::collections::HashMap<&str, &zbus::zvariant::Value<'_>>,
+    ) -> zbus::Result<()>;
 
     /// Reactivate method
-    fn reactivate(&self) -> zbus::Result<()>;
+    fn reactivate(
+        &self,
+        data: std::collections::HashMap<&str, &zbus::zvariant::Value<'_>>,
+    ) -> zbus::Result<()>;
 
     /// Set the storage config according to the JSON schema
-    fn set_config(&self, settings: &str) -> zbus::Result<u32>;
+    fn set_config(
+        &self,
+        settings: &str,
+        data: std::collections::HashMap<&str, &zbus::zvariant::Value<'_>>,
+    ) -> zbus::Result<u32>;
 
     /// Reset the storage config to the default value
-    fn reset_config(&self) -> zbus::Result<u32>;
+    fn reset_config(
+        &self,
+        data: std::collections::HashMap<&str, &zbus::zvariant::Value<'_>>,
+    ) -> zbus::Result<u32>;
+
+    /// Set the storage config model according to the JSON schema
+    fn set_config_model(
+        &self,
+        model: &str,
+        data: std::collections::HashMap<&str, &zbus::zvariant::Value<'_>>,
+    ) -> zbus::Result<u32>;
 
     /// Get the current storage config according to the JSON schema
     fn get_config(&self) -> zbus::Result<String>;
 
-    /// Set the storage config model according to the JSON schema
-    fn set_config_model(&self, model: &str) -> zbus::Result<u32>;
-
     /// Get the storage config model according to the JSON schema
     fn get_config_model(&self) -> zbus::Result<String>;
 

rust/agama-server/src/manager/web.rs

@@ -26,6 +26,7 @@
 //! * `manager_stream` which offers an stream that emits the manager events coming from D-Bus.
 
 use agama_lib::{
+    auth::ClientId,
     error::ServiceError,
     event, logs,
     manager::{FinishMethod, InstallationPhase, InstallerStatus, ManagerClient},
@@ -38,9 +39,11 @@ use axum::{
     http::{header, status::StatusCode, HeaderMap, HeaderValue},
     response::IntoResponse,
     routing::{get, post},
-    Json, Router,
+    Extension, Json, Router,
 };
+use std::collections::HashMap;
 use std::pin::Pin;
+use std::sync::Arc;
 use tokio_stream::{Stream, StreamExt};
 use tokio_util::io::ReaderStream;
 
@@ -129,7 +132,10 @@ pub async fn manager_service(
        )
     )
 )]
-async fn probe_action(State(state): State<ManagerState<'_>>) -> Result<(), Error> {
+async fn probe_action(
+    State(state): State<ManagerState<'_>>,
+    Extension(client_id): Extension<Arc<ClientId>>,
+) -> Result<(), Error> {
     let dbus = state.dbus.clone();
     tokio::spawn(async move {
         let result = dbus
@@ -138,7 +144,7 @@ async fn probe_action(State(state): State<ManagerState<'_>>) -> Result<(), Error
                 "/org/opensuse/Agama/Manager1",
                 Some("org.opensuse.Agama.Manager1"),
                 "Probe",
-                &(),
+                &HashMap::from([("client_id", client_id.to_string())]),
             )
             .await;
         if let Err(error) = result {
@@ -158,8 +164,11 @@ async fn probe_action(State(state): State<ManagerState<'_>>) -> Result<(), Error
       (status = 200, description = "Probing done.")
     )
 )]
-async fn probe_sync_action(State(state): State<ManagerState<'_>>) -> Result<(), Error> {
-    state.manager.probe().await?;
+async fn probe_sync_action(
+    State(state): State<ManagerState<'_>>,
+    Extension(client_id): Extension<Arc<ClientId>>,
+) -> Result<(), Error> {
+    state.manager.probe(client_id.to_string()).await?;
     Ok(())
 }
 
@@ -172,8 +181,11 @@ async fn probe_sync_action(State(state): State<ManagerState<'_>>) -> Result<(),
       (status = 200, description = "Re-probing done.")
     )
 )]
-async fn reprobe_sync_action(State(state): State<ManagerState<'_>>) -> Result<(), Error> {
-    state.manager.reprobe().await?;
+async fn reprobe_sync_action(
+    State(state): State<ManagerState<'_>>,
+    Extension(client_id): Extension<Arc<ClientId>>,
+) -> Result<(), Error> {
+    state.manager.reprobe(client_id.to_string()).await?;
     Ok(())
 }