diff --git a/crates/adapters/tmdb-enrichment/src/person_handler.rs b/crates/adapters/tmdb-enrichment/src/person_handler.rs index 20fb00a..32abfda 100644 --- a/crates/adapters/tmdb-enrichment/src/person_handler.rs +++ b/crates/adapters/tmdb-enrichment/src/person_handler.rs @@ -1,32 +1,19 @@ -use std::sync::Arc; - use async_trait::async_trait; -use chrono::Utc; use domain::{ errors::DomainError, events::DomainEvent, - ports::{EventHandler, PersonCommand, PersonEnrichmentClient, PersonQuery}, + ports::EventHandler, }; -const STALENESS_DAYS: i64 = 90; +use application::context::AppContext; pub struct PersonEnrichmentHandler { - enrichment_client: Arc, - person_query: Arc, - person_command: Arc, + ctx: AppContext, } impl PersonEnrichmentHandler { - pub fn new( - enrichment_client: Arc, - person_query: Arc, - person_command: Arc, - ) -> Self { - Self { - enrichment_client, - person_query, - person_command, - } + pub fn new(ctx: AppContext) -> Self { + Self { ctx } } } @@ -41,21 +28,6 @@ impl EventHandler for PersonEnrichmentHandler { _ => return Ok(()), }; - if let Some(person) = self.person_query.get_by_id(&person_id).await? - && let Some(at) = person.enriched_at() - && (Utc::now() - at).num_days() < STALENESS_DAYS - { - tracing::debug!(person_id = %person_id.value(), "person enrichment still fresh"); - return Ok(()); - } - - tracing::info!(person_id = %person_id.value(), "enriching person from TMDb"); - let data = self - .enrichment_client - .fetch_details(&external_person_id) - .await?; - self.person_command - .update_enrichment(&person_id, &data) - .await + application::person::enrich::execute(&self.ctx, person_id, &external_person_id).await } } diff --git a/crates/application/src/person/enrich.rs b/crates/application/src/person/enrich.rs index 65ca412..cf38ea6 100644 --- a/crates/application/src/person/enrich.rs +++ b/crates/application/src/person/enrich.rs @@ -1,16 +1,35 @@ use crate::context::AppContext; +use chrono::Utc; use domain::{ errors::DomainError, - models::{PersonEnrichmentData, PersonId}, + models::PersonId, }; -pub async fn execute( - ctx: &AppContext, - person_id: PersonId, - data: PersonEnrichmentData, -) -> Result<(), DomainError> { - ctx.repos - .person_command - .update_enrichment(&person_id, &data) - .await +const STALENESS_DAYS: i64 = 90; + +pub async fn execute(ctx: &AppContext, person_id: PersonId, external_id: &str) -> Result<(), DomainError> { + if let Some(person) = ctx.repos.person_query.get_by_id(&person_id).await? + && let Some(at) = person.enriched_at() + && (Utc::now() - at).num_days() < STALENESS_DAYS + { + tracing::debug!(person_id = %person_id.value(), "person enrichment still fresh"); + return Ok(()); + } + + let client = ctx.services.person_enrichment.as_ref().ok_or_else(|| { + DomainError::InfrastructureError("person enrichment client not configured".into()) + })?; + + let data = match client.fetch_details(external_id).await { + Ok(d) => d, + Err(DomainError::NotFound(msg)) => { + tracing::warn!("TMDb person lookup found nothing: {msg}"); + return Ok(()); + } + Err(e) => return Err(e), + }; + + ctx.repos.person_command.update_enrichment(&person_id, &data).await?; + tracing::info!(person_id = %person_id.value(), "person enriched"); + Ok(()) } diff --git a/crates/worker/src/main.rs b/crates/worker/src/main.rs index b6ea2b2..30797de 100644 --- a/crates/worker/src/main.rs +++ b/crates/worker/src/main.rs @@ -146,13 +146,11 @@ async fn main() -> anyhow::Result<()> { Arc::clone(&ctx.repos.search_command), Arc::clone(&ctx.services.object_storage), )) as Arc; - let person_handler = Arc::new(tmdb_enrichment::PersonEnrichmentHandler::new( - Arc::clone(&client) as Arc, - Arc::clone(&ctx.repos.person_query), - Arc::clone(&ctx.repos.person_command), - )) as Arc; ctx.services.person_enrichment = Some(Arc::clone(&client) as Arc); + let person_handler = Arc::new(tmdb_enrichment::PersonEnrichmentHandler::new( + ctx.clone(), + )) as Arc; let job = Arc::new(application::jobs::EnrichmentStalenessJob::new(ctx.clone())) as Arc; (Some(handler), Some(person_handler), Some(job))