Adapt product registration to the new architecture

rust/agama-lib/src/product.rs

@@ -1,10 +1,10 @@
 //! Implements support for handling the product settings
 
 mod client;
-mod proxies;
+pub mod proxies;
 mod settings;
 mod store;
 
-pub use client::{Product, ProductClient};
+pub use client::{Product, ProductClient, RegistrationRequirement};
 pub use settings::ProductSettings;
 pub use store::ProductStore;

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

@@ -2,7 +2,7 @@ use std::collections::HashMap;
 
 use crate::error::ServiceError;
 use crate::software::proxies::SoftwareProductProxy;
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 use zbus::Connection;
 
 use super::proxies::RegistrationProxy;
@@ -18,6 +18,35 @@ pub struct Product {
     pub description: String,
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize, utoipa::ToSchema)]
+pub enum RegistrationRequirement {
+    /// Product does not require registration
+    NotRequired = 0,
+    /// Product has optional registration
+    Optional = 1,
+    /// It is mandatory to register product
+    Mandatory = 2,
+}
+
+impl TryFrom<u32> for RegistrationRequirement {
+    type Error = ();
+
+    fn try_from(v: u32) -> Result<Self, Self::Error> {
+        match v {
+            x if x == RegistrationRequirement::NotRequired as u32 => {
+                Ok(RegistrationRequirement::NotRequired)
+            }
+            x if x == RegistrationRequirement::Optional as u32 => {
+                Ok(RegistrationRequirement::Optional)
+            }
+            x if x == RegistrationRequirement::Mandatory as u32 => {
+                Ok(RegistrationRequirement::Mandatory)
+            }
+            _ => Err(()),
+        }
+    }
+}
+
 /// D-Bus client for the software service
 #[derive(Clone)]
 pub struct ProductClient<'a> {
@@ -86,6 +115,13 @@ impl<'a> ProductClient<'a> {
         Ok(self.registration_proxy.email().await?)
     }
 
+    pub async fn registration_requirement(&self) -> Result<RegistrationRequirement, ServiceError> {
+        let requirement = self.registration_proxy.requirement().await?;
+        // unknown number can happen only if we do programmer mistake
+        let result: RegistrationRequirement = requirement.try_into().unwrap();
+        Ok(result)
+    }
+
     /// register product
     pub async fn register(&self, code: &str, email: &str) -> Result<(u32, String), ServiceError> {
         let mut options: HashMap<&str, zbus::zvariant::Value> = HashMap::new();
@@ -94,4 +130,9 @@ impl<'a> ProductClient<'a> {
         }
         Ok(self.registration_proxy.register(code, options).await?)
     }
+
+    /// de-register product
+    pub async fn deregister(&self) -> Result<(u32, String), ServiceError> {
+        Ok(self.registration_proxy.deregister().await?)
+    }
 }

rust/agama-server/src/software.rs

@@ -1,2 +1,2 @@
 pub mod web;
-pub use web::{software_service, software_stream};
+pub use web::{software_service, software_streams};

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

