refactor: group use cases into DDD bounded contexts
Flat use_cases/ (44 files) + monolithic commands.rs/queries.rs split into diary/, movies/, watchlist/, import/, auth/, users/, integrations/, search/, person/, federation/ — each with own commands.rs, queries.rs, and use case modules. Inline tests extracted to sibling tests/ dirs.
This commit is contained in:
4
crates/application/src/movies/commands.rs
Normal file
4
crates/application/src/movies/commands.rs
Normal file
@@ -0,0 +1,4 @@
|
||||
pub struct EnrichMovieCommand {
|
||||
pub movie_id: domain::value_objects::MovieId,
|
||||
pub profile: domain::models::MovieProfile,
|
||||
}
|
||||
58
crates/application/src/movies/discovery_indexer.rs
Normal file
58
crates/application/src/movies/discovery_indexer.rs
Normal file
@@ -0,0 +1,58 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
events::DomainEvent,
|
||||
models::IndexableDocument,
|
||||
ports::{EventHandler, MovieRepository, SearchCommand},
|
||||
};
|
||||
|
||||
/// Reacts to `MovieDiscovered` and inserts a bare search index entry immediately,
|
||||
/// so movies are findable before TMDb enrichment runs.
|
||||
/// Enrichment will later overwrite this with the full document (cast, genres, etc.).
|
||||
pub struct MovieDiscoveryIndexer {
|
||||
movie_repository: Arc<dyn MovieRepository>,
|
||||
search_command: Arc<dyn SearchCommand>,
|
||||
}
|
||||
|
||||
impl MovieDiscoveryIndexer {
|
||||
pub fn new(
|
||||
movie_repository: Arc<dyn MovieRepository>,
|
||||
search_command: Arc<dyn SearchCommand>,
|
||||
) -> Self {
|
||||
Self {
|
||||
movie_repository,
|
||||
search_command,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EventHandler for MovieDiscoveryIndexer {
|
||||
async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
|
||||
let movie_id = match event {
|
||||
DomainEvent::MovieDiscovered { movie_id, .. } => movie_id.clone(),
|
||||
_ => return Ok(()),
|
||||
};
|
||||
|
||||
let Some(movie) = self.movie_repository.get_movie_by_id(&movie_id).await? else {
|
||||
tracing::warn!(movie_id = %movie_id.value(), "MovieDiscoveryIndexer: movie not found");
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
if let Err(e) = self
|
||||
.search_command
|
||||
.index(IndexableDocument::Movie {
|
||||
id: movie_id.clone(),
|
||||
movie: Box::new(movie),
|
||||
profile: None,
|
||||
})
|
||||
.await
|
||||
{
|
||||
tracing::warn!(movie_id = %movie_id.value(), "failed to index movie on discovery: {e}");
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
94
crates/application/src/movies/enrich_movie.rs
Normal file
94
crates/application/src/movies/enrich_movie.rs
Normal file
@@ -0,0 +1,94 @@
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
models::{CastMember, CrewMember, ExternalPersonId, IndexableDocument, Person, PersonId},
|
||||
ports::{MovieProfileRepository, MovieRepository, PersonCommand, SearchCommand},
|
||||
};
|
||||
|
||||
use crate::movies::commands::EnrichMovieCommand;
|
||||
|
||||
pub async fn execute(
|
||||
movie_repository: &Arc<dyn MovieRepository>,
|
||||
profile_repository: &Arc<dyn MovieProfileRepository>,
|
||||
person_command: &Arc<dyn PersonCommand>,
|
||||
search_command: &Arc<dyn SearchCommand>,
|
||||
cmd: EnrichMovieCommand,
|
||||
) -> Result<(), DomainError> {
|
||||
// 1. Persist the enriched profile (also handles movie_cast, movie_crew, genres, keywords)
|
||||
profile_repository.upsert(&cmd.profile).await?;
|
||||
|
||||
// 2. Upsert persons extracted from cast + crew (no reads — only upsert)
|
||||
let persons = extract_persons(&cmd.profile.cast, &cmd.profile.crew);
|
||||
if !persons.is_empty() {
|
||||
person_command.upsert_batch(&persons).await?;
|
||||
}
|
||||
|
||||
// 3. Fetch the movie for the search index document
|
||||
let Some(movie) = movie_repository.get_movie_by_id(&cmd.movie_id).await? else {
|
||||
tracing::warn!(movie_id = %cmd.movie_id.value(), "enrich_movie: movie not found after profile upsert");
|
||||
return Ok(());
|
||||
};
|
||||
|
||||
// 4. Index the movie in search
|
||||
search_command
|
||||
.index(IndexableDocument::Movie {
|
||||
id: cmd.movie_id.clone(),
|
||||
movie: Box::new(movie),
|
||||
profile: Some(Box::new(cmd.profile.clone())),
|
||||
})
|
||||
.await?;
|
||||
|
||||
// 5. Index each unique person in search (no reads — persons built from in-memory data)
|
||||
for person in &persons {
|
||||
search_command
|
||||
.index(IndexableDocument::Person {
|
||||
id: person.id().clone(),
|
||||
person: Box::new(person.clone()),
|
||||
})
|
||||
.await?;
|
||||
}
|
||||
|
||||
tracing::info!(
|
||||
movie_id = %cmd.movie_id.value(),
|
||||
persons = persons.len(),
|
||||
"enrich_movie: profile stored and search index updated"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Build unique Person values from cast and crew.
|
||||
/// Uses deterministic UUIDv5 so the same tmdb_person_id always maps to the same PersonId.
|
||||
/// No DB reads — persons are built entirely from in-memory TMDb data.
|
||||
fn extract_persons(cast: &[CastMember], crew: &[CrewMember]) -> Vec<Person> {
|
||||
let mut seen: HashMap<u64, Person> = HashMap::new();
|
||||
|
||||
for member in cast {
|
||||
seen.entry(member.tmdb_person_id).or_insert_with(|| {
|
||||
let ext = ExternalPersonId::new(format!("tmdb:{}", member.tmdb_person_id));
|
||||
Person::new(
|
||||
PersonId::from_external(&ext),
|
||||
ext,
|
||||
member.name.clone(),
|
||||
Some("Acting".to_string()),
|
||||
member.profile_path.clone(),
|
||||
)
|
||||
});
|
||||
}
|
||||
|
||||
for member in crew {
|
||||
seen.entry(member.tmdb_person_id).or_insert_with(|| {
|
||||
let ext = ExternalPersonId::new(format!("tmdb:{}", member.tmdb_person_id));
|
||||
Person::new(
|
||||
PersonId::from_external(&ext),
|
||||
ext,
|
||||
member.name.clone(),
|
||||
Some(member.department.clone()),
|
||||
member.profile_path.clone(),
|
||||
)
|
||||
});
|
||||
}
|
||||
|
||||
seen.into_values().collect()
|
||||
}
|
||||
20
crates/application/src/movies/get_movies.rs
Normal file
20
crates/application/src/movies/get_movies.rs
Normal file
@@ -0,0 +1,20 @@
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
models::collections::{PageParams, Paginated},
|
||||
models::{MovieFilter, MovieSummary},
|
||||
};
|
||||
|
||||
use crate::{context::AppContext, movies::queries::GetMoviesQuery};
|
||||
|
||||
pub async fn execute(
|
||||
ctx: &AppContext,
|
||||
query: GetMoviesQuery,
|
||||
) -> Result<Paginated<MovieSummary>, DomainError> {
|
||||
let page = PageParams::new(query.limit, query.offset)?;
|
||||
let filter = MovieFilter {
|
||||
search: query.search,
|
||||
genre: query.genre,
|
||||
language: query.language,
|
||||
};
|
||||
ctx.repos.movie.list_movies(&page, &filter).await
|
||||
}
|
||||
10
crates/application/src/movies/mod.rs
Normal file
10
crates/application/src/movies/mod.rs
Normal file
@@ -0,0 +1,10 @@
|
||||
pub mod commands;
|
||||
pub mod discovery_indexer;
|
||||
pub mod enrich_movie;
|
||||
pub mod get_movies;
|
||||
pub mod queries;
|
||||
pub mod search_cleanup;
|
||||
pub mod sync_poster;
|
||||
|
||||
pub use discovery_indexer::MovieDiscoveryIndexer;
|
||||
pub use search_cleanup::SearchCleanupHandler;
|
||||
7
crates/application/src/movies/queries.rs
Normal file
7
crates/application/src/movies/queries.rs
Normal file
@@ -0,0 +1,7 @@
|
||||
pub struct GetMoviesQuery {
|
||||
pub limit: Option<u32>,
|
||||
pub offset: Option<u32>,
|
||||
pub search: Option<String>,
|
||||
pub genre: Option<String>,
|
||||
pub language: Option<String>,
|
||||
}
|
||||
58
crates/application/src/movies/search_cleanup.rs
Normal file
58
crates/application/src/movies/search_cleanup.rs
Normal file
@@ -0,0 +1,58 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
events::DomainEvent,
|
||||
models::EntityType,
|
||||
ports::{EventHandler, PersonQuery, SearchCommand},
|
||||
};
|
||||
|
||||
pub struct SearchCleanupHandler {
|
||||
search_command: Arc<dyn SearchCommand>,
|
||||
person_query: Arc<dyn PersonQuery>,
|
||||
}
|
||||
|
||||
impl SearchCleanupHandler {
|
||||
pub fn new(search_command: Arc<dyn SearchCommand>, person_query: Arc<dyn PersonQuery>) -> Self {
|
||||
Self {
|
||||
search_command,
|
||||
person_query,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EventHandler for SearchCleanupHandler {
|
||||
async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
|
||||
let movie_id = match event {
|
||||
DomainEvent::MovieDeleted { movie_id, .. } => movie_id.value().to_string(),
|
||||
_ => return Ok(()),
|
||||
};
|
||||
|
||||
if let Err(e) = self
|
||||
.search_command
|
||||
.remove(EntityType::Movie, &movie_id)
|
||||
.await
|
||||
{
|
||||
tracing::warn!("search cleanup failed for movie {movie_id}: {e}");
|
||||
}
|
||||
|
||||
// Remove persons who have no remaining movie credits (orphaned after cascade delete).
|
||||
match self.person_query.list_orphaned_persons().await {
|
||||
Ok(orphans) => {
|
||||
for person_id in orphans {
|
||||
let id = person_id.value().to_string();
|
||||
if let Err(e) = self.search_command.remove(EntityType::Person, &id).await {
|
||||
tracing::warn!("search cleanup failed for orphaned person {id}: {e}");
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => tracing::warn!(
|
||||
"failed to list orphaned persons after movie {movie_id} deletion: {e}"
|
||||
),
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
98
crates/application/src/movies/sync_poster.rs
Normal file
98
crates/application/src/movies/sync_poster.rs
Normal file
@@ -0,0 +1,98 @@
|
||||
use domain::{
|
||||
errors::DomainError,
|
||||
events::DomainEvent,
|
||||
models::IndexableDocument,
|
||||
value_objects::{MovieId, PosterPath},
|
||||
};
|
||||
|
||||
use crate::{context::AppContext, diary::commands::SyncPosterCommand};
|
||||
|
||||
pub async fn execute(ctx: &AppContext, cmd: SyncPosterCommand) -> Result<(), DomainError> {
|
||||
let movie_id = MovieId::from_uuid(cmd.movie_id);
|
||||
|
||||
let mut movie = match ctx.repos.movie.get_movie_by_id(&movie_id).await? {
|
||||
Some(m) => m,
|
||||
None => {
|
||||
tracing::warn!(
|
||||
"Sync cancelled: Movie {} not found in local DB",
|
||||
movie_id.value()
|
||||
);
|
||||
return Err(DomainError::NotFound("Movie not found".into()));
|
||||
}
|
||||
};
|
||||
|
||||
let external_metadata_id = movie
|
||||
.external_metadata_id()
|
||||
.ok_or_else(|| {
|
||||
DomainError::ValidationError(
|
||||
"Movie has no external metadata ID, cannot sync poster".into(),
|
||||
)
|
||||
})?
|
||||
.clone();
|
||||
|
||||
let poster_url = match ctx
|
||||
.services
|
||||
.metadata
|
||||
.get_poster_url(&external_metadata_id)
|
||||
.await
|
||||
{
|
||||
Ok(Some(url)) => url,
|
||||
Ok(None) => return Ok(()),
|
||||
Err(e) => {
|
||||
tracing::warn!("Warning: Failed to find poster URL metadata: {:?}", e);
|
||||
return Err(e);
|
||||
}
|
||||
};
|
||||
|
||||
let image_bytes = ctx
|
||||
.services
|
||||
.poster_fetcher
|
||||
.fetch_poster_bytes(&poster_url)
|
||||
.await?;
|
||||
|
||||
let stored_path = ctx
|
||||
.services
|
||||
.image_storage
|
||||
.store(&movie_id.value().to_string(), &image_bytes)
|
||||
.await?;
|
||||
|
||||
if let Err(e) = ctx
|
||||
.services
|
||||
.event_publisher
|
||||
.publish(&DomainEvent::ImageStored {
|
||||
key: stored_path.clone(),
|
||||
})
|
||||
.await
|
||||
{
|
||||
tracing::warn!("failed to emit ImageStored for {stored_path}: {e}");
|
||||
}
|
||||
|
||||
let poster_path = PosterPath::new(stored_path)?;
|
||||
|
||||
movie.update_poster(poster_path);
|
||||
ctx.repos.movie.upsert_movie(&movie).await?;
|
||||
|
||||
// Refresh search index so the new poster_path is reflected immediately.
|
||||
// Fetch existing profile if available for a complete index document.
|
||||
let profile = ctx
|
||||
.repos
|
||||
.movie_profile
|
||||
.get_by_movie_id(&movie_id)
|
||||
.await
|
||||
.ok()
|
||||
.flatten();
|
||||
if let Err(e) = ctx
|
||||
.repos
|
||||
.search_command
|
||||
.index(IndexableDocument::Movie {
|
||||
id: movie_id.clone(),
|
||||
movie: Box::new(movie),
|
||||
profile: profile.map(Box::new),
|
||||
})
|
||||
.await
|
||||
{
|
||||
tracing::warn!(movie_id = %movie_id.value(), "failed to refresh search index after poster sync: {e}");
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user