Event emitter (#955)

* ci: remove env for devcontainer

* ci: remove post create script

* refactor(backend): more complex structure for application event

* refactor(backend): move review posted to core job

* chore: remove useless function call

* chore(backend): emit event when entity added to collection

* feat(backend): send new performer

* refactor(backend): do not recreate exercise service in miscelleneous action

* refactor(backend): always call job when entity added to collection

* docs: change order of integration types informations

* fix(backend): call correct function to handle new event

* feat(backend): start implementing handler

* chore(backend): change order of arguments supplied

* feat(backend): complete function

* chore(backend): remove useless function calls

* feat(database): remove system information column

* chore(backend): adapt to new database schema

* chore(backend): add debug logging

* fix(backend): better error handling

* fix(backend): select correct identifier for metadata

* chore(database): do not create useless column

* fix(backend): do not force an unwrap

Author: IgnisDa

.devcontainer/devcontainer.json

@@ -4,14 +4,5 @@
 	"service": "ryot-app",
 	"forwardPorts": [],
 	"workspaceFolder": "/workspaces/ryot",
-	"postCreateCommand": ". ${containerWorkspaceFolder}/.devcontainer/scripts/post-create.sh",
-	"remoteUser": "archlinux",
-	"remoteEnv": {
-		"GIT_AUTHOR_NAME": "${localEnv:GIT_AUTHOR_NAME}",
-		"GIT_AUTHOR_EMAIL": "${localEnv:GIT_AUTHOR_EMAIL}",
-		"GIT_COMMITTER_NAME": "${localEnv:GIT_COMMITTER_NAME}",
-		"GIT_COMMITTER_EMAIL": "${localEnv:GIT_COMMITTER_EMAIL}",
-		"VISUAL": "hx",
-		"EDITOR": "hx"
-	}
+	"remoteUser": "archlinux"
 }

.devcontainer/scripts/post-create.sh

@@ -6,6 +6,7 @@ use chrono_tz::Tz;
 use database::{MediaLot, MediaSource};
 use serde::{Deserialize, Serialize};
 use strum::Display;