@@ -8,13 +8,13 @@
 use crate::{
     error::Error,
     web::{
-        common::{issues_router, progress_router, service_status_router},
+        common::{issues_router, progress_router, service_status_router, Streams},
         Event,
     },
 };
 use agama_lib::{
     error::ServiceError,
-    product::{Product, ProductClient},
+    product::{proxies::RegistrationProxy, Product, ProductClient, RegistrationRequirement},
     software::{
         proxies::{Software1Proxy, SoftwareProductProxy},
         Pattern, SelectedBy, SoftwareClient, UnknownSelectedBy,
@@ -26,7 +26,7 @@ use axum::{
     Json, Router,
 };
 use serde::{Deserialize, Serialize};
-use std::{collections::HashMap, pin::Pin};
+use std::collections::HashMap;
 use tokio_stream::{Stream, StreamExt};
 
 #[derive(Clone)]
@@ -49,14 +49,31 @@ pub struct SoftwareConfig {
 /// It emits the Event::ProductChanged and Event::PatternsChanged events.
 ///
 /// * `connection`: D-Bus connection to listen for events.
-pub async fn software_stream(
-    dbus: zbus::Connection,
-) -> Result<Pin<Box<dyn Stream<Item = Event> + Send>>, Error> {
-    let stream = StreamExt::merge(
-        product_changed_stream(dbus.clone()).await?,
-        patterns_changed_stream(dbus.clone()).await?,
-    );
-    Ok(Box::pin(stream))
+pub async fn software_streams(dbus: zbus::Connection) -> Result<Streams, Error> {
+    let result: Streams = vec![
+        (
+            "patterns_changed",
+            Box::pin(patterns_changed_stream(dbus.clone()).await?),
+        ),
+        (
+            "product_changed",
+            Box::pin(product_changed_stream(dbus.clone()).await?),
+        ),
+        (
+            "registration_requirement_changed",
+            Box::pin(registration_requirement_changed_stream(dbus.clone()).await?),
+        ),
+        (
+            "registration_code_changed",
+            Box::pin(registration_code_changed_stream(dbus.clone()).await?),
+        ),
+        (
+            "registration_email_changed",
+            Box::pin(registration_email_changed_stream(dbus.clone()).await?),
+        ),
+    ];
+
+    Ok(result)
 }
 
 async fn product_changed_stream(
@@ -99,6 +116,62 @@ async fn patterns_changed_stream(
     Ok(stream)
 }
 
+async fn registration_requirement_changed_stream(
+    dbus: zbus::Connection,
+) -> Result<impl Stream<Item = Event>, Error> {
+    // TODO: move registration requirement to product in dbus and so just one event will be needed.
+    let proxy = RegistrationProxy::new(&dbus).await?;
+    let stream = proxy
+        .receive_requirement_changed()
+        .await
+        .then(|change| async move {
+            if let Ok(id) = change.get().await {
+                // unwrap is safe as possible numbers is send by our controlled dbus
+                return Some(Event::RegistrationRequirementChanged {
+                    requirement: id.try_into().unwrap(),
+                });
+            }
+            None
+        })
+        .filter_map(|e| e);
+    Ok(stream)
+}
+
+async fn registration_email_changed_stream(
+    dbus: zbus::Connection,
+) -> Result<impl Stream<Item = Event>, Error> {
+    let proxy = RegistrationProxy::new(&dbus).await?;
+    let stream = proxy
+        .receive_email_changed()
+        .await
+        .then(|change| async move {
+            if let Ok(_id) = change.get().await {
+                // TODO: add to stream also proxy and return whole cached registration info
+                return Some(Event::RegistrationChanged);
+            }
+            None
+        })
+        .filter_map(|e| e);
+    Ok(stream)
+}
+
+async fn registration_code_changed_stream(
+    dbus: zbus::Connection,
+) -> Result<impl Stream<Item = Event>, Error> {
+    let proxy = RegistrationProxy::new(&dbus).await?;
+    let stream = proxy
+        .receive_reg_code_changed()
+        .await
+        .then(|change| async move {
+            if let Ok(_id) = change.get().await {
+                return Some(Event::RegistrationChanged);
+            }
+            None
+        })
+        .filter_map(|e| e);
+    Ok(stream)
+}
+
 // Returns a hash replacing the selection "reason" from D-Bus with a SelectedBy variant.
 fn reason_to_selected_by(
     patterns: HashMap<String, u8>,
@@ -130,6 +203,10 @@ pub async fn software_service(dbus: zbus::Connection) -> Result<Router, ServiceE
     let router = Router::new()
         .route("/patterns", get(patterns))
         .route("/products", get(products))
+        .route(
+            "/registration",
+            get(get_registration).post(register).delete(deregister),
+        )
         .route("/proposal", get(proposal))
         .route("/config", put(set_config).get(get_config))
         .route("/probe", post(probe))
@@ -153,6 +230,74 @@ async fn products(State(state): State<SoftwareState<'_>>) -> Result<Json<Vec<Pro
     Ok(Json(products))
 }
 
+/// Information about registration configuration (product, patterns, etc.).
+#[derive(Clone, Serialize, Deserialize, utoipa::ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct RegistrationInfo {
+    /// Registration key. Empty value mean key not used or not registered.
+    key: String,
+    /// Registration email. Empty value mean email not used or not registered.
+    email: String,
+    /// if registration is required, optional or not needed for current product.
+    /// Change only if selected product is changed.
+    requirement: RegistrationRequirement,
+}
+
+/// returns registration info
+///
+/// * `state`: service state.
+#[utoipa::path(get, path = "/software/registration", responses(
+    (status = 200, description = "registration configuration", body = RegistrationInfo),
+    (status = 400, description = "The D-Bus service could not perform the action")
+))]
+async fn get_registration(
+    State(state): State<SoftwareState<'_>>,
+) -> Result<Json<RegistrationInfo>, Error> {
+    let result = RegistrationInfo {
+        key: state.product.registration_code().await?,
+        email: state.product.email().await?,
+        requirement: state.product.registration_requirement().await?,
+    };
+    Ok(Json(result))
+}
+
+/// Software service configuration (product, patterns, etc.).
+#[derive(Clone, Serialize, Deserialize, utoipa::ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct RegistrationParams {
+    /// Registration key.
+    key: String,
+    /// Registration email.
+    email: String,
+}
+
+/// Register product
+///
+/// * `state`: service state.
+#[utoipa::path(post, path = "/software/registration", responses(
+    (status = 200, description = "registration successfull"),
+    (status = 400, description = "The D-Bus service could not perform the action")
+))]
+async fn register(
+    State(state): State<SoftwareState<'_>>,
+    Json(config): Json<RegistrationParams>,
+) -> Result<Json<()>, Error> {
+    state.product.register(&config.key, &config.email).await?;
+    Ok(Json(()))
+}
+
+/// Deregister product
+///
+/// * `state`: service state.
+#[utoipa::path(delete, path = "/software/registration", responses(
+    (status = 200, description = "deregistration successfull"),
+    (status = 400, description = "The D-Bus service could not perform the action")
+))]
+async fn deregister(State(state): State<SoftwareState<'_>>) -> Result<Json<()>, Error> {
+    state.product.deregister().await?;
+    Ok(Json(()))
+}
+
 /// Returns the list of software patterns.
 ///
 /// * `state`: service state.

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

