feat: Various minor improvements

- Exit early if no plugin initializes successfully.
- Moved the log CLI arguments to the bottom and simplified argument help messages.

- Further `TaskTracker` adoption.
- Moved plugin `Uuid` into `AvailablePlugin`.
- Moved service registration functions into service sub modules.
  - Temporarily removed the old deregistration implementation.
- Naming and ordering consistency.

Author: celarye

src/cli.rs

@@ -24,13 +24,10 @@ impl From<String> for CliLogParametersFileRotation {
 #[derive(Parser)]
 #[command(about, long_about = None, version, author)]
 pub struct Cli {
-    #[command(flatten)]
-    pub log_parameters: CliLogParameters,
-
     #[arg(default_value = "./config.yaml", short, long, value_name = "FILE PATH", help = "The path to the program its configuration file", long_help = None)]
     pub config_file: PathBuf,
 
-    #[arg(default_value = "./.env", short, long, value_name = "FILE PATH", help = "The path to an env file, used by the program its configuration file for env var interpolation", long_help = None)]
+    #[arg(default_value = "./.env", short, long, value_name = "FILE PATH", help = "The path to the program its env file", long_help = None)]
     pub env_file: PathBuf,
 
     #[arg(default_value = "./plugins", short, long, value_name = "DIRECTORY PATH", help = "The path to the program its plugin directory", long_help = None)]
@@ -41,6 +38,9 @@ pub struct Cli {
 
     #[arg(default_value_t = false, short, long, help = "Run in restricted mode, in this case plugin permissions are opt in", long_help = None)]
     pub restricted: bool,
+
+    #[command(flatten)]
+    pub log_parameters: CliLogParameters,
 }
 
 #[derive(Args)]

src/config.rs

@@ -6,6 +6,7 @@ use std::{collections::HashMap, fs, path::Path};
 use anyhow::Result;
 use serde::Deserialize;
 use tracing::info;
+use uuid::Uuid;
 
 use crate::config::{plugins::ConfigPlugin, services::ConfigServices};
 
@@ -14,13 +15,18 @@ pub mod services;
 
 #[derive(Deserialize)]
 pub struct Config {
+    #[serde(default = "Config::default_name")]
     pub name: String,
     #[serde(default)]
     pub services: ConfigServices,
     pub plugins: HashMap<String, ConfigPlugin>,
 }
 
 impl Config {
+    fn default_name() -> String {
+        Uuid::new_v4().to_string()
+    }
+
     pub fn new(file_path: &Path, restricted: bool) -> Result<Self> {
         info!("Loading and parsing the config file");
 

src/main.rs

@@ -101,21 +101,21 @@ async fn main() -> Result<ExitCode> {
     let database = database::new(&cli.database_directory)?;
 
     let message_handler = message_handler(
-        database.clone(),
         Arc::new(RwLock::new(Some(channels.core.runtime_tx))),
         Arc::new(RwLock::new(channels.core.job_scheduler_tx)),
         Arc::new(RwLock::new(channels.core.discord_tx)),
-        Arc::new(shutdown_signal_listener),
         channels.core.rx,
+        database.clone(),
+        Arc::new(shutdown_signal_listener),
     );
 
     let setup_result = setup(
         cli.plugin_directory,
-        database,
-        channels.services,
-        channels.runtime,
         config,
         secrets,
+        channels.services,
+        channels.runtime,
+        database,
     )
     .await;
 
@@ -129,18 +129,23 @@ async fn main() -> Result<ExitCode> {
 }
 
 fn message_handler(
-    database: Database,
     runtime_tx: Arc<RwLock<Option<UnboundedSender<RuntimeMessages>>>>,
     job_scheduler_tx: Arc<RwLock<Option<UnboundedSender<JobSchedulerMessages>>>>,
     discord_tx: Arc<RwLock<Option<UnboundedSender<DiscordMessages>>>>,
-    shutdown_signal_listener: Arc<JoinHandle<()>>,
     mut rx: UnboundedReceiver<CoreMessages>,
+    database: Database,
+    shutdown_signal_listener: Arc<JoinHandle<()>>,
 ) -> JoinHandle<Result<()>> {
     debug!("Starting the message handler");
 
     tokio::spawn(async move {
         while let Some(core_message) = rx.recv().await {
             match core_message {
+                CoreMessages::Runtime(runtime_message) => {
+                    if let Some(runtime_tx) = runtime_tx.read().await.as_ref() {
+                        runtime_tx.send(runtime_message).unwrap();
+                    }
+                }
                 CoreMessages::JobScheduler(job_scheduler_message) => {
                     if let Some(job_scheduler_tx) = job_scheduler_tx.read().await.as_ref() {
                         job_scheduler_tx.send(job_scheduler_message).unwrap();
@@ -151,18 +156,13 @@ fn message_handler(
                         discord_tx.send(discord_message).unwrap();
                     }
                 }
-                CoreMessages::Runtime(runtime_message) => {
-                    if let Some(runtime_tx) = runtime_tx.read().await.as_ref() {
-                        runtime_tx.send(runtime_message).unwrap();
-                    }
-                }
                 CoreMessages::Shutdown(shutdown_kind) => {
                     tokio::spawn(shutdown(
-                        shutdown_kind,
                         runtime_tx.clone(),
                         job_scheduler_tx.clone(),
                         discord_tx.clone(),
                         shutdown_signal_listener.clone(),
+                        shutdown_kind,
                     ));
                 }
             }
@@ -174,27 +174,27 @@ fn message_handler(
 
 async fn setup(
     plugin_directory_path: PathBuf,
-    database: Database,
-    service_channels: ChannelsServices,
-    runtime_channels: ChannelsRuntime,
     config: Config,
     secrets: Secrets,
+    service_channels: ChannelsServices,
+    runtime_channels: ChannelsRuntime,
+    database: Database,
 ) -> Result<()> {
     let config_name = Arc::new(config.name);
 
     let available_plugins = registry::get_plugins(
         &plugin_directory_path,
-        database.clone(),
         config_name.clone(),
         config.plugins,
+        database.clone(),
     )
     .await?;
 
     services::setup(
         config.services,
         secrets.services,
-        database.clone(),
         service_channels,
+        database.clone(),
     )
     .await?;
 
@@ -204,9 +204,9 @@ async fn setup(
         .initialize_plugins(
             plugin_directory_path,
             config_name,
-            available_plugins,
-            database,
             runtime_channels.core_tx,
+            database,
+            available_plugins,
         )
         .await?;
 
@@ -293,11 +293,11 @@ fn shutdown_signal_listener(core_tx: UnboundedSender<CoreMessages>) -> JoinHandl
 }
 
 async fn shutdown(
-    shutdown_kind: Shutdown,
     runtime_tx: Arc<RwLock<Option<UnboundedSender<RuntimeMessages>>>>,
     job_scheduler_tx: Arc<RwLock<Option<UnboundedSender<JobSchedulerMessages>>>>,
     discord_tx: Arc<RwLock<Option<UnboundedSender<DiscordMessages>>>>,
     shutdown_signal_listener: Arc<JoinHandle<()>>,
+    shutdown_kind: Shutdown,
 ) {
     let mut shutdown_guard = SHUTDOWN.write().await;
 

src/registry.rs

@@ -27,18 +27,18 @@ static DEFAULT_NAMESPACE_ID: &str = "wpbs-rs";
 #[hotpath::measure]
 pub async fn get_plugins(
     plugin_directory_path: &Path,
-    database: Database,
     config_name: Arc<String>,
     config_plugins: HashMap<String, ConfigPlugin>,
-) -> Result<Vec<(Uuid, AvailablePlugin)>> {
+    database: Database,
+) -> Result<Vec<AvailablePlugin>> {
     info!("Getting all plugins from their respective registries");
 
     let caching_client =
         create_registry_client(&plugin_directory_path.join("binaries").join("remote")).await?;
 
     let mut available_plugins = Vec::new();
 
-    let mut plugin_tasks: Vec<JoinHandle<Result<(Uuid, AvailablePlugin)>>> = Vec::new();
+    let mut plugin_tasks: Vec<JoinHandle<Result<AvailablePlugin>>> = Vec::new();
 
     let plugins_keyspace = database.keyspace("plugins", KeyspaceCreateOptions::default)?;
 
@@ -57,41 +57,38 @@ pub async fn get_plugins(
             if namespace_id == "local" {
                 get_local_plugin(plugin_directory_path, &plugin_id, &plugin_version).await?;
 
-                let uuid = get_plugin_uuid(&plugins_keyspace, &config_name, &plugin_user_id)?;
-
-                return Ok((
-                    uuid,
-                    AvailablePlugin {
-                        namespace_id,
-                        plugin_id,
-                        version: plugin_version,
-                        content_digest: None,
-                        user_id: plugin_user_id,
-                        environment: plugin_config.environment,
-                        permissions: plugin_config.permissions,
-                        settings: plugin_config.settings,
-                    },
-                ));
-            }
-
-            let release =
-                fetch_plugin(caching_client, &namespace_id, &plugin_id, &plugin_version).await?;
-
-            let uuid = get_plugin_uuid(&plugins_keyspace, &config_name, &plugin_user_id)?;
+                let plugin_uuid =
+                    get_plugin_uuid(&plugins_keyspace, &config_name, &plugin_user_id)?;
 
-            Ok((
-                uuid,
-                AvailablePlugin {
+                return Ok(AvailablePlugin {
+                    plugin_uuid,
                     namespace_id,
                     plugin_id,
-                    version: release.version,
-                    content_digest: Some(release.content_digest),
+                    version: plugin_version,
+                    content_digest: None,
                     user_id: plugin_user_id,
                     permissions: plugin_config.permissions,
                     environment: plugin_config.environment,
                     settings: plugin_config.settings,
-                },
-            ))
+                });
+            }
+
+            let release =
+                fetch_plugin(caching_client, &namespace_id, &plugin_id, &plugin_version).await?;
+
+            let plugin_uuid = get_plugin_uuid(&plugins_keyspace, &config_name, &plugin_user_id)?;
+
+            Ok(AvailablePlugin {
+                plugin_uuid,
+                namespace_id,
+                plugin_id,
+                version: release.version,
+                content_digest: Some(release.content_digest),
+                user_id: plugin_user_id,
+                permissions: plugin_config.permissions,
+                environment: plugin_config.environment,
+                settings: plugin_config.settings,
+            })
         }));
     }
 

src/registry/plugins.rs

@@ -4,12 +4,14 @@
 use std::collections::HashMap;
 
 use semver::Version;
+use uuid::Uuid;
 use wasm_pkg_client::ContentDigest;
 use yaml_serde::Value;
 
 use crate::config::plugins::permissions::PluginPermissions;
 
 pub struct AvailablePlugin {
+    pub plugin_uuid: Uuid,
     pub namespace_id: String,
     pub plugin_id: String,
     pub version: Version,