+use uuid::Uuid;
 
 use crate::{
     exporter::ExporterService,
@@ -48,11 +49,6 @@ pub async fn sync_integrations_data(
 ) -> Result<(), Error> {
     tracing::trace!("Getting data from yanked integrations for all users");
     misc_service.yank_integrations_data().await.unwrap();
-    tracing::trace!("Sending data for push integrations for all users");
-    misc_service
-        .send_data_for_push_integrations()
-        .await
-        .unwrap();
     Ok(())
 }
 
@@ -62,7 +58,9 @@ pub async fn sync_integrations_data(
 #[derive(Debug, Deserialize, Serialize, Display)]
 pub enum CoreApplicationJob {
     SyncIntegrationsData(String),
+    ReviewPosted(ReviewPostedEvent),
     BulkProgressUpdate(String, Vec<ProgressUpdateInput>),
+    EntityAddedToCollection(String, Uuid),
 }
 
 impl Message for CoreApplicationJob {
@@ -77,20 +75,23 @@ pub async fn perform_core_application_job(
     tracing::trace!("Started job: {:#?}", name);
     let start = Instant::now();
     let status = match information {
-        CoreApplicationJob::SyncIntegrationsData(user_id) => {
-            misc_service
-                .push_integrations_data_for_user(&user_id)
-                .await
-                .ok();
-            misc_service
-                .yank_integrations_data_for_user(&user_id)
-                .await
-                .is_ok()
+        CoreApplicationJob::SyncIntegrationsData(user_id) => misc_service
+            .yank_integrations_data_for_user(&user_id)
+            .await
+            .is_ok(),
+        CoreApplicationJob::ReviewPosted(event) => {
+            misc_service.handle_review_posted_event(event).await.is_ok()
         }
         CoreApplicationJob::BulkProgressUpdate(user_id, input) => misc_service
             .bulk_progress_update(user_id, input)
             .await
             .is_ok(),
+        CoreApplicationJob::EntityAddedToCollection(user_id, collection_to_entity_id) => {
+            misc_service
+                .handle_entity_added_to_collection_event(user_id, collection_to_entity_id)
+                .await
+                .is_ok()
+        }
     };
     tracing::trace!(
         "Job: {:#?}, Time Taken: {}ms, Successful = {}",
@@ -111,10 +112,10 @@ pub enum ApplicationJob {
     UpdatePerson(String),
     RecalculateCalendarEvents,
     AssociateGroupWithMetadata(MediaLot, MediaSource, String),
-    ReviewPosted(ReviewPostedEvent),
     PerformExport(String, Vec<ExportItem>),
     RecalculateUserSummary(String),
     PerformBackgroundTasks,
+    UpdateExerciseLibrary,
 }
 
 impl Message for ApplicationJob {
@@ -171,13 +172,14 @@ pub async fn perform_application_job(
             })
             .await
             .is_ok(),
-        ApplicationJob::ReviewPosted(event) => {
-            misc_service.handle_review_posted_event(event).await.is_ok()
-        }
         ApplicationJob::PerformExport(user_id, to_export) => exporter_service
             .perform_export(user_id, to_export)
             .await
             .is_ok(),
+        ApplicationJob::UpdateExerciseLibrary => exercise_service
+            .deploy_update_exercise_library_job()
+            .await
+            .is_ok(),
     };
     tracing::trace!(
         "Job: {:#?}, Time Taken: {}ms, Successful = {}",

apps/backend/src/background.rs

@@ -5,8 +5,6 @@ use sea_orm::entity::prelude::*;
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-use crate::models::CollectionToEntitySystemInformation;
-
 #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq, Serialize, Deserialize)]
 #[sea_orm(table_name = "collection_to_entity")]
 pub struct Model {
@@ -23,8 +21,6 @@ pub struct Model {
     pub exercise_id: Option<String>,
     pub workout_id: Option<String>,
     pub information: Option<serde_json::Value>,
-    #[sea_orm(column_type = "Json")]
-    pub system_information: CollectionToEntitySystemInformation,
 }
 
 #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]

apps/backend/src/entities/collection_to_entity.rs

@@ -18,7 +18,7 @@ use slug::slugify;
 use struson::writer::{JsonStreamWriter, JsonWriter};
 
 use crate::{
-    background::ApplicationJob,
+    background::{ApplicationJob, CoreApplicationJob},
     entities::{
         collection, collection_to_entity, exercise,
         prelude::{CollectionToEntity, Exercise, UserMeasurement, UserToEntity, Workout},
@@ -290,6 +290,7 @@ pub struct ExerciseService {
     config: Arc<config::AppConfig>,
     file_storage_service: Arc<FileStorageService>,
     perform_application_job: MemoryStorage<ApplicationJob>,
+    perform_core_application_job: MemoryStorage<CoreApplicationJob>,
 }
 
 impl ExerciseService {
@@ -298,12 +299,14 @@ impl ExerciseService {
         config: Arc<config::AppConfig>,
         file_storage_service: Arc<FileStorageService>,
         perform_application_job: &MemoryStorage<ApplicationJob>,
+        perform_core_application_job: &MemoryStorage<CoreApplicationJob>,
     ) -> Self {
         Self {
             config,
             db: db.clone(),
             file_storage_service,
             perform_application_job: perform_application_job.clone(),
+            perform_core_application_job: perform_core_application_job.clone(),
         }
     }
 }
@@ -787,6 +790,7 @@ impl ExerciseService {
                 exercise_id: Some(exercise.id.clone()),
                 ..Default::default()
             },
+            &self.perform_core_application_job,
         )
         .await?;
         Ok(exercise.id)

apps/backend/src/fitness/resolver.rs

@@ -555,6 +555,7 @@ impl IntegrationService {
         let mut options = RadarrAddMovieOptions::new();
         options.search_for_movie = Some(true);
         resource.add_options = Some(Box::new(options));
+        tracing::debug!("Pushing movie to Radarr {:?}", resource);
         radarr_api_v3_movie_post(&configuration, Some(resource))
             .await
             .trace_ok();
@@ -585,6 +586,7 @@ impl IntegrationService {
         let mut options = SonarrAddSeriesOptions::new();
         options.search_for_missing_episodes = Some(true);
         resource.add_options = Some(Box::new(options));
+        tracing::debug!("Pushing series to Sonarr {:?}", resource);
         sonarr_api_v3_series_post(&configuration, Some(resource))
             .await
             .trace_ok();

apps/backend/src/integrations.rs

@@ -13,7 +13,7 @@ use apalis::{
     layers::{
         limit::RateLimitLayer as ApalisRateLimitLayer, tracing::TraceLayer as ApalisTraceLayer,
     },
-    prelude::{MemoryStorage, Monitor, WorkerBuilder, WorkerFactoryFn},
+    prelude::{MemoryStorage, MessageQueue, Monitor, WorkerBuilder, WorkerFactoryFn},
     utils::TokioExecutor,
 };
 use aws_sdk_s3::config::Region;
@@ -23,6 +23,7 @@ use axum::{
     routing::{get, post, Router},
     Extension,
 };
+use background::ApplicationJob;
 use chrono::{TimeZone, Utc};
 use database::Migrator;
 use itertools::Itertools;
@@ -172,9 +173,8 @@ async fn main() -> Result<()> {
 
     if Exercise::find().count(&db).await? == 0 {
         tracing::info!("Instance does not have exercises data. Deploying job to download them...");
-        app_services
-            .exercise_service
-            .deploy_update_exercise_library_job()
+        perform_application_job_storage
+            .enqueue(ApplicationJob::UpdateExerciseLibrary)
             .await
             .unwrap();
     }
@@ -198,7 +198,7 @@ async fn main() -> Result<()> {
                 base_dir.join("backend-config-schema.yaml"),
                 YamlTemplateRenderer::default(),
             )
-            .unwrap();
+            .ok();
 
         let mut generator = SchemaGenerator::default();
         generator.add::<CompleteExport>();
@@ -207,7 +207,7 @@ async fn main() -> Result<()> {
                 base_dir.join("export-schema.ts"),
                 TypeScriptRenderer::default(),
             )
-            .unwrap();
+            .ok();
     }
 
     let schema = get_schema(&app_services).await;