Private Datasets: absorb helpful commits from command updates (#1016)

* E2E: added the ability to create an account using CLI

* OutboxImmediateImpl::post_message_as_json(): return a dispatch error, if present

CHANGELOG.md

@@ -26,6 +26,7 @@ Recommendation: for ease of reading, use the following order:
   - `DatasetOwnershipService`: moved to the `kamu-dataset` crate area & implemented via `DatasetEntryServiceImpl`
   - GQL, `DatasetMetadata.currentUpstreamDependencies`: indication if datasets not found/not accessed
   - GQL, `DatasetMetadata.currentDownstreamDependencies`: exclude datasets that cannot be accessed
+  - E2E: added the ability to create an account using CLI
 
 ## [0.213.1] - 2024-12-18
 ### Fixed

src/app/cli/src/cli.rs

@@ -1405,7 +1405,10 @@ pub struct SystemDiagnose {}
 #[derive(Debug, clap::Args)]
 #[command(hide = true)]
 pub struct SystemE2e {
-    #[arg(long, value_name = "ACT", value_parser = ["get-last-data-block-path"])]
+    #[arg()]
+    pub arguments: Option<Vec<String>>,
+
+    #[arg(long, value_name = "ACT", value_parser = ["get-last-data-block-path", "account-add"])]
     pub action: String,
 
     /// Local dataset reference

src/app/cli/src/cli_commands.rs

@@ -455,8 +455,11 @@ pub fn get_command(
             }
             cli::SystemSubCommand::E2e(sc) => Box::new(SystemE2ECommand::new(
                 sc.action,
+                sc.arguments.unwrap_or_default(),
                 sc.dataset,
                 cli_catalog.get_one()?,
+                cli_catalog.get_one()?,
+                cli_catalog.get_one()?,
             )),
             cli::SystemSubCommand::Gc(_) => Box::new(GcCommand::new(cli_catalog.get_one()?)),
             cli::SystemSubCommand::GenerateToken(sc) => Box::new(GenerateTokenCommand::new(

src/app/cli/src/commands/system_e2e_command.rs

@@ -9,33 +9,44 @@
 
 use std::sync::Arc;
 
-use internal_error::ResultIntoInternal;
+use internal_error::{ErrorIntoInternal, ResultIntoInternal};
 use kamu::domain::{DatasetRegistry, DatasetRegistryExt, MetadataChainExt};
-use opendatafabric::DatasetRef;
+use kamu_accounts::{AccountConfig, AccountRepository, PROVIDER_PASSWORD};
+use kamu_accounts_services::LoginPasswordAuthProvider;
+use opendatafabric as odf;
 
 use super::{CLIError, Command};
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 pub struct SystemE2ECommand {
     action: String,
-    dataset_ref: Option<DatasetRef>,
+    arguments: Vec<String>,
+    dataset_ref: Option<odf::DatasetRef>,
     dataset_registry: Arc<dyn DatasetRegistry>,
+    account_repo: Arc<dyn AccountRepository>,
+    login_password_auth_provider: Arc<LoginPasswordAuthProvider>,
 }
 
 impl SystemE2ECommand {
     pub fn new<S>(
         action: S,
-        dataset_ref: Option<DatasetRef>,
+        arguments: Vec<String>,
+        dataset_ref: Option<odf::DatasetRef>,
         dataset_registry: Arc<dyn DatasetRegistry>,
+        account_repo: Arc<dyn AccountRepository>,
+        login_password_auth_provider: Arc<LoginPasswordAuthProvider>,
     ) -> Self
     where
         S: Into<String>,
     {
         Self {
             action: action.into(),
+            arguments,
             dataset_ref,
             dataset_registry,
+            account_repo,
+            login_password_auth_provider,
         }
     }
 }
@@ -76,6 +87,29 @@ impl Command for SystemE2ECommand {
 
                 println!("{}", path.display());
             }
+            "account-add" => {
+                if self.arguments.is_empty() {
+                    return Err("Account names have not been provided".int_err().into());
+                };
+
+                for account_name in &self.arguments {
+                    eprint!("Add {account_name}... ");
+
+                    let account_config =
+                        AccountConfig::from_name(odf::AccountName::new_unchecked(account_name));
+                    let account = (&account_config).into();
+
+                    self.account_repo.create_account(&account).await.int_err()?;
+
+                    if account_config.provider == PROVIDER_PASSWORD {
+                        self.login_password_auth_provider
+                            .save_password(&account.account_name, account_config.get_password())
+                            .await?;
+                    }
+
+                    eprintln!("{}", console::style("Done").green());
+                }
+            }
             unexpected_action => panic!("Unexpected action: '{unexpected_action}'"),
         }
 

src/app/cli/src/services/accounts/account_service.rs

@@ -46,7 +46,7 @@ impl AccountService {
                         default_account_name
                     } else {
                         // Use account as username, when there is no data
-                        account.clone()
+                        account
                     },
                     true,
                 )

src/domain/accounts/domain/src/entities/account.rs

@@ -12,6 +12,8 @@ use lazy_static::lazy_static;
 use opendatafabric::{AccountID, AccountName};
 use serde::{Deserialize, Serialize};
 
+use crate::AccountConfig;
+
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
 // TODO: have some length restrictions (0 < .. < limit)
@@ -48,6 +50,25 @@ pub struct Account {
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
+impl From<&AccountConfig> for Account {
+    fn from(account_config: &AccountConfig) -> Self {
+        Account {
+            id: account_config.get_id(),
+            account_name: account_config.account_name.clone(),
+            email: account_config.email.clone(),
+            display_name: account_config.get_display_name(),
+            account_type: account_config.account_type,
+            avatar_url: account_config.avatar_url.clone(),
+            registered_at: account_config.registered_at,
+            is_admin: account_config.is_admin,
+            provider: account_config.provider.clone(),
+            provider_identity_key: account_config.account_name.to_string(),
+        }
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(
     feature = "sqlx",

src/domain/accounts/services/src/predefined_accounts_registrator.rs

@@ -49,18 +49,7 @@ impl PredefinedAccountsRegistrator {
         &self,
         account_config: &AccountConfig,
     ) -> Result<(), InternalError> {
-        let account = Account {
-            id: account_config.get_id(),
-            account_name: account_config.account_name.clone(),
-            email: account_config.email.clone(),
-            display_name: account_config.get_display_name(),
-            account_type: account_config.account_type,
-            avatar_url: account_config.avatar_url.clone(),
-            registered_at: account_config.registered_at,
-            is_admin: account_config.is_admin,
-            provider: account_config.provider.clone(),
-            provider_identity_key: account_config.account_name.to_string(),
-        };
+        let account = account_config.into();
 
         self.account_repository
             .create_account(&account)

src/utils/messaging-outbox/src/services/implementation/outbox_dispatching_impl.rs

@@ -87,14 +87,14 @@ impl Outbox for OutboxDispatchingImpl {
     ) -> Result<(), InternalError> {
         tracing::debug!(content_json = %content_json, "Dispatching outbox message");
 
-        if self.transactional_producers.contains(producer_name) {
-            self.transactional_outbox
+        if self.immediate_producers.contains(producer_name) {
+            self.immediate_outbox
                 .post_message_as_json(producer_name, content_json, version)
                 .await?;
         }
 
-        if self.immediate_producers.contains(producer_name) {
-            self.immediate_outbox
+        if self.transactional_producers.contains(producer_name) {
+            self.transactional_outbox
                 .post_message_as_json(producer_name, content_json, version)
                 .await?;
         }

src/utils/messaging-outbox/src/services/implementation/outbox_immediate_impl.rs

@@ -59,14 +59,15 @@ impl Outbox for OutboxImmediateImpl {
             let dispatch_result = dispatcher
                 .dispatch_message(&self.catalog, self.consumer_filter, &content_json, version)
                 .await;
-            if let Err(e) = dispatch_result {
+            if let Err(e) = &dispatch_result {
                 tracing::error!(
                     error = ?e,
                     error_msg = %e,
                     producer_name,
                     ?content_json,
                     "Immediate outbox message dispatching failed"
                 );
+                return dispatch_result;
             }
         }