@@ -7,7 +7,7 @@
 use crate::{
     error::Error,
     web::{
-        common::{service_status_router, validation_router},
+        common::{service_status_router, validation_router, Streams},
         Event,
     },
 };
@@ -30,16 +30,14 @@ struct UsersState<'a> {
 /// It emits the Event::RootPasswordChange, Event::RootSSHKeyChanged and Event::FirstUserChanged events.
 ///
 /// * `connection`: D-Bus connection to listen for events.
-pub async fn users_streams(
-    dbus: zbus::Connection,
-) -> Result<Vec<(&'static str, Pin<Box<dyn Stream<Item = Event> + Send>>)>, Error> {
+pub async fn users_streams(dbus: zbus::Connection) -> Result<Streams, Error> {
     const FIRST_USER_ID: &str = "first_user";
     const ROOT_PASSWORD_ID: &str = "root_password";
     const ROOT_SSHKEY_ID: &str = "root_sshkey";
     // here we have three streams, but only two events. Reason is
     // that we have three streams from dbus about property change
     // and unify two root user properties into single event to http API
-    let result: Vec<(&str, Pin<Box<dyn Stream<Item = Event> + Send>>)> = vec![
+    let result: Streams = vec![
         (
             FIRST_USER_ID,
             Box::pin(first_user_changed_stream(dbus.clone()).await?),

rust/agama-server/src/web.rs

@@ -10,7 +10,7 @@ use crate::{
     manager::web::{manager_service, manager_stream},
     network::{web::network_service, NetworkManagerAdapter},
     questions::web::{questions_service, questions_stream},
-    software::web::{software_service, software_stream},
+    software::web::{software_service, software_streams},
     users::web::{users_service, users_streams},
     web::common::{issues_stream, progress_stream, service_status_stream},
 };
@@ -110,7 +110,9 @@ async fn run_events_monitor(dbus: zbus::Connection, events: EventsSender) -> Res
     for (id, user_stream) in users_streams(dbus.clone()).await? {
         stream.insert(id, user_stream);
     }
-    stream.insert("software", software_stream(dbus.clone()).await?);
+    for (id, software_stream) in software_streams(dbus.clone()).await? {
+        stream.insert(id, software_stream);
+    }
     stream.insert(
         "software-status",
         service_status_stream(

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

@@ -17,6 +17,8 @@ use crate::error::Error;
 
 use super::Event;
 
+pub type Streams = Vec<(&'static str, Pin<Box<dyn Stream<Item = Event> + Send>>)>;
+
 /// Builds a router to the `org.opensuse.Agama1.ServiceStatus` interface of the
 /// given D-Bus object.
 ///

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

@@ -1,6 +1,7 @@
 use crate::l10n::web::LocaleConfig;
 use agama_lib::{
-    manager::InstallationPhase, progress::Progress, software::SelectedBy, users::FirstUser,
+    manager::InstallationPhase, product::RegistrationRequirement, progress::Progress,
+    software::SelectedBy, users::FirstUser,
 };
 use serde::Serialize;
 use std::collections::HashMap;
@@ -23,6 +24,10 @@ pub enum Event {
     ProductChanged {
         id: String,
     },
+    RegistrationRequirementChanged {
+        requirement: RegistrationRequirement,
+    },
+    RegistrationChanged,
     FirstUserChanged(FirstUser),
     RootChanged {
         password: Option<bool>,