refactor

crates/application/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2024"
 [dependencies]
 async-trait = { workspace = true }
 domain = { workspace = true }
+reqwest = { workspace = true }
 uuid = { workspace = true }
 chrono = { workspace = true }
 tracing = { workspace = true }

crates/application/src/movies/event_handler.rs

@@ -0,0 +1,103 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError,
+    events::DomainEvent,
+    models::MovieProfile,
+    ports::{
+        EventHandler, MovieEnrichmentClient, MovieProfileRepository, MovieRepository,
+        ObjectStorage, PersonCommand, SearchCommand,
+    },
+};
+
+use crate::movies::{
+    commands::EnrichMovieCommand, deps::EnrichMovieDeps, enrich_movie, request_enrichment,
+};
+
+pub struct MovieEnrichmentHandler {
+    enrichment_client: Arc<dyn MovieEnrichmentClient>,
+    movie_repository: Arc<dyn MovieRepository>,
+    profile_repo: Arc<dyn MovieProfileRepository>,
+    person_command: Arc<dyn PersonCommand>,
+    search_command: Arc<dyn SearchCommand>,
+    object_storage: Arc<dyn ObjectStorage>,
+    http: reqwest::Client,
+}
+
+impl MovieEnrichmentHandler {
+    pub fn new(
+        enrichment_client: Arc<dyn MovieEnrichmentClient>,
+        movie_repository: Arc<dyn MovieRepository>,
+        profile_repo: Arc<dyn MovieProfileRepository>,
+        person_command: Arc<dyn PersonCommand>,
+        search_command: Arc<dyn SearchCommand>,
+        object_storage: Arc<dyn ObjectStorage>,
+    ) -> Self {
+        Self {
+            enrichment_client,
+            movie_repository,
+            profile_repo,
+            person_command,
+            search_command,
+            object_storage,
+            http: reqwest::Client::new(),
+        }
+    }
+
+    async fn download_cast_photos(&self, profile: &MovieProfile) {
+        for member in profile.cast.iter().take(5) {
+            let Some(ref path) = member.profile_path else {
+                continue;
+            };
+            let key = format!("cast{path}");
+            if self.object_storage.get(&key).await.is_ok() {
+                continue;
+            }
+            let url = format!("https://image.tmdb.org/t/p/w185{path}");
+            match self.http.get(&url).send().await {
+                Ok(resp) if resp.status().is_success() => {
+                    if let Ok(bytes) = resp.bytes().await
+                        && let Err(e) = self.object_storage.store(&key, &bytes).await
+                    {
+                        tracing::debug!("cast photo store failed for {path}: {e}");
+                    }
+                }
+                _ => tracing::debug!("cast photo download failed for {path}"),
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl EventHandler for MovieEnrichmentHandler {
+    async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
+        let (movie_id, external_metadata_id) = match event {
+            DomainEvent::MovieEnrichmentRequested {
+                movie_id,
+                external_metadata_id,
+            } => (movie_id.clone(), external_metadata_id.clone()),
+            _ => return Ok(()),
+        };
+
+        let Some(profile) = request_enrichment::fetch_if_stale(
+            self.enrichment_client.as_ref(),
+            &self.profile_repo,
+            movie_id.clone(),
+            external_metadata_id.value(),
+        )
+        .await?
+        else {
+            return Ok(());
+        };
+
+        self.download_cast_photos(&profile).await;
+        let enrich_deps = EnrichMovieDeps {
+            movie: self.movie_repository.clone(),
+            movie_profile: self.profile_repo.clone(),
+            person_command: self.person_command.clone(),
+            search_command: self.search_command.clone(),
+        };
+        enrich_movie::execute(&enrich_deps, EnrichMovieCommand { movie_id, profile }).await
+    }
+}

crates/application/src/movies/mod.rs

@@ -2,6 +2,7 @@ pub mod commands;
 pub mod deps;
 pub mod discovery_indexer;
 pub mod enrich_movie;
+pub mod event_handler;
 pub mod get_movie_profile;
 pub mod get_movies;
 pub mod queries;
@@ -11,5 +12,6 @@ pub mod search_cleanup;
 pub mod sync_poster;
 
 pub use discovery_indexer::MovieDiscoveryIndexer;
+pub use event_handler::MovieEnrichmentHandler;
 pub use reindex_search::SearchReindexHandler;
 pub use search_cleanup::SearchCleanupHandler;

crates/application/src/person/event_handler.rs

@@ -0,0 +1,45 @@
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use domain::{
+    errors::DomainError,
+    events::DomainEvent,
+    ports::{EventHandler, PersonCommand, PersonEnrichmentClient, PersonQuery},
+};
+
+use super::deps::EnrichPersonDeps;
+
+pub struct PersonEnrichmentHandler {
+    deps: EnrichPersonDeps,
+}
+
+impl PersonEnrichmentHandler {
+    pub fn new(
+        person_query: Arc<dyn PersonQuery>,
+        person_enrichment: Option<Arc<dyn PersonEnrichmentClient>>,
+        person_command: Arc<dyn PersonCommand>,
+    ) -> Self {
+        Self {
+            deps: EnrichPersonDeps {
+                person_query,
+                person_enrichment,
+                person_command,
+            },
+        }
+    }
+}
+
+#[async_trait]
+impl EventHandler for PersonEnrichmentHandler {
+    async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
+        let (person_id, external_person_id) = match event {
+            DomainEvent::PersonEnrichmentRequested {
+                person_id,
+                external_person_id,
+            } => (person_id.clone(), external_person_id.clone()),
+            _ => return Ok(()),
+        };
+
+        super::enrich::execute(&self.deps, person_id, external_person_id.value()).await
+    }
+}

crates/application/src/person/mod.rs

@@ -1,4 +1,7 @@
 pub mod deps;
 pub mod enrich;
+pub mod event_handler;
 pub mod get;
 pub mod get_credits;
+
+pub use event_handler::PersonEnrichmentHandler;