5829 load plugins via processor

server/repository/src/db_diesel/key_value_store.rs

@@ -16,8 +16,9 @@ table! {
     }
 }
 
-// Database:  https://github.com/openmsupply/open-msupply/blob/d6645711184c63593949c3e8b6dc96b5a5ded39f/server/repository/migrations/postgres/2022-02-11T15-00_create_key_value_store/up.sql#L2-L16
+// Snippet for adding new, including migration : https://github.com/msupply-foundation/open-msupply/wiki/Snippets "New Key Type for KeyValueStore"
 #[derive(DbEnum, Debug, Clone, PartialEq, Eq, Default)]
+#[cfg_attr(test, derive(strum::EnumIter))]
 #[DbValueStyle = "SCREAMING_SNAKE_CASE"]
 pub enum KeyType {
     #[default]
@@ -28,6 +29,7 @@ pub enum KeyType {
     ShipmentTransferProcessorCursor,
     RequisitionTransferProcessorCursor,
     ContactFormProcessorCursor,
+    LoadPluginProcessorCursor,
 
     SettingsSyncUrl,
     SettingsSyncUsername,
@@ -171,3 +173,29 @@ impl<'a> KeyValueStoreRepository<'a> {
         Ok(row.and_then(|row| row.value_bool))
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use strum::IntoEnumIterator;
+    use util::assert_matches;
+
+    use crate::{mock::MockDataInserts, test_db::setup_all};
+
+    #[actix_rt::test]
+    async fn key_type_enum() {
+        let (_, connection, _, _) = setup_all("key_type_enum", MockDataInserts::none()).await;
+
+        let repo = KeyValueStoreRepository::new(&connection);
+        // Try upsert all variants, confirm that diesel enums match postgres
+        for variant in KeyType::iter() {
+            let result = repo.upsert_one(&KeyValueStoreRow {
+                id: variant.clone(),
+                ..Default::default()
+            });
+            assert_eq!(result, Ok(()));
+
+            assert_matches!(repo.get_row(variant.clone()), Ok(Some(_)));
+        }
+    }
+}

server/repository/src/migrations/v2_06_00/add_load_plugin_processor_pg_enum_type.rs

@@ -0,0 +1,22 @@
+use crate::migrations::*;
+
+pub(crate) struct Migrate;
+
+impl MigrationFragment for Migrate {
+    fn identifier(&self) -> &'static str {
+        "add_load_plugin_processor_pg_enum_type"
+    }
+
+    fn migrate(&self, connection: &StorageConnection) -> anyhow::Result<()> {
+        if cfg!(feature = "postgres") {
+            sql!(
+                connection,
+                r#"
+                    ALTER TYPE key_type ADD VALUE IF NOT EXISTS 'LOAD_PLUGIN_PROCESSOR_CURSOR';
+                "#
+            )?;
+        }
+
+        Ok(())
+    }
+}

server/repository/src/migrations/v2_06_00/mod.rs

@@ -4,6 +4,7 @@ use crate::StorageConnection;
 mod add_create_invoice_from_requisition_permission;
 mod add_index_to_sync_buffer;
 mod add_insurance_provider;
+mod add_load_plugin_processor_pg_enum_type;
 mod add_name_next_of_kin_id;
 mod add_program_deleted_datetime;
 mod add_program_id_to_invoice;
@@ -27,6 +28,7 @@ impl Migration for V2_06_00 {
             Box::new(backend_plugins::Migrate),
             Box::new(add_create_invoice_from_requisition_permission::Migrate),
             Box::new(add_name_next_of_kin_id::Migrate),
+            Box::new(add_load_plugin_processor_pg_enum_type::Migrate),
             Box::new(add_program_id_to_invoice::Migrate),
             Box::new(add_insurance_provider::Migrate),
         ]

server/server/src/lib.rs

@@ -147,11 +147,14 @@ pub async fn start_server(
     }
 
     // PLUGIN CONTEXT
-
     PluginContext {
         service_provider: service_provider.clone(),
     }
     .bind();
+    service_provider
+        .plugin_service
+        .reload_all_plugins(&service_context)
+        .unwrap();
 
     // SET LOG CALLBACK FOR WASM FUNCTIONS
     info!("Setting wasm function log callback..");

server/service/src/plugin/mod.rs

@@ -3,6 +3,7 @@ use thiserror::Error;
 
 use crate::{
     backend_plugin::plugin_provider::{PluginBundle, PluginInstance},
+    processors::ProcessorType,
     service_provider::ServiceContext,
     settings::Settings,
     UploadedFile, UploadedFileJsonError,
@@ -43,6 +44,15 @@ pub trait PluginServiceTrait: Sync + Send {
         uploaded_file.as_json_file(settings)
     }
 
+    fn reload_all_plugins(&self, ctx: &ServiceContext) -> Result<(), RepositoryError> {
+        let repo = BackendPluginRowRepository::new(&ctx.connection);
+        for row in repo.all()? {
+            PluginInstance::bind(row);
+        }
+
+        Ok(())
+    }
+
     fn install_uploaded_plugin(
         &self,
         ctx: &ServiceContext,
@@ -56,10 +66,11 @@ pub trait PluginServiceTrait: Sync + Send {
 
                 for row in plugin_bundle.backend_plugins.clone() {
                     backend_repo.upsert_one(row.clone())?;
-                    // Bind plugin to provider (this would ideally happen via processor, going over changelog)
-                    PluginInstance::bind(row)
                 }
 
+                ctx.processors_trigger
+                    .trigger_processor(ProcessorType::LoadPlugin);
+
                 Ok(plugin_bundle)
             })
             .map_err(|error| error.to_inner_error())

server/service/src/processors/general_processor.rs

@@ -11,15 +11,15 @@ use crate::{
     service_provider::{ServiceContext, ServiceProvider},
 };
 
-use super::contact_form::QueueContactEmailProcessor;
+use super::{contact_form::QueueContactEmailProcessor, load_plugin::LoadPlugin};
 
 #[derive(Error, Debug)]
 pub(crate) enum ProcessorError {
-    #[error("{0:?} not found: {1:?}")]
+    #[error("{0} not found: {1}")]
     RecordNotFound(String, String),
-    #[error("{0:?}")]
-    DatabaseError(RepositoryError),
-    #[error("{0:?}")]
+    #[error("Database error")]
+    DatabaseError(#[from] RepositoryError),
+    #[error("Error in email service {0:?}")]
     EmailServiceError(EmailServiceError),
 }
 
@@ -28,12 +28,14 @@ const CHANGELOG_BATCH_SIZE: u32 = 20;
 #[derive(Clone)]
 pub enum ProcessorType {
     ContactFormEmail,
+    LoadPlugin,
 }
 
 impl ProcessorType {
     pub(super) fn get_processor(&self) -> Box<dyn Processor> {
         match self {
             ProcessorType::ContactFormEmail => Box::new(QueueContactEmailProcessor),
+            ProcessorType::LoadPlugin => Box::new(LoadPlugin),
         }
     }
 }

server/service/src/processors/load_plugin/load_plugin.rs

@@ -0,0 +1,43 @@
+use repository::{
+    BackendPluginRowRepository, ChangelogRow, ChangelogTableName, KeyType, StorageConnection,
+};
+
+use crate::{
+    backend_plugin::plugin_provider::PluginInstance,
+    processors::general_processor::{Processor, ProcessorError},
+};
+
+const DESCRIPTION: &str = "Load plugins";
+
+pub(crate) struct LoadPlugin;
+
+impl Processor for LoadPlugin {
+    fn get_description(&self) -> String {
+        DESCRIPTION.to_string()
+    }
+
+    fn try_process_record(
+        &self,
+        connection: &StorageConnection,
+        changelog: &ChangelogRow,
+    ) -> Result<Option<String>, ProcessorError> {
+        let plugin = BackendPluginRowRepository::new(connection)
+            .find_one_by_id(&changelog.record_id)?
+            .ok_or(ProcessorError::RecordNotFound(
+                "Backend plugin".to_string(),
+                changelog.record_id.clone(),
+            ))?;
+
+        PluginInstance::bind(plugin);
+
+        Ok(Some("success".to_string()))
+    }
+
+    fn change_log_table_names(&self) -> Vec<ChangelogTableName> {
+        vec![ChangelogTableName::BackendPlugin]
+    }
+
+    fn cursor_type(&self) -> KeyType {
+        KeyType::LoadPluginProcessorCursor
+    }
+}

server/service/src/processors/load_plugin/mod.rs

@@ -0,0 +1,2 @@
+mod load_plugin;
+pub(crate) use self::load_plugin::*;

server/service/src/processors/mod.rs

@@ -19,6 +19,7 @@ use general_processor::{process_records, ProcessorError};
 
 mod contact_form;
 mod general_processor;
+mod load_plugin;
 pub use general_processor::ProcessorType;
 #[cfg(test)]
 mod test_helpers;

server/service/src/sync/synchroniser.rs

@@ -289,6 +289,9 @@ impl Synchroniser {
         ctx.processors_trigger
             .trigger_processor(ProcessorType::ContactFormEmail);
 
+        ctx.processors_trigger
+            .trigger_processor(ProcessorType::LoadPlugin);
+
         Ok(())
     }
 }