InitOnStartup feature (#865)

* InitOnStartup feature
Outbox initialization unified + app initialization calls simplified
* Added tests

CHANGELOG.md

@@ -18,6 +18,7 @@ Recommendation: for ease of reading, use the following order:
   - Add confirmation step to `system compact` command
   - Add support for patterns to `system compact` to process multiple datasets at once
   - Fixed argument parsing error in `kamu system compact` command
+- Simplified organization of startup initialization code over different components
 ### Fixed
  - Broken catalog issue for server and transactional modes  
 

Cargo.toml

@@ -11,6 +11,7 @@ members = [
     "src/utils/event-sourcing",
     "src/utils/event-sourcing-macros",
     "src/utils/http-common",
+    "src/utils/init-on-startup",
     "src/utils/internal-error",
     "src/utils/kamu-cli-puppet",
     "src/utils/messaging-outbox",
@@ -101,6 +102,7 @@ enum-variants = { version = "0.204.3", path = "src/utils/enum-variants", default
 event-sourcing = { version = "0.204.3", path = "src/utils/event-sourcing", default-features = false }
 event-sourcing-macros = { version = "0.204.3", path = "src/utils/event-sourcing-macros", default-features = false }
 http-common = { version = "0.204.3", path = "src/utils/http-common", default-features = false }
+init-on-startup = { version = "0.204.3", path = "src/utils/init-on-startup", default-features = false }
 internal-error = { version = "0.204.3", path = "src/utils/internal-error", default-features = false }
 kamu-cli-puppet = { version = "0.204.3", path = "src/utils/kamu-cli-puppet", default-features = false }
 kamu-data-utils = { version = "0.204.3", path = "src/utils/data-utils", default-features = false }

src/adapter/graphql/Cargo.toml

@@ -64,6 +64,7 @@ uuid = { version = "1", default-features = false }
 [dev-dependencies]
 # TODO: Limit to mock or in-memory implementations only
 container-runtime = { workspace = true }
+init-on-startup = { workspace = true }
 messaging-outbox = { workspace = true }
 kamu-accounts-inmem = { workspace = true }
 kamu-accounts-services = { workspace = true }

src/adapter/graphql/tests/tests/test_gql_account_flow_configs.rs

@@ -707,14 +707,7 @@ impl FlowConfigHarness {
         // Init dataset with no sources
         let (catalog_anonymous, catalog_authorized) = authentication_catalogs(&catalog_base).await;
 
-        DatabaseTransactionRunner::new(catalog_authorized.clone())
-            .transactional(|transactional_catalog| async move {
-                let initializer = transactional_catalog
-                    .get_one::<DatasetOwnershipServiceInMemoryStateInitializer>()
-                    .unwrap();
-
-                initializer.eager_initialization().await
-            })
+        init_on_startup::run_startup_jobs(&catalog_authorized)
             .await
             .unwrap();
 

src/adapter/graphql/tests/tests/test_gql_data.rs

@@ -86,18 +86,7 @@ async fn create_catalog_with_local_workspace(
         b.build()
     };
 
-    DatabaseTransactionRunner::new(catalog.clone())
-        .transactional(|transactional_catalog| async move {
-            let registrator = transactional_catalog
-                .get_one::<PredefinedAccountsRegistrator>()
-                .unwrap();
-
-            registrator
-                .ensure_predefined_accounts_are_registered()
-                .await
-        })
-        .await
-        .unwrap();
+    init_on_startup::run_startup_jobs(&catalog).await.unwrap();
 
     catalog
 }

src/adapter/graphql/tests/utils/auth_utils.rs

@@ -7,7 +7,6 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-use database_common::DatabaseTransactionRunner;
 use kamu_accounts::*;
 use kamu_accounts_inmem::InMemoryAccountRepository;
 use kamu_accounts_services::{LoginPasswordAuthProvider, PredefinedAccountsRegistrator};
@@ -45,16 +44,7 @@ pub async fn authentication_catalogs(
         .add_value(predefined_accounts_config)
         .build();
 
-    DatabaseTransactionRunner::new(catalog_authorized.clone())
-        .transactional(|transactional_catalog| async move {
-            let registrator = transactional_catalog
-                .get_one::<PredefinedAccountsRegistrator>()
-                .unwrap();
-
-            registrator
-                .ensure_predefined_accounts_are_registered()
-                .await
-        })
+    init_on_startup::run_startup_jobs(&catalog_authorized)
         .await
         .unwrap();
 

src/adapter/http/Cargo.toml

@@ -89,6 +89,7 @@ uuid = { version = "1", default-features = false, features = ["v4"] }
 
 [dev-dependencies]
 container-runtime = { workspace = true }
+init-on-startup = { workspace = true }
 kamu-accounts-inmem = { workspace = true }
 kamu-accounts-services = { workspace = true }
 kamu-datasets-services = { workspace = true }

src/adapter/http/tests/tests/test_platform_login_validate.rs

@@ -79,18 +79,7 @@ impl Harness {
             b.build()
         };
 
-        DatabaseTransactionRunner::new(catalog.clone())
-            .transactional(|transactional_catalog| async move {
-                let registrator = transactional_catalog
-                    .get_one::<PredefinedAccountsRegistrator>()
-                    .unwrap();
-
-                registrator
-                    .ensure_predefined_accounts_are_registered()
-                    .await
-            })
-            .await
-            .unwrap();
+        init_on_startup::run_startup_jobs(&catalog).await.unwrap();
 
         let system_time_source_stub = catalog.get_one::<SystemTimeSourceStub>().unwrap();
 

src/adapter/http/tests/tests/test_upload_local.rs

@@ -11,7 +11,7 @@ use std::net::SocketAddr;
 use std::ops::Add;
 use std::path::{Path, PathBuf};
 
-use database_common::{DatabaseTransactionRunner, NoOpDatabasePlugin};
+use database_common::NoOpDatabasePlugin;
 use internal_error::{InternalError, ResultIntoInternal};
 use kamu::domain::{CacheDir, ServerUrlConfig};
 use kamu_accounts::{JwtAuthenticationConfig, PredefinedAccountsConfig, DEFAULT_ACCOUNT_ID};
@@ -75,18 +75,7 @@ impl Harness {
             b.build()
         };
 
-        DatabaseTransactionRunner::new(catalog.clone())
-            .transactional(|transactional_catalog| async move {
-                let registrator = transactional_catalog
-                    .get_one::<PredefinedAccountsRegistrator>()
-                    .unwrap();
-
-                registrator
-                    .ensure_predefined_accounts_are_registered()
-                    .await
-            })
-            .await
-            .unwrap();
+        init_on_startup::run_startup_jobs(&catalog).await.unwrap();
 
         let authentication_service = catalog.get_one::<AuthenticationServiceImpl>().unwrap();
         let access_token = authentication_service

src/adapter/http/tests/tests/test_upload_s3.rs

@@ -9,7 +9,7 @@
 
 use std::net::SocketAddr;
 
-use database_common::{DatabaseTransactionRunner, NoOpDatabasePlugin};
+use database_common::NoOpDatabasePlugin;
 use dill::Component;
 use http::{HeaderMap, HeaderName, HeaderValue};
 use internal_error::{InternalError, ResultIntoInternal};
@@ -73,18 +73,7 @@ impl Harness {
             b.build()
         };
 
-        DatabaseTransactionRunner::new(catalog.clone())
-            .transactional(|transactional_catalog| async move {
-                let registrator = transactional_catalog
-                    .get_one::<PredefinedAccountsRegistrator>()
-                    .unwrap();
-
-                registrator
-                    .ensure_predefined_accounts_are_registered()
-                    .await
-            })
-            .await
-            .unwrap();
+        init_on_startup::run_startup_jobs(&catalog).await.unwrap();
 
         let authentication_service = catalog.get_one::<AuthenticationServiceImpl>().unwrap();
         let access_token = authentication_service

src/app/cli/Cargo.toml

@@ -46,6 +46,7 @@ container-runtime = { workspace = true }
 database-common = { workspace = true }
 database-common-macros = { workspace = true }
 http-common = { workspace = true }
+init-on-startup = { workspace = true }
 internal-error = { workspace = true }
 observability = { workspace = true, features = ["prometheus"] }
 time-source = { workspace = true }