Compare commits

...

16 Commits

Author SHA1 Message Date
57520c00f3 refactor: move AppContext to presentation crate, structurally enforce boundary
All checks were successful
CI / Check / Test (push) Successful in 39m33s
2026-06-11 23:18:28 +02:00
b5cc7f8371 refactor(search): fix test to not use AppContext 2026-06-11 23:00:01 +02:00
9ca5ada924 refactor(auth): LoginDeps, RegisterDeps, RefreshDeps, RegisterAndLoginDeps, RefreshSessionCleanupJob 2026-06-11 22:58:42 +02:00
70d1f10e3d refactor(users): fix test files to not use AppContext 2026-06-11 22:49:44 +02:00
61980b0cfb refactor(users): GetProfileDeps, UpdateProfileDeps, scoped Arc deps 2026-06-11 22:47:17 +02:00
7bf5c47f5b refactor(diary): DeleteReviewDeps, GetMovieSocialPageDeps, GetActivityFeedDeps 2026-06-11 22:37:35 +02:00
ddf100cfc2 refactor(wrapup): scoped deps — HandleWrapUpRequestedDeps, flat-Arc jobs 2026-06-11 22:29:30 +02:00
cdff0de53d refactor(movies): collapse single-field deps structs to Arc params 2026-06-11 22:17:09 +02:00
1e62f12903 refactor(movies): EnrichMovieDeps, ReindexSearchDeps, SyncPosterDeps, SearchReindexHandler, EnrichmentStalenessJob 2026-06-11 22:13:25 +02:00
66bd138927 refactor(person): EnrichPersonDeps + GetPersonDeps, PersonEnrichmentHandler 2026-06-11 22:05:38 +02:00
b29f3020e6 refactor(integrations): IngestWatchEventDeps, scoped Arc deps, WatchEventCleanupJob 2026-06-11 22:01:15 +02:00
76edd52bb0 refactor(import): fix test files to not use AppContext 2026-06-11 21:51:24 +02:00
b5ff43d9dc refactor(import): scoped Arc deps, ImportSessionCleanupJob 2026-06-11 21:49:15 +02:00
b552c1d156 refactor(watchlist): WatchlistAddDeps, scoped Arc deps 2026-06-11 21:40:48 +02:00
f006ba00a8 refactor(goals): scoped Arc deps instead of AppContext 2026-06-11 21:36:11 +02:00
2b295e10ba refactor(search): scoped Arc dep instead of AppContext 2026-06-11 21:32:19 +02:00
162 changed files with 2170 additions and 1728 deletions

View File

@@ -1,9 +1,18 @@
.DEFAULT_GOAL := check
# Run the full local check suite — same order as CI would.
check: fmt-check clippy test
check: fmt-check clippy test check-appcontext
@echo "✅ All checks passed"
# Enforce that no application use case imports AppContext (god-object guard).
check-appcontext:
@if grep -rn "AppContext" crates/application/src --include="*.rs" | grep -q .; then \
echo "❌ AppContext found in application crate:"; \
grep -rn "AppContext" crates/application/src --include="*.rs"; \
exit 1; \
fi
@echo "✅ No AppContext in application crate"
# Apply rustfmt to all files.
fmt:
cargo fmt

View File

@@ -1,6 +1,8 @@
use std::sync::Arc;
use application::movies::{commands::EnrichMovieCommand, enrich_movie, request_enrichment};
use application::movies::{
commands::EnrichMovieCommand, deps::EnrichMovieDeps, enrich_movie, request_enrichment,
};
use async_trait::async_trait;
use domain::{
errors::DomainError,
@@ -89,13 +91,12 @@ impl EventHandler for MovieEnrichmentHandler {
};
self.download_cast_photos(&profile).await;
enrich_movie::execute(
&self.movie_repository,
&self.profile_repo,
&self.person_command,
&self.search_command,
EnrichMovieCommand { movie_id, 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
}
}

View File

@@ -1,15 +1,31 @@
use async_trait::async_trait;
use domain::{errors::DomainError, events::DomainEvent, ports::EventHandler};
use std::sync::Arc;
use application::context::AppContext;
use async_trait::async_trait;
use domain::{
errors::DomainError,
events::DomainEvent,
ports::{EventHandler, PersonCommand, PersonEnrichmentClient, PersonQuery},
};
use application::person::deps::EnrichPersonDeps;
pub struct PersonEnrichmentHandler {
ctx: AppContext,
deps: EnrichPersonDeps,
}
impl PersonEnrichmentHandler {
pub fn new(ctx: AppContext) -> Self {
Self { ctx }
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,
},
}
}
}
@@ -24,6 +40,6 @@ impl EventHandler for PersonEnrichmentHandler {
_ => return Ok(()),
};
application::person::enrich::execute(&self.ctx, person_id, &external_person_id).await
application::person::enrich::execute(&self.deps, person_id, &external_person_id).await
}
}

View File

@@ -0,0 +1,33 @@
use std::sync::Arc;
use domain::ports::{AuthService, PasswordHasher, RefreshSessionRepository, UserRepository};
use crate::config::AppConfig;
pub struct LoginDeps {
pub user: Arc<dyn UserRepository>,
pub password_hasher: Arc<dyn PasswordHasher>,
pub auth: Arc<dyn AuthService>,
pub refresh_session: Arc<dyn RefreshSessionRepository>,
pub config: AppConfig,
}
pub struct RegisterDeps {
pub user: Arc<dyn UserRepository>,
pub password_hasher: Arc<dyn PasswordHasher>,
pub config: AppConfig,
}
pub struct RefreshDeps {
pub refresh_session: Arc<dyn RefreshSessionRepository>,
pub auth: Arc<dyn AuthService>,
pub config: AppConfig,
}
pub struct RegisterAndLoginDeps {
pub user: Arc<dyn UserRepository>,
pub password_hasher: Arc<dyn PasswordHasher>,
pub auth: Arc<dyn AuthService>,
pub refresh_session: Arc<dyn RefreshSessionRepository>,
pub config: AppConfig,
}

View File

@@ -3,7 +3,7 @@ use uuid::Uuid;
use domain::{errors::DomainError, models::RefreshSession, value_objects::Email};
use crate::{auth::queries::LoginQuery, context::AppContext};
use crate::auth::{deps::LoginDeps, queries::LoginQuery};
pub struct LoginResult {
pub token: String,
@@ -14,17 +14,15 @@ pub struct LoginResult {
pub role: String,
}
pub async fn execute(ctx: &AppContext, query: LoginQuery) -> Result<LoginResult, DomainError> {
pub async fn execute(deps: &LoginDeps, query: LoginQuery) -> Result<LoginResult, DomainError> {
let email = Email::new(query.email)?;
let user = ctx
.repos
let user = deps
.user
.find_by_email(&email)
.await?
.ok_or_else(|| DomainError::Unauthorized("Invalid credentials".into()))?;
let valid = ctx
.services
let valid = deps
.password_hasher
.verify(&query.password, user.password_hash())
.await?;
@@ -32,10 +30,10 @@ pub async fn execute(ctx: &AppContext, query: LoginQuery) -> Result<LoginResult,
return Err(DomainError::Unauthorized("Invalid credentials".into()));
}
let generated = ctx.services.auth.generate_token(user.id()).await?;
let generated = deps.auth.generate_token(user.id()).await?;
let refresh_token = Uuid::new_v4().to_string();
let refresh_expires = Utc::now() + Duration::seconds(ctx.config.refresh_ttl_seconds as i64);
let refresh_expires = Utc::now() + Duration::seconds(deps.config.refresh_ttl_seconds as i64);
let session = RefreshSession {
id: Uuid::new_v4(),
user_id: user.id().clone(),
@@ -43,7 +41,7 @@ pub async fn execute(ctx: &AppContext, query: LoginQuery) -> Result<LoginResult,
expires_at: refresh_expires,
created_at: Utc::now(),
};
ctx.repos.refresh_session.create(&session).await?;
deps.refresh_session.create(&session).await?;
Ok(LoginResult {
token: generated.token,

View File

@@ -1,9 +1,12 @@
use domain::errors::DomainError;
use std::sync::Arc;
use crate::context::AppContext;
use domain::{errors::DomainError, ports::RefreshSessionRepository};
pub async fn execute(ctx: &AppContext, refresh_token: &str) -> Result<(), DomainError> {
ctx.repos.refresh_session.revoke(refresh_token).await
pub async fn execute(
refresh_session: Arc<dyn RefreshSessionRepository>,
refresh_token: &str,
) -> Result<(), DomainError> {
refresh_session.revoke(refresh_token).await
}
#[cfg(test)]

View File

@@ -1,4 +1,5 @@
pub mod commands;
pub mod deps;
pub mod login;
pub mod logout;
pub mod queries;

View File

@@ -3,7 +3,7 @@ use uuid::Uuid;
use domain::{errors::DomainError, models::RefreshSession};
use crate::context::AppContext;
use crate::auth::deps::RefreshDeps;
pub struct RefreshResult {
pub token: String,
@@ -12,30 +12,29 @@ pub struct RefreshResult {
}
pub async fn execute(
ctx: &AppContext,
deps: &RefreshDeps,
old_refresh_token: &str,
) -> Result<RefreshResult, DomainError> {
let session = ctx
.repos
let session = deps
.refresh_session
.get_by_token(old_refresh_token)
.await?
.ok_or_else(|| DomainError::Unauthorized("Invalid refresh token".into()))?;
if session.expires_at < Utc::now() {
ctx.repos.refresh_session.revoke(old_refresh_token).await?;
deps.refresh_session.revoke(old_refresh_token).await?;
return Err(DomainError::Unauthorized("Refresh token expired".into()));
}
// Revoke old token (rotation)
ctx.repos.refresh_session.revoke(old_refresh_token).await?;
deps.refresh_session.revoke(old_refresh_token).await?;
// Generate new access token
let generated = ctx.services.auth.generate_token(&session.user_id).await?;
let generated = deps.auth.generate_token(&session.user_id).await?;
// Create new refresh session
let new_refresh_token = Uuid::new_v4().to_string();
let refresh_expires = Utc::now() + Duration::seconds(ctx.config.refresh_ttl_seconds as i64);
let refresh_expires = Utc::now() + Duration::seconds(deps.config.refresh_ttl_seconds as i64);
let new_session = RefreshSession {
id: Uuid::new_v4(),
user_id: session.user_id,
@@ -43,7 +42,7 @@ pub async fn execute(
expires_at: refresh_expires,
created_at: Utc::now(),
};
ctx.repos.refresh_session.create(&new_session).await?;
deps.refresh_session.create(&new_session).await?;
Ok(RefreshResult {
token: generated.token,

View File

@@ -4,10 +4,10 @@ use domain::{
value_objects::{Email, Password, Username},
};
use crate::{auth::commands::RegisterCommand, context::AppContext};
use crate::auth::{commands::RegisterCommand, deps::RegisterDeps};
pub async fn execute(ctx: &AppContext, cmd: RegisterCommand) -> Result<(), DomainError> {
if !ctx.config.allow_registration {
pub async fn execute(deps: &RegisterDeps, cmd: RegisterCommand) -> Result<(), DomainError> {
if !deps.config.allow_registration {
return Err(DomainError::Unauthorized("Registration is disabled".into()));
}
@@ -15,21 +15,20 @@ pub async fn execute(ctx: &AppContext, cmd: RegisterCommand) -> Result<(), Domai
let email = Email::new(cmd.email)?;
let username = Username::new(cmd.username)?;
if ctx.repos.user.find_by_email(&email).await?.is_some() {
if deps.user.find_by_email(&email).await?.is_some() {
return Err(DomainError::ValidationError(
"Email already registered".into(),
));
}
if ctx.repos.user.find_by_username(&username).await?.is_some() {
if deps.user.find_by_username(&username).await?.is_some() {
return Err(DomainError::ValidationError(
"Username already taken".into(),
));
}
let hash = ctx.services.password_hasher.hash(password.value()).await?;
ctx.repos
.user
let hash = deps.password_hasher.hash(password.value()).await?;
deps.user
.save(&User::new(email, username, hash, cmd.role))
.await
}

View File

@@ -1,18 +1,25 @@
use domain::errors::DomainError;
use crate::{
auth::commands::RegisterAndLoginCommand,
auth::{login, register},
context::AppContext,
use crate::auth::{
commands::{RegisterAndLoginCommand, RegisterCommand},
deps::{LoginDeps, RegisterAndLoginDeps, RegisterDeps},
login::{self, LoginResult},
queries::LoginQuery,
register,
};
pub async fn execute(
ctx: &AppContext,
deps: &RegisterAndLoginDeps,
cmd: RegisterAndLoginCommand,
) -> Result<login::LoginResult, DomainError> {
) -> Result<LoginResult, DomainError> {
let reg_deps = RegisterDeps {
user: deps.user.clone(),
password_hasher: deps.password_hasher.clone(),
config: deps.config.clone(),
};
register::execute(
ctx,
crate::auth::commands::RegisterCommand {
&reg_deps,
RegisterCommand {
email: cmd.email.clone(),
username: cmd.username,
password: cmd.password.clone(),
@@ -21,9 +28,16 @@ pub async fn execute(
)
.await?;
let log_deps = LoginDeps {
user: deps.user.clone(),
password_hasher: deps.password_hasher.clone(),
auth: deps.auth.clone(),
refresh_session: deps.refresh_session.clone(),
config: deps.config.clone(),
};
login::execute(
ctx,
crate::auth::queries::LoginQuery {
&log_deps,
LoginQuery {
email: cmd.email,
password: cmd.password,
},

View File

@@ -4,15 +4,24 @@ use domain::models::UserRole;
use domain::testing::InMemoryUserRepository;
use crate::{
auth::commands::RegisterCommand,
auth::queries::LoginQuery,
auth::{login, register},
auth::{
commands::RegisterCommand,
deps::{LoginDeps, RegisterDeps},
login,
queries::LoginQuery,
register,
},
test_helpers::TestContextBuilder,
};
async fn setup_user(ctx: &crate::context::AppContext, email: &str, password: &str) {
async fn setup_user(b: &TestContextBuilder, email: &str, password: &str) {
let deps = RegisterDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
config: b.config.clone(),
};
register::execute(
ctx,
&deps,
RegisterCommand {
email: email.to_string(),
username: "testuser".to_string(),
@@ -27,14 +36,18 @@ async fn setup_user(ctx: &crate::context::AppContext, email: &str, password: &st
#[tokio::test]
async fn test_login_valid_credentials_returns_token() {
let users = InMemoryUserRepository::new();
let ctx = TestContextBuilder::new()
.with_users(Arc::clone(&users) as _)
.build();
setup_user(&ctx, "carol@example.com", "secret123").await;
let b = TestContextBuilder::new().with_users(Arc::clone(&users) as _);
setup_user(&b, "carol@example.com", "secret123").await;
let deps = LoginDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
auth: b.auth_service.clone(),
refresh_session: b.refresh_session_repo.clone(),
config: b.config.clone(),
};
let result = login::execute(
&ctx,
&deps,
LoginQuery {
email: "carol@example.com".into(),
password: "secret123".into(),
@@ -51,14 +64,18 @@ async fn test_login_valid_credentials_returns_token() {
#[tokio::test]
async fn test_login_wrong_password_fails() {
let users = InMemoryUserRepository::new();
let ctx = TestContextBuilder::new()
.with_users(Arc::clone(&users) as _)
.build();
setup_user(&ctx, "dave@example.com", "correct_password").await;
let b = TestContextBuilder::new().with_users(Arc::clone(&users) as _);
setup_user(&b, "dave@example.com", "correct_password").await;
let deps = LoginDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
auth: b.auth_service.clone(),
refresh_session: b.refresh_session_repo.clone(),
config: b.config.clone(),
};
let result = login::execute(
&ctx,
&deps,
LoginQuery {
email: "dave@example.com".into(),
password: "wrong_password".into(),
@@ -71,10 +88,16 @@ async fn test_login_wrong_password_fails() {
#[tokio::test]
async fn test_login_unknown_email_fails() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let deps = LoginDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
auth: b.auth_service.clone(),
refresh_session: b.refresh_session_repo.clone(),
config: b.config.clone(),
};
let result = login::execute(
&ctx,
&deps,
LoginQuery {
email: "nobody@example.com".into(),
password: "anything".into(),

View File

@@ -4,21 +4,28 @@ use domain::models::UserRole;
use domain::testing::InMemoryUserRepository;
use crate::{
auth::commands::RegisterCommand,
auth::queries::LoginQuery,
auth::{login, logout, refresh, register},
auth::{
commands::RegisterCommand,
deps::{LoginDeps, RefreshDeps, RegisterDeps},
login, logout,
queries::LoginQuery,
refresh, register,
},
test_helpers::TestContextBuilder,
};
#[tokio::test]
async fn logout_revokes_refresh_token() {
let users = InMemoryUserRepository::new();
let ctx = TestContextBuilder::new()
.with_users(Arc::clone(&users) as _)
.build();
let b = TestContextBuilder::new().with_users(Arc::clone(&users) as _);
let reg_deps = RegisterDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
config: b.config.clone(),
};
register::execute(
&ctx,
&reg_deps,
RegisterCommand {
email: "bob@example.com".to_string(),
username: "bob".to_string(),
@@ -29,8 +36,15 @@ async fn logout_revokes_refresh_token() {
.await
.unwrap();
let login_deps = LoginDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
auth: b.auth_service.clone(),
refresh_session: b.refresh_session_repo.clone(),
config: b.config.clone(),
};
let login_result = login::execute(
&ctx,
&login_deps,
LoginQuery {
email: "bob@example.com".into(),
password: "password123".into(),
@@ -39,17 +53,22 @@ async fn logout_revokes_refresh_token() {
.await
.unwrap();
logout::execute(&ctx, &login_result.refresh_token)
logout::execute(b.refresh_session_repo.clone(), &login_result.refresh_token)
.await
.unwrap();
let refresh_attempt = refresh::execute(&ctx, &login_result.refresh_token).await;
let refresh_deps = RefreshDeps {
refresh_session: b.refresh_session_repo.clone(),
auth: b.auth_service.clone(),
config: b.config.clone(),
};
let refresh_attempt = refresh::execute(&refresh_deps, &login_result.refresh_token).await;
assert!(refresh_attempt.is_err());
}
#[tokio::test]
async fn logout_with_unknown_token_succeeds() {
let ctx = TestContextBuilder::new().build();
let result = logout::execute(&ctx, "nonexistent-token").await;
let b = TestContextBuilder::new();
let result = logout::execute(b.refresh_session_repo.clone(), "nonexistent-token").await;
assert!(result.is_ok());
}

View File

@@ -4,15 +4,24 @@ use domain::models::UserRole;
use domain::testing::InMemoryUserRepository;
use crate::{
auth::commands::RegisterCommand,
auth::queries::LoginQuery,
auth::{login, refresh, register},
auth::{
commands::RegisterCommand,
deps::{LoginDeps, RefreshDeps, RegisterDeps},
login,
queries::LoginQuery,
refresh, register,
},
test_helpers::TestContextBuilder,
};
async fn login_user(ctx: &crate::context::AppContext) -> login::LoginResult {
async fn login_user(b: &TestContextBuilder) -> login::LoginResult {
let reg_deps = RegisterDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
config: b.config.clone(),
};
register::execute(
ctx,
&reg_deps,
RegisterCommand {
email: "alice@example.com".to_string(),
username: "alice".to_string(),
@@ -23,8 +32,15 @@ async fn login_user(ctx: &crate::context::AppContext) -> login::LoginResult {
.await
.unwrap();
let login_deps = LoginDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
auth: b.auth_service.clone(),
refresh_session: b.refresh_session_repo.clone(),
config: b.config.clone(),
};
login::execute(
ctx,
&login_deps,
LoginQuery {
email: "alice@example.com".into(),
password: "password123".into(),
@@ -37,13 +53,15 @@ async fn login_user(ctx: &crate::context::AppContext) -> login::LoginResult {
#[tokio::test]
async fn refresh_returns_new_tokens() {
let users = InMemoryUserRepository::new();
let ctx = TestContextBuilder::new()
.with_users(Arc::clone(&users) as _)
.build();
let b = TestContextBuilder::new().with_users(Arc::clone(&users) as _);
let login_result = login_user(&b).await;
let login_result = login_user(&ctx).await;
let result = refresh::execute(&ctx, &login_result.refresh_token)
let deps = RefreshDeps {
refresh_session: b.refresh_session_repo.clone(),
auth: b.auth_service.clone(),
config: b.config.clone(),
};
let result = refresh::execute(&deps, &login_result.refresh_token)
.await
.unwrap();
@@ -55,33 +73,37 @@ async fn refresh_returns_new_tokens() {
#[tokio::test]
async fn refresh_rotates_token_old_one_invalid() {
let users = InMemoryUserRepository::new();
let ctx = TestContextBuilder::new()
.with_users(Arc::clone(&users) as _)
.build();
let login_result = login_user(&ctx).await;
let b = TestContextBuilder::new().with_users(Arc::clone(&users) as _);
let login_result = login_user(&b).await;
let old_token = login_result.refresh_token.clone();
refresh::execute(&ctx, &old_token).await.unwrap();
let deps = RefreshDeps {
refresh_session: b.refresh_session_repo.clone(),
auth: b.auth_service.clone(),
config: b.config.clone(),
};
refresh::execute(&deps, &old_token).await.unwrap();
let retry = refresh::execute(&ctx, &old_token).await;
let retry = refresh::execute(&deps, &old_token).await;
assert!(retry.is_err());
}
#[tokio::test]
async fn refresh_with_new_token_works() {
let users = InMemoryUserRepository::new();
let ctx = TestContextBuilder::new()
.with_users(Arc::clone(&users) as _)
.build();
let b = TestContextBuilder::new().with_users(Arc::clone(&users) as _);
let login_result = login_user(&b).await;
let login_result = login_user(&ctx).await;
let first = refresh::execute(&ctx, &login_result.refresh_token)
let deps = RefreshDeps {
refresh_session: b.refresh_session_repo.clone(),
auth: b.auth_service.clone(),
config: b.config.clone(),
};
let first = refresh::execute(&deps, &login_result.refresh_token)
.await
.unwrap();
let second = refresh::execute(&ctx, &first.refresh_token).await.unwrap();
let second = refresh::execute(&deps, &first.refresh_token).await.unwrap();
assert!(!second.token.is_empty());
assert_ne!(second.refresh_token, first.refresh_token);
@@ -89,8 +111,12 @@ async fn refresh_with_new_token_works() {
#[tokio::test]
async fn refresh_with_unknown_token_fails() {
let ctx = TestContextBuilder::new().build();
let result = refresh::execute(&ctx, "nonexistent-token").await;
let b = TestContextBuilder::new();
let deps = RefreshDeps {
refresh_session: b.refresh_session_repo.clone(),
auth: b.auth_service.clone(),
config: b.config.clone(),
};
let result = refresh::execute(&deps, "nonexistent-token").await;
assert!(result.is_err());
}

View File

@@ -5,7 +5,10 @@ use domain::ports::UserRepository;
use domain::testing::InMemoryUserRepository;
use domain::value_objects::Email;
use crate::{auth::commands::RegisterCommand, auth::register, test_helpers::TestContextBuilder};
use crate::{
auth::{commands::RegisterCommand, deps::RegisterDeps, register},
test_helpers::TestContextBuilder,
};
fn cmd(email: &str) -> RegisterCommand {
RegisterCommand {
@@ -19,11 +22,14 @@ fn cmd(email: &str) -> RegisterCommand {
#[tokio::test]
async fn test_register_creates_user() {
let users = InMemoryUserRepository::new();
let ctx = TestContextBuilder::new()
.with_users(Arc::clone(&users) as _)
.build();
let b = TestContextBuilder::new().with_users(Arc::clone(&users) as _);
let deps = RegisterDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
config: b.config.clone(),
};
register::execute(&ctx, cmd("alice@example.com"))
register::execute(&deps, cmd("alice@example.com"))
.await
.unwrap();
@@ -36,22 +42,30 @@ async fn test_register_creates_user() {
#[tokio::test]
async fn test_register_duplicate_email_fails() {
let users = InMemoryUserRepository::new();
let ctx = TestContextBuilder::new()
.with_users(Arc::clone(&users) as _)
.build();
let b = TestContextBuilder::new().with_users(Arc::clone(&users) as _);
let deps = RegisterDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
config: b.config.clone(),
};
register::execute(&ctx, cmd("bob@example.com"))
register::execute(&deps, cmd("bob@example.com"))
.await
.unwrap();
let result = register::execute(&ctx, cmd("bob@example.com")).await;
let result = register::execute(&deps, cmd("bob@example.com")).await;
assert!(result.is_err(), "duplicate email should fail");
}
#[tokio::test]
async fn test_register_short_password_fails() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let deps = RegisterDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
config: b.config.clone(),
};
let result = register::execute(
&ctx,
&deps,
RegisterCommand {
email: "x@y.com".to_string(),
username: "testuser".to_string(),

View File

@@ -1,13 +1,21 @@
use crate::auth::commands::RegisterAndLoginCommand;
use crate::auth::deps::RegisterAndLoginDeps;
use crate::auth::register_and_login;
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn registers_and_returns_token() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let deps = RegisterAndLoginDeps {
user: b.user_repo.clone(),
password_hasher: b.password_hasher.clone(),
auth: b.auth_service.clone(),
refresh_session: b.refresh_session_repo.clone(),
config: b.config.clone(),
};
let result = register_and_login::execute(
&ctx,
&deps,
RegisterAndLoginCommand {
email: "new@example.com".into(),
username: "newuser".into(),

View File

@@ -1,16 +1,15 @@
use crate::{context::AppContext, diary::commands::DeleteReviewCommand};
use crate::diary::{commands::DeleteReviewCommand, deps::DeleteReviewDeps};
use domain::{
errors::DomainError,
events::DomainEvent,
value_objects::{ReviewId, UserId},
};
pub async fn execute(ctx: &AppContext, cmd: DeleteReviewCommand) -> Result<(), DomainError> {
pub async fn execute(deps: &DeleteReviewDeps, cmd: DeleteReviewCommand) -> Result<(), DomainError> {
let review_id = ReviewId::from_uuid(cmd.review_id);
let requesting_user_id = UserId::from_uuid(cmd.requesting_user_id);
let review = ctx
.repos
let review = deps
.review
.get_review_by_id(&review_id)
.await?
@@ -21,10 +20,9 @@ pub async fn execute(ctx: &AppContext, cmd: DeleteReviewCommand) -> Result<(), D
}
let movie_id = review.movie_id().clone();
ctx.repos.review.delete_review(&review_id).await?;
deps.review.delete_review(&review_id).await?;
if let Err(e) = ctx
.services
if let Err(e) = deps
.event_publisher
.publish(&DomainEvent::ReviewDeleted {
review_id: review_id.clone(),
@@ -35,13 +33,12 @@ pub async fn execute(ctx: &AppContext, cmd: DeleteReviewCommand) -> Result<(), D
tracing::warn!("failed to publish ReviewDeleted: {e}");
}
let history = ctx.repos.diary.get_review_history(&movie_id).await?;
let history = deps.diary.get_review_history(&movie_id).await?;
if history.viewings().is_empty() {
let poster_path = history.movie().poster_path().cloned();
ctx.repos.movie.delete_movie(&movie_id).await?;
deps.movie.delete_movie(&movie_id).await?;
// best-effort: movie is already deleted, so publish failure is non-fatal
if let Err(e) = ctx
.services
if let Err(e) = deps
.event_publisher
.publish(&DomainEvent::MovieDeleted {
movie_id,

View File

@@ -0,0 +1,27 @@
use std::sync::Arc;
use domain::ports::{
DiaryRepository, EventPublisher, MovieProfileRepository, MovieRepository, ReviewRepository,
SocialQueryPort,
};
use crate::config::AppConfig;
pub struct DeleteReviewDeps {
pub review: Arc<dyn ReviewRepository>,
pub diary: Arc<dyn DiaryRepository>,
pub movie: Arc<dyn MovieRepository>,
pub event_publisher: Arc<dyn EventPublisher>,
}
pub struct GetMovieSocialPageDeps {
pub movie: Arc<dyn MovieRepository>,
pub diary: Arc<dyn DiaryRepository>,
pub movie_profile: Arc<dyn MovieProfileRepository>,
}
pub struct GetActivityFeedDeps {
pub diary: Arc<dyn DiaryRepository>,
pub social_query: Arc<dyn SocialQueryPort>,
pub config: AppConfig,
}

View File

@@ -1,15 +1,22 @@
use domain::{errors::DomainError, value_objects::UserId};
use std::sync::Arc;
use crate::{context::AppContext, diary::queries::ExportQuery};
use domain::{
errors::DomainError,
ports::{DiaryExporter, DiaryRepository},
value_objects::UserId,
};
pub async fn execute(ctx: &AppContext, query: ExportQuery) -> Result<Vec<u8>, DomainError> {
let entries = ctx
.repos
.diary
use crate::diary::queries::ExportQuery;
pub async fn execute(
diary: &Arc<dyn DiaryRepository>,
diary_exporter: &Arc<dyn DiaryExporter>,
query: ExportQuery,
) -> Result<Vec<u8>, DomainError> {
let entries = diary
.get_user_history(&UserId::from_uuid(query.user_id))
.await?;
ctx.services
.diary_exporter
diary_exporter
.serialize_entries(&entries, query.format)
.await
}

View File

@@ -1,4 +1,4 @@
use crate::{context::AppContext, diary::queries::GetActivityFeedQuery};
use crate::diary::{deps::GetActivityFeedDeps, queries::GetActivityFeedQuery};
use domain::{
errors::DomainError,
models::{
@@ -9,15 +9,14 @@ use domain::{
};
pub async fn execute(
ctx: &AppContext,
deps: &GetActivityFeedDeps,
query: GetActivityFeedQuery,
) -> Result<Paginated<FeedEntry>, DomainError> {
let page = PageParams::new(Some(query.limit), Some(query.offset))?;
let following = build_following_filter(ctx, &query).await;
let following = build_following_filter(deps, &query).await;
ctx.repos
.diary
deps.diary
.query_activity_feed_filtered(
&page,
&query.sort_by,
@@ -28,15 +27,14 @@ pub async fn execute(
}
async fn build_following_filter(
ctx: &AppContext,
deps: &GetActivityFeedDeps,
query: &GetActivityFeedQuery,
) -> Option<FollowingFilter> {
if !query.filter_following {
return None;
}
let viewer_id = query.viewer_user_id?;
let urls = ctx
.repos
let urls = deps
.social_query
.get_accepted_following_urls(viewer_id)
.await
@@ -47,7 +45,7 @@ async fn build_following_filter(
remote_actor_urls: vec![],
});
}
let base_url = &ctx.config.base_url;
let base_url = &deps.config.base_url;
let mut local_ids = vec![viewer_id];
let mut remote_urls = Vec::new();
for url in urls {

View File

@@ -1,16 +1,19 @@
use std::sync::Arc;
use domain::{
errors::DomainError,
models::{
DiaryEntry, DiaryFilter, SortDirection,
collections::{PageParams, Paginated},
},
ports::DiaryRepository,
value_objects::{MovieId, UserId},
};
use crate::{context::AppContext, diary::queries::GetDiaryQuery};
use crate::diary::queries::GetDiaryQuery;
pub async fn execute(
ctx: &AppContext,
diary: &Arc<dyn DiaryRepository>,
query: GetDiaryQuery,
) -> Result<Paginated<DiaryEntry>, DomainError> {
let page = PageParams::new(query.limit, query.offset)?;
@@ -25,7 +28,7 @@ pub async fn execute(
search: None,
};
ctx.repos.diary.query_diary(&filter).await
diary.query_diary(&filter).await
}
#[cfg(test)]

View File

@@ -7,7 +7,7 @@ use domain::{
value_objects::MovieId,
};
use crate::{context::AppContext, diary::queries::GetMovieSocialPageQuery};
use crate::diary::{deps::GetMovieSocialPageDeps, queries::GetMovieSocialPageQuery};
pub struct MovieSocialPageResult {
pub movie: Movie,
@@ -17,23 +17,22 @@ pub struct MovieSocialPageResult {
}
pub async fn execute(
ctx: &AppContext,
deps: &GetMovieSocialPageDeps,
query: GetMovieSocialPageQuery,
) -> Result<MovieSocialPageResult, DomainError> {
let movie_id = MovieId::from_uuid(query.movie_id);
let page = PageParams::new(Some(query.limit), Some(query.offset))?;
let movie = ctx
.repos
let movie = deps
.movie
.get_movie_by_id(&movie_id)
.await?
.ok_or_else(|| DomainError::NotFound(format!("Movie {}", query.movie_id)))?;
let (stats, reviews, profile) = tokio::try_join!(
ctx.repos.diary.get_movie_stats(&movie_id),
ctx.repos.diary.get_movie_social_feed(&movie_id, &page),
ctx.repos.movie_profile.get_by_movie_id(&movie_id),
deps.diary.get_movie_stats(&movie_id),
deps.diary.get_movie_social_feed(&movie_id, &page),
deps.movie_profile.get_by_movie_id(&movie_id),
)?;
Ok(MovieSocialPageResult {

View File

@@ -1,19 +1,22 @@
use std::sync::Arc;
use domain::{
errors::DomainError,
models::ReviewHistory,
ports::DiaryRepository,
services::review_history::{ReviewHistoryAnalyzer, Trend},
value_objects::MovieId,
};
use crate::{context::AppContext, diary::queries::GetReviewHistoryQuery};
use crate::diary::queries::GetReviewHistoryQuery;
pub async fn execute(
ctx: &AppContext,
diary: &Arc<dyn DiaryRepository>,
query: GetReviewHistoryQuery,
) -> Result<(ReviewHistory, Trend), DomainError> {
let movie_id = MovieId::from_uuid(query.movie_id);
let mut history = ctx.repos.diary.get_review_history(&movie_id).await?;
let mut history = diary.get_review_history(&movie_id).await?;
let trend = ReviewHistoryAnalyzer::rating_trend(&history)?;

View File

@@ -1,9 +1,14 @@
use std::sync::Arc;
use domain::errors::DomainError;
use crate::{context::AppContext, diary::commands::LogReviewCommand};
use crate::{diary::commands::LogReviewCommand, ports::ReviewLogger};
pub async fn execute(ctx: &AppContext, cmd: LogReviewCommand) -> Result<(), DomainError> {
ctx.services.review_logger.log_review(cmd).await
pub async fn execute(
review_logger: &Arc<dyn ReviewLogger>,
cmd: LogReviewCommand,
) -> Result<(), DomainError> {
review_logger.log_review(cmd).await
}
#[cfg(test)]

View File

@@ -1,5 +1,6 @@
pub mod commands;
pub mod delete_review;
pub mod deps;
pub mod export_diary;
pub mod get_activity_feed;
pub mod get_diary;

View File

@@ -12,7 +12,7 @@ use domain::{
};
use crate::{
diary::commands::DeleteReviewCommand, diary::delete_review, test_helpers::TestContextBuilder,
diary::commands::DeleteReviewCommand, diary::delete_review, diary::deps::DeleteReviewDeps,
};
fn make_movie() -> Movie {
@@ -51,15 +51,15 @@ async fn test_delete_review_removes_it() {
reviews.save_review(&review).await.unwrap();
diary.seed_history(movie.clone(), vec![]);
let ctx = TestContextBuilder::new()
.with_movies(Arc::clone(&movies) as _)
.with_reviews(Arc::clone(&reviews) as _)
.with_diary(Arc::clone(&diary) as _)
.with_event_publisher(Arc::clone(&events) as _)
.build();
let deps = DeleteReviewDeps {
review: Arc::clone(&reviews) as _,
diary: diary.clone() as _,
movie: Arc::clone(&movies) as _,
event_publisher: Arc::clone(&events) as _,
};
delete_review::execute(
&ctx,
&deps,
DeleteReviewCommand {
review_id: review.id().value(),
requesting_user_id: user_id.value(),
@@ -78,6 +78,9 @@ async fn test_delete_review_removes_it() {
#[tokio::test]
async fn test_delete_review_wrong_user_is_unauthorized() {
let reviews = InMemoryReviewRepository::new();
let diary = FakeDiaryRepository::new();
let movies = InMemoryMovieRepository::new();
let events = NoopEventPublisher::new();
let movie_id = MovieId::from_uuid(uuid::Uuid::new_v4());
let owner_id = UserId::from_uuid(uuid::Uuid::new_v4());
@@ -86,12 +89,15 @@ async fn test_delete_review_wrong_user_is_unauthorized() {
reviews.save_review(&review).await.unwrap();
let ctx = TestContextBuilder::new()
.with_reviews(Arc::clone(&reviews) as _)
.build();
let deps = DeleteReviewDeps {
review: Arc::clone(&reviews) as _,
diary: diary as _,
movie: movies as _,
event_publisher: Arc::clone(&events) as _,
};
let result = delete_review::execute(
&ctx,
&deps,
DeleteReviewCommand {
review_id: review.id().value(),
requesting_user_id: other_id,

View File

@@ -2,18 +2,27 @@ use std::sync::Arc;
use async_trait::async_trait;
use domain::errors::DomainError;
use domain::testing::{FakeDiaryRepository, NoopSocialQueryPort};
use crate::{
diary::get_activity_feed, diary::queries::GetActivityFeedQuery,
test_helpers::TestContextBuilder,
config::AppConfig, diary::deps::GetActivityFeedDeps, diary::get_activity_feed,
diary::queries::GetActivityFeedQuery, test_helpers::TestContextBuilder,
};
fn default_deps() -> GetActivityFeedDeps {
GetActivityFeedDeps {
diary: FakeDiaryRepository::new() as _,
social_query: Arc::new(NoopSocialQueryPort),
config: TestContextBuilder::new().config,
}
}
#[tokio::test]
async fn returns_empty_feed() {
let ctx = TestContextBuilder::new().build();
let deps = default_deps();
let result = get_activity_feed::execute(
&ctx,
&deps,
GetActivityFeedQuery {
limit: 10,
offset: 0,
@@ -32,12 +41,12 @@ async fn returns_empty_feed() {
#[tokio::test]
async fn returns_feed_with_following_filter() {
let ctx = TestContextBuilder::new().build();
let deps = default_deps();
let viewer = uuid::Uuid::new_v4();
let result = get_activity_feed::execute(
&ctx,
&deps,
GetActivityFeedQuery {
limit: 10,
offset: 0,
@@ -93,12 +102,24 @@ async fn following_filter_parses_local_and_remote_urls() {
let social = Arc::new(FakeSocialWithFollowing(following_urls));
let ctx = TestContextBuilder::new()
.with_social_query(social as _)
.build();
let deps = GetActivityFeedDeps {
diary: FakeDiaryRepository::new() as _,
social_query: social as _,
config: AppConfig {
allow_registration: true,
base_url: "http://localhost:3000".into(),
rate_limit: 20,
refresh_ttl_seconds: 2_592_000,
wrapup: crate::config::WrapUpConfig {
font_path: None,
logo_path: None,
bg_dir: None,
},
},
};
let result = get_activity_feed::execute(
&ctx,
&deps,
GetActivityFeedQuery {
limit: 10,
offset: 0,
@@ -118,10 +139,10 @@ async fn following_filter_parses_local_and_remote_urls() {
#[tokio::test]
async fn following_filter_without_viewer_returns_none() {
let ctx = TestContextBuilder::new().build();
let deps = default_deps();
let result = get_activity_feed::execute(
&ctx,
&deps,
GetActivityFeedQuery {
limit: 10,
offset: 0,

View File

@@ -1,11 +1,14 @@
use crate::{diary::get_diary, diary::queries::GetDiaryQuery, test_helpers::TestContextBuilder};
use domain::testing::FakeDiaryRepository;
use std::sync::Arc;
use crate::{diary::get_diary, diary::queries::GetDiaryQuery};
#[tokio::test]
async fn returns_empty_page() {
let ctx = TestContextBuilder::new().build();
let diary = FakeDiaryRepository::new() as Arc<dyn domain::ports::DiaryRepository>;
let result = get_diary::execute(
&ctx,
&diary,
GetDiaryQuery {
limit: None,
offset: None,

View File

@@ -5,21 +5,25 @@ use uuid::Uuid;
use domain::{
models::Movie,
ports::MovieRepository,
testing::InMemoryMovieRepository,
testing::{FakeDiaryRepository, InMemoryMovieProfileRepository, InMemoryMovieRepository},
value_objects::{MovieTitle, ReleaseYear},
};
use crate::{
diary::get_movie_social_page, diary::queries::GetMovieSocialPageQuery,
test_helpers::TestContextBuilder,
diary::deps::GetMovieSocialPageDeps, diary::get_movie_social_page,
diary::queries::GetMovieSocialPageQuery,
};
#[tokio::test]
async fn fails_when_movie_not_found() {
let ctx = TestContextBuilder::new().build();
let deps = GetMovieSocialPageDeps {
movie: InMemoryMovieRepository::new(),
diary: FakeDiaryRepository::new() as _,
movie_profile: InMemoryMovieProfileRepository::new(),
};
let result = get_movie_social_page::execute(
&ctx,
&deps,
GetMovieSocialPageQuery {
movie_id: Uuid::new_v4(),
limit: 10,
@@ -45,12 +49,14 @@ async fn returns_movie_social_page() {
let movie_uuid = movie.id().value();
movies.upsert_movie(&movie).await.unwrap();
let ctx = TestContextBuilder::new()
.with_movies(Arc::clone(&movies) as _)
.build();
let deps = GetMovieSocialPageDeps {
movie: Arc::clone(&movies) as _,
diary: FakeDiaryRepository::new() as _,
movie_profile: InMemoryMovieProfileRepository::new(),
};
let result = get_movie_social_page::execute(
&ctx,
&deps,
GetMovieSocialPageQuery {
movie_id: movie_uuid,
limit: 10,

View File

@@ -1,13 +1,13 @@
use std::sync::Arc;
use domain::{
models::Movie,
ports::DiaryRepository,
services::review_history::Trend,
value_objects::{MovieTitle, ReleaseYear},
};
use crate::{
diary::get_review_history, diary::queries::GetReviewHistoryQuery,
test_helpers::TestContextBuilder,
};
use crate::{diary::get_review_history, diary::queries::GetReviewHistoryQuery};
#[tokio::test]
async fn returns_empty_history() {
@@ -22,10 +22,9 @@ async fn returns_empty_history() {
let diary = domain::testing::FakeDiaryRepository::new();
diary.seed_history(movie, vec![]);
let diary: Arc<dyn DiaryRepository> = diary;
let ctx = TestContextBuilder::new().with_diary(diary as _).build();
let (history, trend) = get_review_history::execute(&ctx, GetReviewHistoryQuery { movie_id })
let (history, trend) = get_review_history::execute(&diary, GetReviewHistoryQuery { movie_id })
.await
.unwrap();

View File

@@ -17,24 +17,18 @@ use crate::{
test_helpers::TestContextBuilder,
};
fn build_ctx_with_real_logger(
fn build_logger(
movies: &Arc<InMemoryMovieRepository>,
reviews: &Arc<InMemoryReviewRepository>,
events: &Arc<NoopEventPublisher>,
) -> crate::context::AppContext {
let logger = Arc::new(DefaultReviewLogger::new(
) -> Arc<dyn crate::ports::ReviewLogger> {
Arc::new(DefaultReviewLogger::new(
Arc::clone(movies) as _,
Arc::clone(reviews) as _,
crate::test_helpers::TestContextBuilder::new().watchlist_repo,
TestContextBuilder::new().watchlist_repo,
Arc::new(domain::testing::FakeMetadataClient) as _,
Arc::clone(events) as _,
));
TestContextBuilder::new()
.with_movies(Arc::clone(movies) as _)
.with_reviews(Arc::clone(reviews) as _)
.with_event_publisher(Arc::clone(events) as _)
.with_review_logger(logger)
.build()
))
}
fn movie_input_manual(title: &str, year: u16) -> MovieInput {
@@ -62,7 +56,7 @@ async fn test_log_review_creates_movie_and_review() {
let movies = InMemoryMovieRepository::new();
let reviews = InMemoryReviewRepository::new();
let events = NoopEventPublisher::new();
let ctx = build_ctx_with_real_logger(&movies, &reviews, &events);
let logger = build_logger(&movies, &reviews, &events);
let user_id = uuid::Uuid::new_v4();
let cmd = LogReviewCommand {
@@ -73,7 +67,7 @@ async fn test_log_review_creates_movie_and_review() {
watched_at: Utc::now().naive_utc(),
};
log_review::execute(&ctx, cmd).await.unwrap();
log_review::execute(&logger, cmd).await.unwrap();
assert_eq!(reviews.count(), 1, "review should be saved");
assert!(!events.published().is_empty(), "events should be published");
@@ -95,7 +89,7 @@ async fn test_log_review_reuses_existing_movie() {
movies.upsert_movie(&existing_movie).await.unwrap();
let events = NoopEventPublisher::new();
let ctx = build_ctx_with_real_logger(&movies, &reviews, &events);
let logger = build_logger(&movies, &reviews, &events);
let cmd = LogReviewCommand {
user_id: uuid::Uuid::new_v4(),
@@ -105,7 +99,7 @@ async fn test_log_review_reuses_existing_movie() {
watched_at: Utc::now().naive_utc(),
};
log_review::execute(&ctx, cmd).await.unwrap();
log_review::execute(&logger, cmd).await.unwrap();
assert_eq!(movies.count(), 1, "no duplicate movie");
assert_eq!(reviews.count(), 1);
@@ -116,7 +110,8 @@ async fn test_log_review_with_invalid_rating_fails() {
let movies = InMemoryMovieRepository::new();
let reviews = InMemoryReviewRepository::new();
let events = NoopEventPublisher::new();
let ctx = build_ctx_with_real_logger(&movies, &reviews, &events);
let logger = build_logger(&movies, &reviews, &events);
let cmd = LogReviewCommand {
user_id: uuid::Uuid::new_v4(),
input: movie_input_manual("Some Film", 2000),
@@ -124,6 +119,6 @@ async fn test_log_review_with_invalid_rating_fails() {
comment: None,
watched_at: Utc::now().naive_utc(),
};
let result = log_review::execute(&ctx, cmd).await;
let result = log_review::execute(&logger, cmd).await;
assert!(result.is_err(), "rating > 5 should fail");
}

View File

@@ -1,48 +1,42 @@
use std::sync::Arc;
use domain::{
errors::DomainError,
events::DomainEvent,
models::{Goal, GoalType, GoalWithProgress},
ports::{EventPublisher, GoalRepository},
value_objects::UserId,
};
use super::commands::CreateGoalCommand;
use crate::context::AppContext;
pub async fn execute(
ctx: &AppContext,
goal: Arc<dyn GoalRepository>,
event_publisher: Arc<dyn EventPublisher>,
cmd: CreateGoalCommand,
) -> Result<GoalWithProgress, DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
let existing = ctx
.repos
.goal
.find_by_user_and_year(&user_id, cmd.year)
.await?;
let existing = goal.find_by_user_and_year(&user_id, cmd.year).await?;
if existing.is_some() {
return Err(DomainError::ValidationError(
"Goal already exists for this year".into(),
));
}
let goal = Goal::new(
let g = Goal::new(
user_id.clone(),
cmd.year,
cmd.target_count,
GoalType::Movies,
)?;
ctx.repos.goal.save(&goal).await?;
goal.save(&g).await?;
let current_count = ctx
.repos
.goal
.count_reviews_in_year(&user_id, cmd.year)
.await?;
let current_count = goal.count_reviews_in_year(&user_id, cmd.year).await?;
ctx.services
.event_publisher
event_publisher
.publish(&DomainEvent::GoalCreated {
goal_id: goal.id().clone(),
goal_id: g.id().clone(),
user_id,
year: cmd.year,
target_count: cmd.target_count,
@@ -50,7 +44,7 @@ pub async fn execute(
.await?;
Ok(GoalWithProgress {
goal,
goal: g,
current_count,
})
}

View File

@@ -1,24 +1,31 @@
use domain::{errors::DomainError, events::DomainEvent, value_objects::UserId};
use std::sync::Arc;
use domain::{
errors::DomainError,
events::DomainEvent,
ports::{EventPublisher, GoalRepository},
value_objects::UserId,
};
use super::commands::DeleteGoalCommand;
use crate::context::AppContext;
pub async fn execute(ctx: &AppContext, cmd: DeleteGoalCommand) -> Result<(), DomainError> {
pub async fn execute(
goal: Arc<dyn GoalRepository>,
event_publisher: Arc<dyn EventPublisher>,
cmd: DeleteGoalCommand,
) -> Result<(), DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
let goal = ctx
.repos
.goal
let g = goal
.find_by_user_and_year(&user_id, cmd.year)
.await?
.ok_or_else(|| DomainError::NotFound(format!("Goal for year {}", cmd.year)))?;
ctx.repos.goal.delete(goal.id(), &user_id).await?;
goal.delete(g.id(), &user_id).await?;
ctx.services
.event_publisher
event_publisher
.publish(&DomainEvent::GoalDeleted {
goal_id: goal.id().clone(),
goal_id: g.id().clone(),
user_id,
year: cmd.year,
})

View File

@@ -1,30 +1,25 @@
use domain::{errors::DomainError, models::GoalWithProgress, value_objects::UserId};
use std::sync::Arc;
use domain::{
errors::DomainError, models::GoalWithProgress, ports::GoalRepository, value_objects::UserId,
};
use super::queries::GetGoalQuery;
use crate::context::AppContext;
pub async fn execute(
ctx: &AppContext,
goal: Arc<dyn GoalRepository>,
query: GetGoalQuery,
) -> Result<Option<GoalWithProgress>, DomainError> {
let user_id = UserId::from_uuid(query.user_id);
let goal = ctx
.repos
.goal
.find_by_user_and_year(&user_id, query.year)
.await?;
let found = goal.find_by_user_and_year(&user_id, query.year).await?;
let Some(goal) = goal else { return Ok(None) };
let Some(g) = found else { return Ok(None) };
let current_count = ctx
.repos
.goal
.count_reviews_in_year(&user_id, query.year)
.await?;
let current_count = goal.count_reviews_in_year(&user_id, query.year).await?;
Ok(Some(GoalWithProgress {
goal,
goal: g,
current_count,
}))
}

View File

@@ -1,24 +1,23 @@
use domain::{errors::DomainError, models::GoalWithProgress, value_objects::UserId};
use std::sync::Arc;
use domain::{
errors::DomainError, models::GoalWithProgress, ports::GoalRepository, value_objects::UserId,
};
use super::queries::ListGoalsQuery;
use crate::context::AppContext;
pub async fn execute(
ctx: &AppContext,
goal: Arc<dyn GoalRepository>,
query: ListGoalsQuery,
) -> Result<Vec<GoalWithProgress>, DomainError> {
let user_id = UserId::from_uuid(query.user_id);
let goals = ctx.repos.goal.list_for_user(&user_id).await?;
let goals = goal.list_for_user(&user_id).await?;
let mut result = Vec::with_capacity(goals.len());
for goal in goals {
let current_count = ctx
.repos
.goal
.count_reviews_in_year(&user_id, goal.year())
.await?;
for g in goals {
let current_count = goal.count_reviews_in_year(&user_id, g.year()).await?;
result.push(GoalWithProgress {
goal,
goal: g,
current_count,
});
}

View File

@@ -10,15 +10,11 @@ use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn creates_goal_and_returns_progress() {
let goals = InMemoryGoalRepository::new();
goals.set_review_count(Uuid::nil(), 2025, 5);
let events = NoopEventPublisher::new();
let ctx = TestContextBuilder::new()
.with_goal(Arc::clone(&goals) as _)
.with_event_publisher(Arc::clone(&events) as _)
.build();
let result = create::execute(
&ctx,
Arc::clone(&goals) as _,
Arc::clone(&events) as _,
CreateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
@@ -30,19 +26,40 @@ async fn creates_goal_and_returns_progress() {
assert_eq!(result.goal.year(), 2025);
assert_eq!(result.goal.target_count(), 50);
assert_eq!(result.current_count, 0);
assert_eq!(goals.count(), 1);
}
#[tokio::test]
async fn creates_goal_with_review_count() {
let goals = InMemoryGoalRepository::new();
goals.set_review_count(Uuid::nil(), 2025, 5);
let events = NoopEventPublisher::new();
let result = create::execute(
Arc::clone(&goals) as _,
Arc::clone(&events) as _,
CreateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
target_count: 50,
},
)
.await
.unwrap();
assert_eq!(result.current_count, 5);
assert_eq!(goals.count(), 1);
}
#[tokio::test]
async fn emits_goal_created_event() {
let b = TestContextBuilder::new();
let events = NoopEventPublisher::new();
let ctx = TestContextBuilder::new()
.with_event_publisher(Arc::clone(&events) as _)
.build();
create::execute(
&ctx,
b.goal_repo.clone(),
Arc::clone(&events) as _,
CreateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
@@ -62,17 +79,20 @@ async fn emits_goal_created_event() {
#[tokio::test]
async fn rejects_duplicate_year() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let cmd = CreateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
target_count: 10,
};
create::execute(&ctx, cmd).await.unwrap();
create::execute(b.goal_repo.clone(), b.event_publisher.clone(), cmd)
.await
.unwrap();
let result = create::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
CreateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
@@ -86,9 +106,10 @@ async fn rejects_duplicate_year() {
#[tokio::test]
async fn rejects_year_before_2020() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let result = create::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
CreateGoalCommand {
user_id: Uuid::nil(),
year: 2019,
@@ -102,9 +123,10 @@ async fn rejects_year_before_2020() {
#[tokio::test]
async fn rejects_zero_target() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let result = create::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
CreateGoalCommand {
user_id: Uuid::nil(),
year: 2025,

View File

@@ -13,13 +13,10 @@ use crate::test_helpers::TestContextBuilder;
async fn deletes_existing_goal() {
let goals = InMemoryGoalRepository::new();
let events = NoopEventPublisher::new();
let ctx = TestContextBuilder::new()
.with_goal(Arc::clone(&goals) as _)
.with_event_publisher(Arc::clone(&events) as _)
.build();
create::execute(
&ctx,
Arc::clone(&goals) as _,
Arc::clone(&events) as _,
CreateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
@@ -31,7 +28,8 @@ async fn deletes_existing_goal() {
assert_eq!(goals.count(), 1);
delete::execute(
&ctx,
Arc::clone(&goals) as _,
Arc::clone(&events) as _,
DeleteGoalCommand {
user_id: Uuid::nil(),
year: 2025,
@@ -45,9 +43,10 @@ async fn deletes_existing_goal() {
#[tokio::test]
async fn fails_when_not_found() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let result = delete::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
DeleteGoalCommand {
user_id: Uuid::nil(),
year: 2025,

View File

@@ -5,9 +5,10 @@ use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn returns_goal_when_exists() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
create::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
CreateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
@@ -18,7 +19,7 @@ async fn returns_goal_when_exists() {
.unwrap();
let result = get::execute(
&ctx,
b.goal_repo.clone(),
GetGoalQuery {
user_id: Uuid::nil(),
year: 2025,
@@ -33,9 +34,9 @@ async fn returns_goal_when_exists() {
#[tokio::test]
async fn returns_none_when_missing() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let result = get::execute(
&ctx,
b.goal_repo.clone(),
GetGoalQuery {
user_id: Uuid::nil(),
year: 2025,

View File

@@ -5,9 +5,9 @@ use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn returns_empty_when_no_goals() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let result = list::execute(
&ctx,
b.goal_repo.clone(),
ListGoalsQuery {
user_id: Uuid::nil(),
},
@@ -20,10 +20,11 @@ async fn returns_empty_when_no_goals() {
#[tokio::test]
async fn returns_all_goals_for_user() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
for year in [2023, 2024, 2025] {
create::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
CreateGoalCommand {
user_id: Uuid::nil(),
year,
@@ -35,7 +36,7 @@ async fn returns_all_goals_for_user() {
}
let result = list::execute(
&ctx,
b.goal_repo.clone(),
ListGoalsQuery {
user_id: Uuid::nil(),
},

View File

@@ -8,9 +8,10 @@ use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn updates_target_count() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
create::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
CreateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
@@ -21,7 +22,8 @@ async fn updates_target_count() {
.unwrap();
let result = update::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
UpdateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
@@ -36,9 +38,10 @@ async fn updates_target_count() {
#[tokio::test]
async fn fails_when_goal_not_found() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let result = update::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
UpdateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
@@ -52,9 +55,10 @@ async fn fails_when_goal_not_found() {
#[tokio::test]
async fn rejects_zero_target() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
create::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
CreateGoalCommand {
user_id: Uuid::nil(),
year: 2025,
@@ -65,7 +69,8 @@ async fn rejects_zero_target() {
.unwrap();
let result = update::execute(
&ctx,
b.goal_repo.clone(),
b.event_publisher.clone(),
UpdateGoalCommand {
user_id: Uuid::nil(),
year: 2025,

View File

@@ -1,36 +1,35 @@
use std::sync::Arc;
use domain::{
errors::DomainError, events::DomainEvent, models::GoalWithProgress, value_objects::UserId,
errors::DomainError,
events::DomainEvent,
models::GoalWithProgress,
ports::{EventPublisher, GoalRepository},
value_objects::UserId,
};
use super::commands::UpdateGoalCommand;
use crate::context::AppContext;
pub async fn execute(
ctx: &AppContext,
goal: Arc<dyn GoalRepository>,
event_publisher: Arc<dyn EventPublisher>,
cmd: UpdateGoalCommand,
) -> Result<GoalWithProgress, DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
let mut goal = ctx
.repos
.goal
let mut g = goal
.find_by_user_and_year(&user_id, cmd.year)
.await?
.ok_or_else(|| DomainError::NotFound(format!("Goal for year {}", cmd.year)))?;
goal.update_target(cmd.target_count)?;
ctx.repos.goal.update(&goal).await?;
g.update_target(cmd.target_count)?;
goal.update(&g).await?;
let current_count = ctx
.repos
.goal
.count_reviews_in_year(&user_id, cmd.year)
.await?;
let current_count = goal.count_reviews_in_year(&user_id, cmd.year).await?;
ctx.services
.event_publisher
event_publisher
.publish(&DomainEvent::GoalUpdated {
goal_id: goal.id().clone(),
goal_id: g.id().clone(),
user_id,
year: cmd.year,
target_count: cmd.target_count,
@@ -38,7 +37,7 @@ pub async fn execute(
.await?;
Ok(GoalWithProgress {
goal,
goal: g,
current_count,
})
}

View File

@@ -1,21 +1,24 @@
use std::sync::Arc;
use domain::{
errors::DomainError,
models::{AnnotatedRow, import::RowResult},
ports::{DocumentParser, ImportSessionRepository, MovieRepository},
value_objects::{ExternalMetadataId, ImportSessionId, MovieTitle, ReleaseYear, UserId},
};
use crate::{context::AppContext, import::commands::ApplyImportMappingCommand};
use crate::import::commands::ApplyImportMappingCommand;
pub async fn execute(
ctx: &AppContext,
import_session: Arc<dyn ImportSessionRepository>,
document_parser: Arc<dyn DocumentParser>,
movie: Arc<dyn MovieRepository>,
cmd: ApplyImportMappingCommand,
) -> Result<Vec<AnnotatedRow>, DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
let session_id = ImportSessionId::from_uuid(cmd.session_id);
let mappings = cmd.mappings;
let mut session = ctx
.repos
.import_session
let mut session = import_session
.get(&session_id, &user_id)
.await?
.ok_or_else(|| DomainError::NotFound("import session".into()))?;
@@ -25,22 +28,22 @@ pub async fn execute(
.clone()
.ok_or_else(|| DomainError::ValidationError("session has no parsed file".into()))?;
let mut annotated = ctx
.services
.document_parser
.apply_mapping(&parsed, &mappings);
let mut annotated = document_parser.apply_mapping(&parsed, &mappings);
mark_duplicates(ctx, &mut annotated).await?;
mark_duplicates(movie, &mut annotated).await?;
session.field_mappings = Some(mappings);
session.row_results = Some(annotated.clone());
ctx.repos.import_session.update(&session).await?;
import_session.update(&session).await?;
Ok(annotated)
}
async fn mark_duplicates(ctx: &AppContext, rows: &mut [AnnotatedRow]) -> Result<(), DomainError> {
async fn mark_duplicates(
movie: Arc<dyn MovieRepository>,
rows: &mut [AnnotatedRow],
) -> Result<(), DomainError> {
let mut ext_ids = Vec::new();
let mut title_year_pairs = Vec::new();
@@ -63,12 +66,8 @@ async fn mark_duplicates(ctx: &AppContext, rows: &mut [AnnotatedRow]) -> Result<
}
}
let known_ext = ctx.repos.movie.existing_external_ids(&ext_ids).await?;
let known_ty = ctx
.repos
.movie
.existing_title_year_pairs(&title_year_pairs)
.await?;
let known_ext = movie.existing_external_ids(&ext_ids).await?;
let known_ty = movie.existing_title_year_pairs(&title_year_pairs).await?;
for row in rows.iter_mut() {
if let RowResult::Valid(ref r) = row.result {

View File

@@ -1,31 +1,34 @@
use crate::{context::AppContext, import::commands::ApplyImportProfileCommand};
use std::sync::Arc;
use crate::import::commands::ApplyImportProfileCommand;
use domain::{
errors::DomainError,
ports::{ImportProfileRepository, ImportSessionRepository},
value_objects::{ImportProfileId, ImportSessionId, UserId},
};
/// Copies the profile's field_mappings onto the session. Caller must then invoke
/// apply_import_mapping to regenerate row_results with the new mappings.
pub async fn execute(ctx: &AppContext, cmd: ApplyImportProfileCommand) -> Result<(), DomainError> {
pub async fn execute(
import_profile: Arc<dyn ImportProfileRepository>,
import_session: Arc<dyn ImportSessionRepository>,
cmd: ApplyImportProfileCommand,
) -> Result<(), DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
let session_id = ImportSessionId::from_uuid(cmd.session_id);
let profile_id = ImportProfileId::from_uuid(cmd.profile_id);
let profile = ctx
.repos
.import_profile
let profile = import_profile
.get(&profile_id, &user_id)
.await?
.ok_or_else(|| DomainError::NotFound("import profile".into()))?;
let mut session = ctx
.repos
.import_session
let mut session = import_session
.get(&session_id, &user_id)
.await?
.ok_or_else(|| DomainError::NotFound("import session".into()))?;
session.field_mappings = Some(profile.field_mappings);
session.row_results = None;
ctx.repos.import_session.update(&session).await
import_session.update(&session).await
}
#[cfg(test)]

View File

@@ -1,8 +1,9 @@
use crate::context::AppContext;
use domain::errors::DomainError;
use std::sync::Arc;
pub async fn execute(ctx: &AppContext) -> Result<u64, DomainError> {
ctx.repos.import_session.delete_expired().await
use domain::{errors::DomainError, ports::ImportSessionRepository};
pub async fn execute(import_session: Arc<dyn ImportSessionRepository>) -> Result<u64, DomainError> {
import_session.delete_expired().await
}
#[cfg(test)]

View File

@@ -1,11 +1,14 @@
use std::sync::Arc;
use chrono::Utc;
use domain::{
errors::DomainError,
models::ImportSession,
ports::{DocumentParser, ImportSessionRepository},
value_objects::{ImportSessionId, UserId},
};
use crate::{context::AppContext, import::commands::CreateImportSessionCommand};
use crate::import::commands::CreateImportSessionCommand;
pub struct CreateSessionResult {
pub session_id: ImportSessionId,
@@ -14,18 +17,14 @@ pub struct CreateSessionResult {
}
pub async fn execute(
ctx: &AppContext,
import_session: Arc<dyn ImportSessionRepository>,
document_parser: Arc<dyn DocumentParser>,
cmd: CreateImportSessionCommand,
) -> Result<CreateSessionResult, DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
ctx.repos
.import_session
.delete_expired_for_user(&user_id)
.await?;
import_session.delete_expired_for_user(&user_id).await?;
let parsed = ctx
.services
.document_parser
let parsed = document_parser
.parse(&cmd.bytes, cmd.format)
.map_err(|e| DomainError::ValidationError(e.to_string()))?;
@@ -37,7 +36,7 @@ pub async fn execute(
let session_id = session.id.clone();
session.parsed_file = Some(parsed);
ctx.repos.import_session.create(&session).await?;
import_session.create(&session).await?;
Ok(CreateSessionResult {
session_id,

View File

@@ -1,19 +1,24 @@
use crate::{context::AppContext, import::commands::DeleteImportProfileCommand};
use std::sync::Arc;
use crate::import::commands::DeleteImportProfileCommand;
use domain::{
errors::DomainError,
ports::ImportProfileRepository,
value_objects::{ImportProfileId, UserId},
};
pub async fn execute(ctx: &AppContext, cmd: DeleteImportProfileCommand) -> Result<(), DomainError> {
pub async fn execute(
import_profile: Arc<dyn ImportProfileRepository>,
cmd: DeleteImportProfileCommand,
) -> Result<(), DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
let profile_id = ImportProfileId::from_uuid(cmd.profile_id);
ctx.repos
.import_profile
import_profile
.get(&profile_id, &user_id)
.await?
.ok_or_else(|| DomainError::NotFound("import profile".into()))?;
ctx.repos.import_profile.delete(&profile_id).await
import_profile.delete(&profile_id).await
}
#[cfg(test)]

View File

@@ -1,15 +1,18 @@
use std::sync::Arc;
use chrono::NaiveDateTime;
use domain::{
errors::DomainError,
models::{ImportRow, import::RowResult},
ports::ImportSessionRepository,
value_objects::{ImportSessionId, UserId},
};
use uuid::Uuid;
use crate::{
context::AppContext,
diary::commands::{LogReviewCommand, MovieInput},
import::commands::ExecuteImportCommand,
ports::ReviewLogger,
};
pub struct ImportSummary {
@@ -19,15 +22,14 @@ pub struct ImportSummary {
}
pub async fn execute(
ctx: &AppContext,
import_session: Arc<dyn ImportSessionRepository>,
review_logger: Arc<dyn ReviewLogger>,
cmd: ExecuteImportCommand,
) -> Result<ImportSummary, DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
let session_id = ImportSessionId::from_uuid(cmd.session_id);
let confirmed_indices = cmd.confirmed_indices;
let session = ctx
.repos
.import_session
let session = import_session
.get(&session_id, &user_id)
.await?
.ok_or_else(|| DomainError::NotFound("import session".into()))?;
@@ -46,7 +48,7 @@ pub async fn execute(
}
match annotated.result {
RowResult::Valid(row) => match row_to_command(&row, user_id.value()) {
Ok(cmd) => match ctx.services.review_logger.log_review(cmd).await {
Ok(cmd) => match review_logger.log_review(cmd).await {
Ok(_) => imported += 1,
Err(e) => failed.push((idx, e.to_string())),
},
@@ -58,7 +60,7 @@ pub async fn execute(
}
}
ctx.repos.import_session.delete(&session_id).await?;
import_session.delete(&session_id).await?;
Ok(ImportSummary {
imported,

View File

@@ -1,11 +1,15 @@
use crate::context::AppContext;
use domain::{errors::DomainError, models::ImportProfile, value_objects::UserId};
use std::sync::Arc;
use domain::{
errors::DomainError, models::ImportProfile, ports::ImportProfileRepository,
value_objects::UserId,
};
pub async fn execute(
ctx: &AppContext,
import_profile: Arc<dyn ImportProfileRepository>,
user_id: &UserId,
) -> Result<Vec<ImportProfile>, DomainError> {
ctx.repos.import_profile.list_for_user(user_id).await
import_profile.list_for_user(user_id).await
}
#[cfg(test)]

View File

@@ -1,21 +1,23 @@
use crate::{context::AppContext, import::commands::SaveImportProfileCommand};
use std::sync::Arc;
use crate::import::commands::SaveImportProfileCommand;
use chrono::Utc;
use domain::{
errors::DomainError,
models::ImportProfile,
ports::{ImportProfileRepository, ImportSessionRepository},
value_objects::{ImportProfileId, ImportSessionId, UserId},
};
pub async fn execute(
ctx: &AppContext,
import_session: Arc<dyn ImportSessionRepository>,
import_profile: Arc<dyn ImportProfileRepository>,
cmd: SaveImportProfileCommand,
) -> Result<ImportProfileId, DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
let session_id = ImportSessionId::from_uuid(cmd.session_id);
let session = ctx
.repos
.import_session
let session = import_session
.get(&session_id, &user_id)
.await?
.ok_or_else(|| DomainError::NotFound("import session".into()))?;
@@ -30,7 +32,7 @@ pub async fn execute(
Utc::now().naive_utc(),
);
let id = profile.id.clone();
ctx.repos.import_profile.save(&profile).await?;
import_profile.save(&profile).await?;
Ok(id)
}

View File

@@ -8,7 +8,7 @@ use domain::{
import::{ImportRow, ParsedFile, RowResult},
},
ports::{DocumentParser, MovieRepository},
testing::InMemoryMovieRepository,
testing::{InMemoryImportSessionRepository, InMemoryMovieRepository},
value_objects::{ExternalMetadataId, MovieTitle, ReleaseYear},
};
@@ -21,11 +21,13 @@ use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn applies_mapping_to_session() {
let ctx = TestContextBuilder::new().build();
let sessions = InMemoryImportSessionRepository::new();
let b = TestContextBuilder::new();
let user_id = Uuid::new_v4();
let session = create_session::execute(
&ctx,
Arc::clone(&sessions) as _,
b.document_parser.clone(),
CreateImportSessionCommand {
user_id,
bytes: b"title\nTest".to_vec(),
@@ -36,7 +38,9 @@ async fn applies_mapping_to_session() {
.unwrap();
let rows = apply_mapping::execute(
&ctx,
Arc::clone(&sessions) as _,
b.document_parser.clone(),
b.movie_repo.clone(),
ApplyImportMappingCommand {
user_id,
session_id: session.session_id.value(),
@@ -51,10 +55,13 @@ async fn applies_mapping_to_session() {
#[tokio::test]
async fn fails_when_session_not_found() {
let ctx = TestContextBuilder::new().build();
let sessions = InMemoryImportSessionRepository::new();
let b = TestContextBuilder::new();
let result = apply_mapping::execute(
&ctx,
Arc::clone(&sessions) as _,
b.document_parser.clone(),
b.movie_repo.clone(),
ApplyImportMappingCommand {
user_id: Uuid::new_v4(),
session_id: Uuid::new_v4(),
@@ -102,6 +109,7 @@ impl DocumentParser for DuplicateTestParser {
#[tokio::test]
async fn marks_duplicate_by_external_id() {
let movies = InMemoryMovieRepository::new();
let sessions = InMemoryImportSessionRepository::new();
let ext_id = ExternalMetadataId::new("tt1234567".into()).unwrap();
let movie = Movie::new(
@@ -113,23 +121,20 @@ async fn marks_duplicate_by_external_id() {
);
movies.upsert_movie(&movie).await.unwrap();
let parser = DuplicateTestParser {
let parser = Arc::new(DuplicateTestParser {
rows: vec![ImportRow {
title: Some("Known Movie".into()),
release_year: Some("2020".into()),
external_metadata_id: Some("tt1234567".into()),
..ImportRow::default()
}],
};
});
let ctx = TestContextBuilder::new()
.with_movies(Arc::clone(&movies) as _)
.with_document_parser(Arc::new(parser) as _)
.build();
let user_id = Uuid::new_v4();
let session = create_session::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::clone(&parser) as _,
CreateImportSessionCommand {
user_id,
bytes: b"title\nKnown Movie".to_vec(),
@@ -140,7 +145,9 @@ async fn marks_duplicate_by_external_id() {
.unwrap();
let rows = apply_mapping::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::clone(&parser) as _,
Arc::clone(&movies) as _,
ApplyImportMappingCommand {
user_id,
session_id: session.session_id.value(),
@@ -157,6 +164,7 @@ async fn marks_duplicate_by_external_id() {
#[tokio::test]
async fn marks_duplicate_by_title_and_year() {
let movies = InMemoryMovieRepository::new();
let sessions = InMemoryImportSessionRepository::new();
let movie = Movie::new(
None,
@@ -167,22 +175,19 @@ async fn marks_duplicate_by_title_and_year() {
);
movies.upsert_movie(&movie).await.unwrap();
let parser = DuplicateTestParser {
let parser = Arc::new(DuplicateTestParser {
rows: vec![ImportRow {
title: Some("Duplicate Film".into()),
release_year: Some("2022".into()),
..ImportRow::default()
}],
};
});
let ctx = TestContextBuilder::new()
.with_movies(Arc::clone(&movies) as _)
.with_document_parser(Arc::new(parser) as _)
.build();
let user_id = Uuid::new_v4();
let session = create_session::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::clone(&parser) as _,
CreateImportSessionCommand {
user_id,
bytes: b"title\nDuplicate Film".to_vec(),
@@ -193,7 +198,9 @@ async fn marks_duplicate_by_title_and_year() {
.unwrap();
let rows = apply_mapping::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::clone(&parser) as _,
Arc::clone(&movies) as _,
ApplyImportMappingCommand {
user_id,
session_id: session.session_id.value(),

View File

@@ -3,19 +3,20 @@ use std::sync::Arc;
use chrono::Utc;
use domain::models::ImportProfile;
use domain::ports::{ImportProfileRepository, ImportSessionRepository};
use domain::testing::InMemoryImportProfileRepository;
use domain::testing::{InMemoryImportProfileRepository, InMemoryImportSessionRepository};
use domain::value_objects::{ImportProfileId, UserId};
use uuid::Uuid;
use crate::import::{apply_profile, commands::ApplyImportProfileCommand};
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn fails_when_profile_not_found() {
let ctx = TestContextBuilder::new().build();
let profiles = InMemoryImportProfileRepository::new();
let sessions = InMemoryImportSessionRepository::new();
let result = apply_profile::execute(
&ctx,
Arc::clone(&profiles) as _,
Arc::clone(&sessions) as _,
ApplyImportProfileCommand {
user_id: Uuid::new_v4(),
session_id: Uuid::new_v4(),
@@ -30,6 +31,7 @@ async fn fails_when_profile_not_found() {
#[tokio::test]
async fn fails_when_session_not_found() {
let profiles = InMemoryImportProfileRepository::new();
let sessions = InMemoryImportSessionRepository::new();
let user_id = Uuid::new_v4();
let profile = ImportProfile::new(
@@ -42,12 +44,9 @@ async fn fails_when_session_not_found() {
let profile_id = profile.id.clone();
profiles.save(&profile).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_profiles(Arc::clone(&profiles) as _)
.build();
let result = apply_profile::execute(
&ctx,
Arc::clone(&profiles) as _,
Arc::clone(&sessions) as _,
ApplyImportProfileCommand {
user_id,
session_id: Uuid::new_v4(),
@@ -87,13 +86,9 @@ async fn applies_profile_mappings_to_session() {
let session_id = session.id.clone();
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_profiles(Arc::clone(&profiles) as _)
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
apply_profile::execute(
&ctx,
Arc::clone(&profiles) as _,
Arc::clone(&sessions) as _,
ApplyImportProfileCommand {
user_id,
session_id: session_id.value(),

View File

@@ -3,16 +3,12 @@ use std::sync::Arc;
use domain::testing::InMemoryImportSessionRepository;
use crate::import::cleanup;
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn returns_zero_when_nothing_expired() {
let sessions = InMemoryImportSessionRepository::new();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = cleanup::execute(&ctx).await.unwrap();
let result = cleanup::execute(Arc::clone(&sessions) as _).await.unwrap();
assert_eq!(result, 0);
}

View File

@@ -1,14 +1,20 @@
use std::sync::Arc;
use uuid::Uuid;
use domain::testing::InMemoryImportSessionRepository;
use crate::import::{commands::CreateImportSessionCommand, create_session};
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn creates_session_with_parsed_file() {
let ctx = TestContextBuilder::new().build();
let sessions = InMemoryImportSessionRepository::new();
let b = TestContextBuilder::new();
let result = create_session::execute(
&ctx,
Arc::clone(&sessions) as _,
b.document_parser.clone(),
CreateImportSessionCommand {
user_id: Uuid::new_v4(),
bytes: b"col1\nval1".to_vec(),

View File

@@ -4,17 +4,13 @@ use domain::testing::InMemoryImportProfileRepository;
use uuid::Uuid;
use crate::import::{commands::DeleteImportProfileCommand, delete_profile};
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn fails_when_profile_not_found() {
let profiles = InMemoryImportProfileRepository::new();
let ctx = TestContextBuilder::new()
.with_import_profiles(Arc::clone(&profiles) as _)
.build();
let result = delete_profile::execute(
&ctx,
Arc::clone(&profiles) as _,
DeleteImportProfileCommand {
user_id: Uuid::new_v4(),
profile_id: Uuid::new_v4(),

View File

@@ -9,7 +9,7 @@ use uuid::Uuid;
use crate::import::commands::ExecuteImportCommand;
use crate::import::execute;
use crate::test_helpers::TestContextBuilder;
use crate::test_helpers::NoopReviewLogger;
fn make_session_with_rows(user_id: UserId, session_id: ImportSessionId) -> ImportSession {
let now = Utc::now().naive_utc();
@@ -52,12 +52,9 @@ async fn imports_confirmed_rows() {
let session = make_session_with_rows(UserId::from_uuid(uid), sid.clone());
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -81,12 +78,9 @@ async fn skips_unconfirmed_rows() {
let session = make_session_with_rows(UserId::from_uuid(uid), sid.clone());
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -102,9 +96,11 @@ async fn skips_unconfirmed_rows() {
#[tokio::test]
async fn fails_when_session_not_found() {
let ctx = TestContextBuilder::new().build();
let sessions = InMemoryImportSessionRepository::new();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: Uuid::new_v4(),
session_id: Uuid::new_v4(),
@@ -138,12 +134,9 @@ async fn handles_datetime_format() {
}]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -179,12 +172,9 @@ async fn fails_on_invalid_rating() {
}]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -220,12 +210,9 @@ async fn fails_on_missing_watched_at() {
}]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -261,12 +248,9 @@ async fn imports_row_with_external_metadata_id() {
}]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -302,12 +286,9 @@ async fn imports_row_with_director_and_comment() {
}]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -343,12 +324,9 @@ async fn handles_space_separated_datetime_format() {
}]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -379,12 +357,9 @@ async fn reports_invalid_row_result_errors() {
}]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -422,12 +397,9 @@ async fn fails_on_missing_rating() {
}]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -464,12 +436,9 @@ async fn fails_on_unparseable_date() {
}]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -506,12 +475,9 @@ async fn imports_row_without_release_year() {
}]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),
@@ -535,12 +501,9 @@ async fn deletes_session_after_import() {
sessions.create(&session).await.unwrap();
assert_eq!(sessions.count(), 1);
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
execute::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::new(NoopReviewLogger),
ExecuteImportCommand {
user_id: uid,
session_id: sid.value(),

View File

@@ -5,17 +5,15 @@ use domain::value_objects::UserId;
use uuid::Uuid;
use crate::import::list_profiles;
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn returns_empty_when_no_profiles() {
let profiles = InMemoryImportProfileRepository::new();
let ctx = TestContextBuilder::new()
.with_import_profiles(Arc::clone(&profiles) as _)
.build();
let user_id = UserId::from_uuid(Uuid::new_v4());
let result = list_profiles::execute(&ctx, &user_id).await.unwrap();
let result = list_profiles::execute(Arc::clone(&profiles) as _, &user_id)
.await
.unwrap();
assert!(result.is_empty());
}

View File

@@ -3,22 +3,20 @@ use std::sync::Arc;
use chrono::Utc;
use domain::models::ImportSession;
use domain::ports::ImportSessionRepository;
use domain::testing::InMemoryImportSessionRepository;
use domain::testing::{InMemoryImportProfileRepository, InMemoryImportSessionRepository};
use domain::value_objects::{ImportSessionId, UserId};
use uuid::Uuid;
use crate::import::{commands::SaveImportProfileCommand, save_profile};
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn fails_when_session_not_found() {
let sessions = InMemoryImportSessionRepository::new();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let profiles = InMemoryImportProfileRepository::new();
let result = save_profile::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::clone(&profiles) as _,
SaveImportProfileCommand {
user_id: Uuid::new_v4(),
session_id: Uuid::new_v4(),
@@ -33,6 +31,7 @@ async fn fails_when_session_not_found() {
#[tokio::test]
async fn saves_profile_from_session() {
let sessions = InMemoryImportSessionRepository::new();
let profiles = InMemoryImportProfileRepository::new();
let user_id = Uuid::new_v4();
let sid = ImportSessionId::generate();
@@ -44,12 +43,9 @@ async fn saves_profile_from_session() {
session.field_mappings = Some(vec![]);
sessions.create(&session).await.unwrap();
let ctx = TestContextBuilder::new()
.with_import_sessions(Arc::clone(&sessions) as _)
.build();
let result = save_profile::execute(
&ctx,
Arc::clone(&sessions) as _,
Arc::clone(&profiles) as _,
SaveImportProfileCommand {
user_id,
session_id: sid.value(),

View File

@@ -1,14 +1,11 @@
use std::sync::Arc;
use chrono::Duration;
use domain::errors::DomainError;
use domain::{errors::DomainError, ports::WatchEventRepository};
use crate::context::AppContext;
pub async fn execute(ctx: &AppContext) -> Result<u64, DomainError> {
pub async fn execute(watch_event: Arc<dyn WatchEventRepository>) -> Result<u64, DomainError> {
let cutoff = chrono::Utc::now().naive_utc() - Duration::days(30);
ctx.repos
.watch_event
.delete_non_pending_older_than(cutoff)
.await
watch_event.delete_non_pending_older_than(cutoff).await
}
#[cfg(test)]

View File

@@ -1,24 +1,29 @@
use std::sync::Arc;
use domain::{
errors::DomainError,
models::WatchEventStatus,
ports::WatchEventRepository,
value_objects::{UserId, WatchEventId},
};
use crate::{
context::AppContext,
diary::commands::{LogReviewCommand, MovieInput},
integrations::commands::ConfirmWatchEventsCommand,
ports::ReviewLogger,
};
pub async fn execute(ctx: &AppContext, cmd: ConfirmWatchEventsCommand) -> Result<u32, DomainError> {
pub async fn execute(
watch_event: Arc<dyn WatchEventRepository>,
review_logger: Arc<dyn ReviewLogger>,
cmd: ConfirmWatchEventsCommand,
) -> Result<u32, DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
let mut confirmed = 0u32;
for c in cmd.confirmations {
let event_id = WatchEventId::from_uuid(c.watch_event_id);
let event = ctx
.repos
.watch_event
let event = watch_event
.get_by_id(&event_id)
.await?
.ok_or_else(|| DomainError::NotFound(format!("WatchEvent {}", c.watch_event_id)))?;
@@ -53,10 +58,9 @@ pub async fn execute(ctx: &AppContext, cmd: ConfirmWatchEventsCommand) -> Result
watched_at: *event.watched_at(),
};
ctx.services.review_logger.log_review(review_cmd).await?;
review_logger.log_review(review_cmd).await?;
ctx.repos
.watch_event
watch_event
.update_status(&event_id, WatchEventStatus::Confirmed)
.await?;

View File

@@ -0,0 +1,9 @@
use std::sync::Arc;
use domain::ports::{EventPublisher, WatchEventRepository, WebhookTokenRepository};
pub struct IngestWatchEventDeps {
pub webhook_token: Arc<dyn WebhookTokenRepository>,
pub watch_event: Arc<dyn WatchEventRepository>,
pub event_publisher: Arc<dyn EventPublisher>,
}

View File

@@ -1,12 +1,18 @@
use std::sync::Arc;
use domain::{
errors::DomainError,
models::WatchEventStatus,
ports::WatchEventRepository,
value_objects::{UserId, WatchEventId},
};
use crate::{context::AppContext, integrations::commands::DismissWatchEventsCommand};
use crate::integrations::commands::DismissWatchEventsCommand;
pub async fn execute(ctx: &AppContext, cmd: DismissWatchEventsCommand) -> Result<u32, DomainError> {
pub async fn execute(
watch_event: Arc<dyn WatchEventRepository>,
cmd: DismissWatchEventsCommand,
) -> Result<u32, DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
if cmd.event_ids.is_empty() {
return Ok(0);
@@ -18,7 +24,7 @@ pub async fn execute(ctx: &AppContext, cmd: DismissWatchEventsCommand) -> Result
.map(|id| WatchEventId::from_uuid(*id))
.collect();
let events = ctx.repos.watch_event.get_by_ids(&ids).await?;
let events = watch_event.get_by_ids(&ids).await?;
if events.len() != ids.len() {
return Err(DomainError::NotFound(
@@ -31,9 +37,7 @@ pub async fn execute(ctx: &AppContext, cmd: DismissWatchEventsCommand) -> Result
}
}
let count = ctx
.repos
.watch_event
let count = watch_event
.update_status_batch(&ids, WatchEventStatus::Dismissed)
.await?;

View File

@@ -1,7 +1,11 @@
use domain::{errors::DomainError, models::WebhookToken, value_objects::UserId};
use std::sync::Arc;
use domain::{
errors::DomainError, models::WebhookToken, ports::WebhookTokenRepository, value_objects::UserId,
};
use sha2::{Digest, Sha256};
use crate::{context::AppContext, integrations::commands::GenerateWebhookTokenCommand};
use crate::integrations::commands::GenerateWebhookTokenCommand;
pub struct GeneratedWebhookToken {
pub token_plaintext: String,
@@ -9,7 +13,7 @@ pub struct GeneratedWebhookToken {
}
pub async fn execute(
ctx: &AppContext,
webhook_token: Arc<dyn WebhookTokenRepository>,
cmd: GenerateWebhookTokenCommand,
) -> Result<GeneratedWebhookToken, DomainError> {
let plaintext = generate_random_token();
@@ -18,7 +22,7 @@ pub async fn execute(
let user_id = UserId::from_uuid(cmd.user_id);
let token = WebhookToken::new(user_id, hash, cmd.provider, cmd.label);
ctx.repos.webhook_token.save(&token).await?;
webhook_token.save(&token).await?;
Ok(GeneratedWebhookToken {
token_plaintext: plaintext,

View File

@@ -1,13 +1,17 @@
use domain::{errors::DomainError, models::WatchEvent, value_objects::UserId};
use std::sync::Arc;
use crate::{context::AppContext, integrations::queries::GetWatchQueueQuery};
use domain::{
errors::DomainError, models::WatchEvent, ports::WatchEventRepository, value_objects::UserId,
};
use crate::integrations::queries::GetWatchQueueQuery;
pub async fn execute(
ctx: &AppContext,
watch_event: Arc<dyn WatchEventRepository>,
query: GetWatchQueueQuery,
) -> Result<Vec<WatchEvent>, DomainError> {
let user_id = UserId::from_uuid(query.user_id);
ctx.repos.watch_event.list_pending(&user_id).await
watch_event.list_pending(&user_id).await
}
#[cfg(test)]

View File

@@ -1,13 +1,17 @@
use domain::{errors::DomainError, models::WebhookToken, value_objects::UserId};
use std::sync::Arc;
use crate::{context::AppContext, integrations::queries::GetWebhookTokensQuery};
use domain::{
errors::DomainError, models::WebhookToken, ports::WebhookTokenRepository, value_objects::UserId,
};
use crate::integrations::queries::GetWebhookTokensQuery;
pub async fn execute(
ctx: &AppContext,
webhook_token: Arc<dyn WebhookTokenRepository>,
query: GetWebhookTokensQuery,
) -> Result<Vec<WebhookToken>, DomainError> {
let user_id = UserId::from_uuid(query.user_id);
ctx.repos.webhook_token.list_by_user(&user_id).await
webhook_token.list_by_user(&user_id).await
}
#[cfg(test)]

View File

@@ -3,26 +3,21 @@ use domain::{
errors::DomainError, events::DomainEvent, models::WatchEvent, ports::MediaServerParser,
};
use crate::{context::AppContext, integrations::commands::IngestWatchEventCommand};
use crate::integrations::{commands::IngestWatchEventCommand, deps::IngestWatchEventDeps};
pub async fn execute(
ctx: &AppContext,
deps: &IngestWatchEventDeps,
cmd: IngestWatchEventCommand,
parser: &dyn MediaServerParser,
) -> Result<(), DomainError> {
let token_hash = super::generate_token::hash_token(&cmd.token);
let webhook_token = ctx
.repos
let webhook_token = deps
.webhook_token
.find_by_token_hash(&token_hash)
.await?
.ok_or_else(|| DomainError::Unauthorized("invalid webhook token".into()))?;
let _ = ctx
.repos
.webhook_token
.touch_last_used(webhook_token.id())
.await;
let _ = deps.webhook_token.touch_last_used(webhook_token.id()).await;
let parsed = match parser.parse_playback_event(&cmd.raw_payload)? {
Some(event) => event,
@@ -34,8 +29,7 @@ pub async fn execute(
if let Some(ref ext_id) = external_metadata_id {
let one_hour_ago = chrono::Utc::now().naive_utc() - Duration::hours(1);
if ctx
.repos
if deps
.watch_event
.find_duplicate(&user_id, ext_id, one_hour_ago)
.await?
@@ -55,10 +49,9 @@ pub async fn execute(
None,
);
ctx.repos.watch_event.save(&event).await?;
deps.watch_event.save(&event).await?;
let _ = ctx
.services
let _ = deps
.event_publisher
.publish(&DomainEvent::WatchEventIngested {
user_id: event.user_id().clone(),

View File

@@ -1,6 +1,7 @@
pub mod cleanup;
pub mod commands;
pub mod confirm;
pub mod deps;
pub mod dismiss;
pub mod generate_token;
pub mod get_queue;

View File

@@ -1,14 +1,20 @@
use std::sync::Arc;
use domain::{
errors::DomainError,
ports::WebhookTokenRepository,
value_objects::{UserId, WebhookTokenId},
};
use crate::{context::AppContext, integrations::commands::RevokeWebhookTokenCommand};
use crate::integrations::commands::RevokeWebhookTokenCommand;
pub async fn execute(ctx: &AppContext, cmd: RevokeWebhookTokenCommand) -> Result<(), DomainError> {
pub async fn execute(
webhook_token: Arc<dyn WebhookTokenRepository>,
cmd: RevokeWebhookTokenCommand,
) -> Result<(), DomainError> {
let user_id = UserId::from_uuid(cmd.user_id);
let token_id = WebhookTokenId::from_uuid(cmd.token_id);
ctx.repos.webhook_token.delete(&token_id, &user_id).await
webhook_token.delete(&token_id, &user_id).await
}
#[cfg(test)]

View File

@@ -1,11 +1,15 @@
use std::sync::Arc;
use domain::ports::WatchEventRepository;
use domain::testing::InMemoryWatchEventRepository;
use crate::integrations::cleanup;
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn returns_zero_when_nothing_to_clean() {
let ctx = TestContextBuilder::new().build();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let count = cleanup::execute(&ctx).await.unwrap();
let count = cleanup::execute(watch_events).await.unwrap();
assert_eq!(count, 0);
}

View File

@@ -8,12 +8,15 @@ use uuid::Uuid;
use crate::integrations::commands::{ConfirmWatchEventsCommand, WatchEventConfirmation};
use crate::integrations::confirm;
use crate::test_helpers::TestContextBuilder;
use crate::test_helpers::NoopReviewLogger;
fn noop_logger() -> Arc<dyn crate::ports::ReviewLogger> {
Arc::new(NoopReviewLogger)
}
#[tokio::test]
async fn confirms_watch_event_via_review_logger() {
let watch_events = InMemoryWatchEventRepository::new();
let events = NoopEventPublisher::new();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let uid = Uuid::new_v4();
let event = WatchEvent::new(
@@ -28,13 +31,9 @@ async fn confirms_watch_event_via_review_logger() {
let event_id = event.id().value();
watch_events.save(&event).await.unwrap();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&watch_events) as _)
.with_event_publisher(Arc::clone(&events) as _)
.build();
let result = confirm::execute(
&ctx,
Arc::clone(&watch_events),
noop_logger(),
ConfirmWatchEventsCommand {
user_id: uid,
confirmations: vec![WatchEventConfirmation {
@@ -52,10 +51,11 @@ async fn confirms_watch_event_via_review_logger() {
#[tokio::test]
async fn empty_confirmations_returns_zero() {
let ctx = TestContextBuilder::new().build();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let result = confirm::execute(
&ctx,
Arc::clone(&watch_events),
noop_logger(),
ConfirmWatchEventsCommand {
user_id: Uuid::new_v4(),
confirmations: vec![],
@@ -69,8 +69,7 @@ async fn empty_confirmations_returns_zero() {
#[tokio::test]
async fn confirms_event_with_external_metadata_id_and_no_movie_id() {
let watch_events = InMemoryWatchEventRepository::new();
let events = NoopEventPublisher::new();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let uid = Uuid::new_v4();
let event = WatchEvent::new(
@@ -85,13 +84,9 @@ async fn confirms_event_with_external_metadata_id_and_no_movie_id() {
let event_id = event.id().value();
watch_events.save(&event).await.unwrap();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&watch_events) as _)
.with_event_publisher(Arc::clone(&events) as _)
.build();
let result = confirm::execute(
&ctx,
Arc::clone(&watch_events),
noop_logger(),
ConfirmWatchEventsCommand {
user_id: uid,
confirmations: vec![WatchEventConfirmation {
@@ -109,7 +104,7 @@ async fn confirms_event_with_external_metadata_id_and_no_movie_id() {
#[tokio::test]
async fn rejects_other_users_event() {
let watch_events = InMemoryWatchEventRepository::new();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let owner = Uuid::new_v4();
let intruder = Uuid::new_v4();
@@ -125,12 +120,9 @@ async fn rejects_other_users_event() {
let event_id = event.id().value();
watch_events.save(&event).await.unwrap();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&watch_events) as _)
.build();
let result = confirm::execute(
&ctx,
Arc::clone(&watch_events),
noop_logger(),
ConfirmWatchEventsCommand {
user_id: intruder,
confirmations: vec![WatchEventConfirmation {
@@ -147,10 +139,11 @@ async fn rejects_other_users_event() {
#[tokio::test]
async fn fails_when_event_not_found() {
let ctx = TestContextBuilder::new().build();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let result = confirm::execute(
&ctx,
Arc::clone(&watch_events),
noop_logger(),
ConfirmWatchEventsCommand {
user_id: Uuid::new_v4(),
confirmations: vec![WatchEventConfirmation {
@@ -167,7 +160,7 @@ async fn fails_when_event_not_found() {
#[tokio::test]
async fn confirms_event_with_movie_id() {
let watch_events = InMemoryWatchEventRepository::new();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let events = NoopEventPublisher::new();
let uid = Uuid::new_v4();
let movie_uuid = Uuid::new_v4();
@@ -199,23 +192,18 @@ async fn confirms_event_with_movie_id() {
// Build a real review logger
let reviews = domain::testing::InMemoryReviewRepository::new();
let watchlist = domain::testing::InMemoryWatchlistRepository::new();
let review_logger = std::sync::Arc::new(crate::diary::review_logger::DefaultReviewLogger::new(
std::sync::Arc::clone(&movies) as _,
std::sync::Arc::clone(&reviews) as _,
std::sync::Arc::clone(&watchlist) as _,
std::sync::Arc::new(domain::testing::FakeMetadataClient) as _,
std::sync::Arc::clone(&events) as _,
));
let ctx = TestContextBuilder::new()
.with_watch_events(std::sync::Arc::clone(&watch_events) as _)
.with_event_publisher(std::sync::Arc::clone(&events) as _)
.with_movies(std::sync::Arc::clone(&movies) as _)
.with_review_logger(review_logger as _)
.build();
let review_logger: Arc<dyn crate::ports::ReviewLogger> =
Arc::new(crate::diary::review_logger::DefaultReviewLogger::new(
Arc::clone(&movies) as _,
Arc::clone(&reviews) as _,
Arc::clone(&watchlist) as _,
Arc::new(domain::testing::FakeMetadataClient) as _,
Arc::clone(&events) as _,
));
let result = confirm::execute(
&ctx,
Arc::clone(&watch_events),
review_logger,
ConfirmWatchEventsCommand {
user_id: uid,
confirmations: vec![WatchEventConfirmation {
@@ -233,8 +221,7 @@ async fn confirms_event_with_movie_id() {
#[tokio::test]
async fn confirms_event_without_movie_id_and_without_external_metadata_id() {
let watch_events = InMemoryWatchEventRepository::new();
let events = NoopEventPublisher::new();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let uid = Uuid::new_v4();
let event = WatchEvent::new(
@@ -249,13 +236,9 @@ async fn confirms_event_without_movie_id_and_without_external_metadata_id() {
let event_id = event.id().value();
watch_events.save(&event).await.unwrap();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&watch_events) as _)
.with_event_publisher(Arc::clone(&events) as _)
.build();
let result = confirm::execute(
&ctx,
Arc::clone(&watch_events),
noop_logger(),
ConfirmWatchEventsCommand {
user_id: uid,
confirmations: vec![WatchEventConfirmation {
@@ -273,8 +256,7 @@ async fn confirms_event_without_movie_id_and_without_external_metadata_id() {
#[tokio::test]
async fn confirms_multiple_events() {
let watch_events = InMemoryWatchEventRepository::new();
let events = NoopEventPublisher::new();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let uid = Uuid::new_v4();
let event1 = WatchEvent::new(
@@ -302,13 +284,9 @@ async fn confirms_multiple_events() {
watch_events.save(&event1).await.unwrap();
watch_events.save(&event2).await.unwrap();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&watch_events) as _)
.with_event_publisher(Arc::clone(&events) as _)
.build();
let result = confirm::execute(
&ctx,
Arc::clone(&watch_events),
noop_logger(),
ConfirmWatchEventsCommand {
user_id: uid,
confirmations: vec![
@@ -333,14 +311,13 @@ async fn confirms_multiple_events() {
#[tokio::test]
async fn confirms_event_without_year() {
let watch_events = InMemoryWatchEventRepository::new();
let events = NoopEventPublisher::new();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let uid = Uuid::new_v4();
let event = WatchEvent::new(
UserId::from_uuid(uid),
"No Year Movie".into(),
None, // no year
None,
None,
WatchEventSource::Jellyfin,
chrono::Utc::now().naive_utc(),
@@ -349,13 +326,9 @@ async fn confirms_event_without_year() {
let event_id = event.id().value();
watch_events.save(&event).await.unwrap();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&watch_events) as _)
.with_event_publisher(Arc::clone(&events) as _)
.build();
let result = confirm::execute(
&ctx,
Arc::clone(&watch_events),
noop_logger(),
ConfirmWatchEventsCommand {
user_id: uid,
confirmations: vec![WatchEventConfirmation {

View File

@@ -7,17 +7,13 @@ use domain::value_objects::UserId;
use uuid::Uuid;
use crate::integrations::{commands::DismissWatchEventsCommand, dismiss};
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn dismisses_empty_list_returns_zero() {
let events = InMemoryWatchEventRepository::new();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&events) as _)
.build();
let events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let result = dismiss::execute(
&ctx,
Arc::clone(&events),
DismissWatchEventsCommand {
user_id: Uuid::new_v4(),
event_ids: vec![],
@@ -31,13 +27,10 @@ async fn dismisses_empty_list_returns_zero() {
#[tokio::test]
async fn fails_when_event_not_found() {
let events = InMemoryWatchEventRepository::new();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&events) as _)
.build();
let events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let result = dismiss::execute(
&ctx,
Arc::clone(&events),
DismissWatchEventsCommand {
user_id: Uuid::new_v4(),
event_ids: vec![Uuid::new_v4()],
@@ -50,7 +43,7 @@ async fn fails_when_event_not_found() {
#[tokio::test]
async fn dismisses_existing_events() {
let watch_events = InMemoryWatchEventRepository::new();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let uid = Uuid::new_v4();
let user_id = UserId::from_uuid(uid);
@@ -77,12 +70,8 @@ async fn dismisses_existing_events() {
watch_events.save(&e1).await.unwrap();
watch_events.save(&e2).await.unwrap();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&watch_events) as _)
.build();
let result = dismiss::execute(
&ctx,
Arc::clone(&watch_events),
DismissWatchEventsCommand {
user_id: uid,
event_ids: vec![id1, id2],

View File

@@ -1,22 +1,19 @@
use std::sync::Arc;
use domain::models::WatchEventSource;
use domain::ports::WebhookTokenRepository;
use domain::testing::InMemoryWebhookTokenRepository;
use uuid::Uuid;
use crate::integrations::{commands::GenerateWebhookTokenCommand, generate_token};
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn generates_token_and_saves() {
let tokens = InMemoryWebhookTokenRepository::new();
let ctx = TestContextBuilder::new()
.with_webhook_tokens(Arc::clone(&tokens) as _)
.build();
let tokens: Arc<dyn WebhookTokenRepository> = InMemoryWebhookTokenRepository::new();
let user_id = Uuid::new_v4();
let result = generate_token::execute(
&ctx,
Arc::clone(&tokens),
GenerateWebhookTokenCommand {
user_id,
provider: WatchEventSource::Jellyfin,
@@ -28,9 +25,7 @@ async fn generates_token_and_saves() {
assert!(!result.token_plaintext.is_empty());
let saved = ctx
.repos
.webhook_token
let saved = tokens
.list_by_user(&domain::value_objects::UserId::from_uuid(user_id))
.await
.unwrap();

View File

@@ -8,17 +8,13 @@ use domain::value_objects::UserId;
use uuid::Uuid;
use crate::integrations::{get_queue, queries::GetWatchQueueQuery};
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn returns_empty_when_no_events() {
let events = InMemoryWatchEventRepository::new();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&events) as _)
.build();
let events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let result = get_queue::execute(
&ctx,
Arc::clone(&events),
GetWatchQueueQuery {
user_id: Uuid::new_v4(),
},
@@ -31,10 +27,7 @@ async fn returns_empty_when_no_events() {
#[tokio::test]
async fn returns_pending_events() {
let events = InMemoryWatchEventRepository::new();
let ctx = TestContextBuilder::new()
.with_watch_events(Arc::clone(&events) as _)
.build();
let events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let user_id = Uuid::new_v4();
let event = WatchEvent::new(
@@ -48,7 +41,7 @@ async fn returns_pending_events() {
);
events.save(&event).await.unwrap();
let result = get_queue::execute(&ctx, GetWatchQueueQuery { user_id })
let result = get_queue::execute(Arc::clone(&events), GetWatchQueueQuery { user_id })
.await
.unwrap();

View File

@@ -1,6 +1,7 @@
use std::sync::Arc;
use domain::models::WatchEventSource;
use domain::ports::WebhookTokenRepository;
use domain::testing::InMemoryWebhookTokenRepository;
use uuid::Uuid;
@@ -8,17 +9,13 @@ use crate::integrations::{
commands::GenerateWebhookTokenCommand, generate_token, get_tokens,
queries::GetWebhookTokensQuery,
};
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn returns_empty_when_no_tokens() {
let tokens = InMemoryWebhookTokenRepository::new();
let ctx = TestContextBuilder::new()
.with_webhook_tokens(Arc::clone(&tokens) as _)
.build();
let tokens: Arc<dyn WebhookTokenRepository> = InMemoryWebhookTokenRepository::new();
let result = get_tokens::execute(
&ctx,
Arc::clone(&tokens),
GetWebhookTokensQuery {
user_id: Uuid::new_v4(),
},
@@ -31,15 +28,12 @@ async fn returns_empty_when_no_tokens() {
#[tokio::test]
async fn returns_tokens_after_generate() {
let tokens = InMemoryWebhookTokenRepository::new();
let ctx = TestContextBuilder::new()
.with_webhook_tokens(Arc::clone(&tokens) as _)
.build();
let tokens: Arc<dyn WebhookTokenRepository> = InMemoryWebhookTokenRepository::new();
let user_id = Uuid::new_v4();
generate_token::execute(
&ctx,
Arc::clone(&tokens),
GenerateWebhookTokenCommand {
user_id,
provider: WatchEventSource::Jellyfin,
@@ -50,7 +44,7 @@ async fn returns_tokens_after_generate() {
.unwrap();
generate_token::execute(
&ctx,
Arc::clone(&tokens),
GenerateWebhookTokenCommand {
user_id,
provider: WatchEventSource::Plex,
@@ -60,7 +54,7 @@ async fn returns_tokens_after_generate() {
.await
.unwrap();
let result = get_tokens::execute(&ctx, GetWebhookTokensQuery { user_id })
let result = get_tokens::execute(Arc::clone(&tokens), GetWebhookTokensQuery { user_id })
.await
.unwrap();

View File

@@ -1,12 +1,15 @@
use std::sync::Arc;
use domain::models::WatchEventSource;
use domain::testing::InMemoryWebhookTokenRepository;
use domain::ports::{EventPublisher, WatchEventRepository, WebhookTokenRepository};
use domain::testing::{
InMemoryWatchEventRepository, InMemoryWebhookTokenRepository, NoopEventPublisher,
};
use uuid::Uuid;
use crate::integrations::commands::GenerateWebhookTokenCommand;
use crate::integrations::{commands::IngestWatchEventCommand, generate_token, ingest};
use crate::test_helpers::TestContextBuilder;
use crate::integrations::commands::{GenerateWebhookTokenCommand, IngestWatchEventCommand};
use crate::integrations::deps::IngestWatchEventDeps;
use crate::integrations::{generate_token, ingest};
struct FakeParser;
@@ -26,14 +29,13 @@ impl domain::ports::MediaServerParser for FakeParser {
#[tokio::test]
async fn ingests_watch_event() {
let tokens = InMemoryWebhookTokenRepository::new();
let ctx = TestContextBuilder::new()
.with_webhook_tokens(Arc::clone(&tokens) as _)
.build();
let tokens: Arc<dyn WebhookTokenRepository> = InMemoryWebhookTokenRepository::new();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let event_publisher: Arc<dyn EventPublisher> = NoopEventPublisher::new();
let user_id = Uuid::new_v4();
let generated = generate_token::execute(
&ctx,
Arc::clone(&tokens),
GenerateWebhookTokenCommand {
user_id,
provider: WatchEventSource::Jellyfin,
@@ -43,8 +45,14 @@ async fn ingests_watch_event() {
.await
.unwrap();
let deps = IngestWatchEventDeps {
webhook_token: Arc::clone(&tokens),
watch_event: Arc::clone(&watch_events),
event_publisher: Arc::clone(&event_publisher),
};
let result = ingest::execute(
&ctx,
&deps,
IngestWatchEventCommand {
token: generated.token_plaintext,
raw_payload: vec![],
@@ -59,10 +67,18 @@ async fn ingests_watch_event() {
#[tokio::test]
async fn rejects_invalid_token() {
let ctx = TestContextBuilder::new().build();
let tokens: Arc<dyn WebhookTokenRepository> = InMemoryWebhookTokenRepository::new();
let watch_events: Arc<dyn WatchEventRepository> = InMemoryWatchEventRepository::new();
let event_publisher: Arc<dyn EventPublisher> = NoopEventPublisher::new();
let deps = IngestWatchEventDeps {
webhook_token: Arc::clone(&tokens),
watch_event: Arc::clone(&watch_events),
event_publisher: Arc::clone(&event_publisher),
};
let result = ingest::execute(
&ctx,
&deps,
IngestWatchEventCommand {
token: "bad-token".into(),
raw_payload: vec![],

View File

@@ -1,6 +1,7 @@
use std::sync::Arc;
use domain::models::WatchEventSource;
use domain::ports::WebhookTokenRepository;
use domain::testing::InMemoryWebhookTokenRepository;
use uuid::Uuid;
@@ -10,19 +11,15 @@ use crate::integrations::{
queries::GetWebhookTokensQuery,
revoke_token,
};
use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn revokes_existing_token() {
let tokens = InMemoryWebhookTokenRepository::new();
let ctx = TestContextBuilder::new()
.with_webhook_tokens(Arc::clone(&tokens) as _)
.build();
let tokens: Arc<dyn WebhookTokenRepository> = InMemoryWebhookTokenRepository::new();
let user_id = Uuid::new_v4();
let generated = generate_token::execute(
&ctx,
Arc::clone(&tokens),
GenerateWebhookTokenCommand {
user_id,
provider: WatchEventSource::Jellyfin,
@@ -34,11 +31,14 @@ async fn revokes_existing_token() {
let token_id = generated.token.id().value();
revoke_token::execute(&ctx, RevokeWebhookTokenCommand { user_id, token_id })
.await
.unwrap();
revoke_token::execute(
Arc::clone(&tokens),
RevokeWebhookTokenCommand { user_id, token_id },
)
.await
.unwrap();
let remaining = get_tokens::execute(&ctx, GetWebhookTokensQuery { user_id })
let remaining = get_tokens::execute(Arc::clone(&tokens), GetWebhookTokensQuery { user_id })
.await
.unwrap();

View File

@@ -1,17 +1,27 @@
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use domain::{errors::DomainError, events::DomainEvent, ports::PeriodicJob};
use crate::context::AppContext;
use domain::{
errors::DomainError,
events::DomainEvent,
ports::{EventPublisher, MovieProfileRepository, PeriodicJob},
};
pub struct EnrichmentStalenessJob {
ctx: AppContext,
movie_profile: Arc<dyn MovieProfileRepository>,
event_publisher: Arc<dyn EventPublisher>,
}
impl EnrichmentStalenessJob {
pub fn new(ctx: AppContext) -> Self {
Self { ctx }
pub fn new(
movie_profile: Arc<dyn MovieProfileRepository>,
event_publisher: Arc<dyn EventPublisher>,
) -> Self {
Self {
movie_profile,
event_publisher,
}
}
}
@@ -22,7 +32,7 @@ impl PeriodicJob for EnrichmentStalenessJob {
}
async fn run(&self) -> Result<(), DomainError> {
let stale = self.ctx.repos.movie_profile.list_stale().await?;
let stale = self.movie_profile.list_stale().await?;
if stale.is_empty() {
return Ok(());
}
@@ -32,7 +42,7 @@ impl PeriodicJob for EnrichmentStalenessJob {
movie_id,
external_metadata_id,
};
self.ctx.services.event_publisher.publish(&event).await?;
self.event_publisher.publish(&event).await?;
}
Ok(())
}

View File

@@ -1,17 +1,19 @@
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use domain::{errors::DomainError, ports::PeriodicJob};
use crate::context::AppContext;
use domain::{
errors::DomainError,
ports::{ImportSessionRepository, PeriodicJob},
};
pub struct ImportSessionCleanupJob {
ctx: AppContext,
import_session: Arc<dyn ImportSessionRepository>,
}
impl ImportSessionCleanupJob {
pub fn new(ctx: AppContext) -> Self {
Self { ctx }
pub fn new(import_session: Arc<dyn ImportSessionRepository>) -> Self {
Self { import_session }
}
}
@@ -22,7 +24,7 @@ impl PeriodicJob for ImportSessionCleanupJob {
}
async fn run(&self) -> Result<(), DomainError> {
let n = crate::import::cleanup::execute(&self.ctx).await?;
let n = crate::import::cleanup::execute(self.import_session.clone()).await?;
tracing::info!("import session cleanup: removed {} expired sessions", n);
Ok(())
}

View File

@@ -1,28 +1,30 @@
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use domain::{errors::DomainError, ports::PeriodicJob};
use crate::context::AppContext;
use domain::{
errors::DomainError,
ports::{PeriodicJob, RefreshSessionRepository},
};
pub struct RefreshSessionCleanupJob {
ctx: AppContext,
refresh_session: Arc<dyn RefreshSessionRepository>,
}
impl RefreshSessionCleanupJob {
pub fn new(ctx: AppContext) -> Self {
Self { ctx }
pub fn new(refresh_session: Arc<dyn RefreshSessionRepository>) -> Self {
Self { refresh_session }
}
}
#[async_trait]
impl PeriodicJob for RefreshSessionCleanupJob {
fn interval(&self) -> Duration {
Duration::from_secs(86400)
Duration::from_secs(3600)
}
async fn run(&self) -> Result<(), DomainError> {
let n = self.ctx.repos.refresh_session.delete_expired().await?;
let n = self.refresh_session.delete_expired().await?;
if n > 0 {
tracing::info!("refresh session cleanup: removed {n} expired sessions");
}

View File

@@ -1,17 +1,19 @@
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use domain::{errors::DomainError, ports::PeriodicJob};
use crate::context::AppContext;
use domain::{
errors::DomainError,
ports::{PeriodicJob, WatchEventRepository},
};
pub struct WatchEventCleanupJob {
ctx: AppContext,
watch_event: Arc<dyn WatchEventRepository>,
}
impl WatchEventCleanupJob {
pub fn new(ctx: AppContext) -> Self {
Self { ctx }
pub fn new(watch_event: Arc<dyn WatchEventRepository>) -> Self {
Self { watch_event }
}
}
@@ -22,7 +24,7 @@ impl PeriodicJob for WatchEventCleanupJob {
}
async fn run(&self) -> Result<(), DomainError> {
let n = crate::integrations::cleanup::execute(&self.ctx).await?;
let n = crate::integrations::cleanup::execute(self.watch_event.clone()).await?;
if n > 0 {
tracing::info!("watch event cleanup: removed {n} old entries");
}

View File

@@ -1,18 +1,30 @@
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use chrono::Datelike;
use domain::{errors::DomainError, ports::PeriodicJob};
use crate::context::AppContext;
use domain::{
errors::DomainError,
ports::{EventPublisher, PeriodicJob, UserRepository, WrapUpRepository},
};
pub struct WrapUpAutoGenerateJob {
ctx: AppContext,
user: Arc<dyn UserRepository>,
wrapup_repo: Arc<dyn WrapUpRepository>,
event_publisher: Arc<dyn EventPublisher>,
}
impl WrapUpAutoGenerateJob {
pub fn new(ctx: AppContext) -> Self {
Self { ctx }
pub fn new(
user: Arc<dyn UserRepository>,
wrapup_repo: Arc<dyn WrapUpRepository>,
event_publisher: Arc<dyn EventPublisher>,
) -> Self {
Self {
user,
wrapup_repo,
event_publisher,
}
}
}
@@ -33,12 +45,10 @@ impl PeriodicJob for WrapUpAutoGenerateJob {
let end = chrono::NaiveDate::from_ymd_opt(year + 1, 1, 1)
.ok_or_else(|| DomainError::ValidationError("invalid date".into()))?;
let users = self.ctx.repos.user.list_with_stats().await?;
let users = self.user.list_with_stats().await?;
for user in &users {
if user.total_movies > 0 {
let existing = self
.ctx
.repos
.wrapup_repo
.find_existing(Some(user.user_id.value()), start, end)
.await?;
@@ -48,7 +58,13 @@ impl PeriodicJob for WrapUpAutoGenerateJob {
start_date: start,
end_date: end,
};
if let Err(e) = crate::wrapup::generate::execute(&self.ctx, cmd).await {
if let Err(e) = crate::wrapup::generate::execute(
self.wrapup_repo.clone(),
self.event_publisher.clone(),
cmd,
)
.await
{
tracing::warn!(
"auto-generate wrapup for user {} failed: {e}",
user.user_id.value()
@@ -58,19 +74,20 @@ impl PeriodicJob for WrapUpAutoGenerateJob {
}
}
let existing = self
.ctx
.repos
.wrapup_repo
.find_existing(None, start, end)
.await?;
let existing = self.wrapup_repo.find_existing(None, start, end).await?;
if existing.is_none() {
let cmd = crate::wrapup::commands::RequestWrapUpCommand {
user_id: None,
start_date: start,
end_date: end,
};
if let Err(e) = crate::wrapup::generate::execute(&self.ctx, cmd).await {
if let Err(e) = crate::wrapup::generate::execute(
self.wrapup_repo.clone(),
self.event_publisher.clone(),
cmd,
)
.await
{
tracing::warn!("auto-generate global wrapup failed: {e}");
}
}
@@ -80,12 +97,12 @@ impl PeriodicJob for WrapUpAutoGenerateJob {
}
pub struct WrapUpCleanupJob {
ctx: AppContext,
wrapup_repo: Arc<dyn WrapUpRepository>,
}
impl WrapUpCleanupJob {
pub fn new(ctx: AppContext) -> Self {
Self { ctx }
pub fn new(wrapup_repo: Arc<dyn WrapUpRepository>) -> Self {
Self { wrapup_repo }
}
}
@@ -97,12 +114,7 @@ impl PeriodicJob for WrapUpCleanupJob {
async fn run(&self) -> Result<(), DomainError> {
let cutoff = chrono::Utc::now().naive_utc() - chrono::Duration::days(7);
let n = self
.ctx
.repos
.wrapup_repo
.delete_failed_older_than(cutoff)
.await?;
let n = self.wrapup_repo.delete_failed_older_than(cutoff).await?;
if n > 0 {
tracing::info!("wrapup cleanup: removed {n} failed records");
}

View File

@@ -1,5 +1,4 @@
pub mod config;
pub mod context;
pub mod jobs;
pub mod ports;
pub mod worker;

View File

@@ -0,0 +1,31 @@
use std::sync::Arc;
use domain::ports::{
EventPublisher, MetadataClient, MovieProfileRepository, MovieRepository, ObjectStorage,
PersonCommand, PersonQuery, PosterFetcherClient, SearchCommand,
};
pub struct SyncPosterDeps {
pub movie: Arc<dyn MovieRepository>,
pub movie_profile: Arc<dyn MovieProfileRepository>,
pub metadata: Arc<dyn MetadataClient>,
pub poster_fetcher: Arc<dyn PosterFetcherClient>,
pub object_storage: Arc<dyn ObjectStorage>,
pub event_publisher: Arc<dyn EventPublisher>,
pub search_command: Arc<dyn SearchCommand>,
}
pub struct EnrichMovieDeps {
pub movie: Arc<dyn MovieRepository>,
pub movie_profile: Arc<dyn MovieProfileRepository>,
pub person_command: Arc<dyn PersonCommand>,
pub search_command: Arc<dyn SearchCommand>,
}
pub struct ReindexSearchDeps {
pub movie: Arc<dyn MovieRepository>,
pub movie_profile: Arc<dyn MovieProfileRepository>,
pub search_command: Arc<dyn SearchCommand>,
pub person_command: Arc<dyn PersonCommand>,
pub person_query: Arc<dyn PersonQuery>,
}

View File

@@ -1,38 +1,30 @@
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;
use crate::movies::{commands::EnrichMovieCommand, deps::EnrichMovieDeps};
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> {
pub async fn execute(deps: &EnrichMovieDeps, cmd: EnrichMovieCommand) -> Result<(), DomainError> {
// 1. Persist the enriched profile (also handles movie_cast, movie_crew, genres, keywords)
profile_repository.upsert(&cmd.profile).await?;
deps.movie_profile.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?;
deps.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 {
let Some(movie) = deps.movie.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
deps.search_command
.index(IndexableDocument::Movie {
id: cmd.movie_id.clone(),
movie: Box::new(movie),
@@ -42,7 +34,7 @@ pub async fn execute(
// 5. Index each unique person in search (no reads — persons built from in-memory data)
for person in &persons {
search_command
deps.search_command
.index(IndexableDocument::Person {
id: person.id().clone(),
person: Box::new(person.clone()),

View File

@@ -1,12 +1,13 @@
use std::sync::Arc;
use domain::{
errors::DomainError,
models::{CastMember, CrewMember, ExternalPersonId, MovieProfile, PersonId},
ports::MovieProfileRepository,
value_objects::MovieId,
};
use uuid::Uuid;
use crate::context::AppContext;
pub struct GetMovieProfileQuery {
pub movie_id: Uuid,
}
@@ -60,11 +61,11 @@ fn resolve_crew(member: &CrewMember) -> CrewMemberWithId {
}
pub async fn execute(
ctx: &AppContext,
movie_profile: Arc<dyn MovieProfileRepository>,
query: GetMovieProfileQuery,
) -> Result<Option<MovieProfileResult>, DomainError> {
let movie_id = MovieId::from_uuid(query.movie_id);
let profile = ctx.repos.movie_profile.get_by_movie_id(&movie_id).await?;
let profile = movie_profile.get_by_movie_id(&movie_id).await?;
Ok(profile.map(|p| {
let cast = p.cast.iter().map(resolve_cast).collect();

View File

@@ -1,13 +1,16 @@
use std::sync::Arc;
use domain::{
errors::DomainError,
models::collections::{PageParams, Paginated},
models::{MovieFilter, MovieSummary},
ports::MovieRepository,
};
use crate::{context::AppContext, movies::queries::GetMoviesQuery};
use crate::movies::queries::GetMoviesQuery;
pub async fn execute(
ctx: &AppContext,
movie: Arc<dyn MovieRepository>,
query: GetMoviesQuery,
) -> Result<Paginated<MovieSummary>, DomainError> {
let page = PageParams::new(query.limit, query.offset)?;
@@ -16,7 +19,7 @@ pub async fn execute(
genre: query.genre,
language: query.language,
};
ctx.repos.movie.list_movies(&page, &filter).await
movie.list_movies(&page, &filter).await
}
#[cfg(test)]

View File

@@ -1,4 +1,5 @@
pub mod commands;
pub mod deps;
pub mod discovery_indexer;
pub mod enrich_movie;
pub mod get_movie_profile;

View File

@@ -7,7 +7,7 @@ use domain::{
};
use std::sync::atomic::{AtomicBool, Ordering};
use crate::context::AppContext;
use crate::movies::deps::ReindexSearchDeps;
const BATCH_SIZE: u32 = 500;
@@ -17,10 +17,10 @@ pub struct ReindexResult {
pub persons_backfilled: u64,
}
pub async fn execute(ctx: &AppContext) -> Result<ReindexResult, DomainError> {
let movies_indexed = reindex_movies(ctx).await?;
let persons_backfilled = backfill_persons(ctx).await?;
let persons_indexed = reindex_persons(ctx).await?;
pub async fn execute(deps: &ReindexSearchDeps) -> Result<ReindexResult, DomainError> {
let movies_indexed = reindex_movies(deps).await?;
let persons_backfilled = backfill_persons(deps).await?;
let persons_indexed = reindex_persons(deps).await?;
Ok(ReindexResult {
movies_indexed,
@@ -29,12 +29,11 @@ pub async fn execute(ctx: &AppContext) -> Result<ReindexResult, DomainError> {
})
}
async fn reindex_movies(ctx: &AppContext) -> Result<u64, DomainError> {
async fn reindex_movies(deps: &ReindexSearchDeps) -> Result<u64, DomainError> {
let mut count: u64 = 0;
let mut offset: u32 = 0;
loop {
let page = ctx
.repos
let page = deps
.movie
.list_movies(
&PageParams {
@@ -47,10 +46,9 @@ async fn reindex_movies(ctx: &AppContext) -> Result<u64, DomainError> {
for summary in &page.items {
let movie_id = summary.movie.id().clone();
let profile = ctx.repos.movie_profile.get_by_movie_id(&movie_id).await?;
let profile = deps.movie_profile.get_by_movie_id(&movie_id).await?;
if let Err(e) = ctx
.repos
if let Err(e) = deps
.search_command
.index(IndexableDocument::Movie {
id: movie_id.clone(),
@@ -73,11 +71,10 @@ async fn reindex_movies(ctx: &AppContext) -> Result<u64, DomainError> {
Ok(count)
}
async fn backfill_persons(ctx: &AppContext) -> Result<u64, DomainError> {
async fn backfill_persons(deps: &ReindexSearchDeps) -> Result<u64, DomainError> {
let mut total = 0u64;
loop {
let (count, has_more) = ctx
.repos
let (count, has_more) = deps
.person_command
.backfill_from_credits_batch(BATCH_SIZE)
.await?;
@@ -90,15 +87,14 @@ async fn backfill_persons(ctx: &AppContext) -> Result<u64, DomainError> {
Ok(total)
}
async fn reindex_persons(ctx: &AppContext) -> Result<u64, DomainError> {
async fn reindex_persons(deps: &ReindexSearchDeps) -> Result<u64, DomainError> {
let mut count: u64 = 0;
let mut offset: u32 = 0;
loop {
let persons = ctx.repos.person_query.list_page(BATCH_SIZE, offset).await?;
let persons = deps.person_query.list_page(BATCH_SIZE, offset).await?;
for person in &persons {
if let Err(e) = ctx
.repos
if let Err(e) = deps
.search_command
.index(IndexableDocument::Person {
id: person.id().clone(),
@@ -121,14 +117,14 @@ async fn reindex_persons(ctx: &AppContext) -> Result<u64, DomainError> {
}
pub struct SearchReindexHandler {
ctx: AppContext,
deps: ReindexSearchDeps,
running: AtomicBool,
}
impl SearchReindexHandler {
pub fn new(ctx: AppContext) -> Self {
pub fn new(deps: ReindexSearchDeps) -> Self {
Self {
ctx,
deps,
running: AtomicBool::new(false),
}
}
@@ -147,7 +143,7 @@ impl EventHandler for SearchReindexHandler {
}
tracing::info!("search reindex started");
let result = execute(&self.ctx).await;
let result = execute(&self.deps).await;
self.running.store(false, Ordering::SeqCst);
let r = result?;

View File

@@ -5,12 +5,12 @@ use domain::{
value_objects::{MovieId, PosterPath},
};
use crate::{context::AppContext, diary::commands::SyncPosterCommand};
use crate::{diary::commands::SyncPosterCommand, movies::deps::SyncPosterDeps};
pub async fn execute(ctx: &AppContext, cmd: SyncPosterCommand) -> Result<(), DomainError> {
pub async fn execute(deps: &SyncPosterDeps, 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? {
let mut movie = match deps.movie.get_movie_by_id(&movie_id).await? {
Some(m) => m,
None => {
tracing::warn!(
@@ -30,12 +30,7 @@ pub async fn execute(ctx: &AppContext, cmd: SyncPosterCommand) -> Result<(), Dom
})?
.clone();
let poster_url = match ctx
.services
.metadata
.get_poster_url(&external_metadata_id)
.await
{
let poster_url = match deps.metadata.get_poster_url(&external_metadata_id).await {
Ok(Some(url)) => url,
Ok(None) => return Ok(()),
Err(e) => {
@@ -44,20 +39,14 @@ pub async fn execute(ctx: &AppContext, cmd: SyncPosterCommand) -> Result<(), Dom
}
};
let image_bytes = ctx
.services
.poster_fetcher
.fetch_poster_bytes(&poster_url)
.await?;
let image_bytes = deps.poster_fetcher.fetch_poster_bytes(&poster_url).await?;
let stored_path = ctx
.services
let stored_path = deps
.object_storage
.store(&movie_id.value().to_string(), &image_bytes)
.await?;
if let Err(e) = ctx
.services
if let Err(e) = deps
.event_publisher
.publish(&DomainEvent::ImageStored {
key: stored_path.clone(),
@@ -70,19 +59,17 @@ pub async fn execute(ctx: &AppContext, cmd: SyncPosterCommand) -> Result<(), Dom
let poster_path = PosterPath::new(stored_path)?;
movie.update_poster(poster_path);
ctx.repos.movie.upsert_movie(&movie).await?;
deps.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
let profile = deps
.movie_profile
.get_by_movie_id(&movie_id)
.await
.ok()
.flatten();
if let Err(e) = ctx
.repos
if let Err(e) = deps
.search_command
.index(IndexableDocument::Movie {
id: movie_id.clone(),

View File

@@ -11,15 +11,13 @@ use domain::{
value_objects::{MovieId, MovieTitle, ReleaseYear},
};
use crate::movies::{commands::EnrichMovieCommand, enrich_movie};
use crate::movies::{commands::EnrichMovieCommand, deps::EnrichMovieDeps, enrich_movie};
#[tokio::test]
async fn stores_profile_and_indexes() {
let movie_repo = InMemoryMovieRepository::new();
let profile_repo = InMemoryMovieProfileRepository::new();
let search_cmd: Arc<dyn domain::ports::SearchCommand> = Arc::new(FakeSearchCommand);
// PanicPersonCommand is safe here — empty cast/crew means upsert_batch is never called
let person_cmd: Arc<dyn domain::ports::PersonCommand> = Arc::new(PanicPersonCommand);
let movie = Movie::new(
None,
@@ -51,11 +49,15 @@ async fn stores_profile_and_indexes() {
enriched_at: Utc::now(),
};
let deps = EnrichMovieDeps {
movie: movie_repo as Arc<_>,
movie_profile: Arc::clone(&profile_repo) as Arc<_>,
person_command: Arc::new(PanicPersonCommand),
search_command: Arc::new(FakeSearchCommand),
};
enrich_movie::execute(
&(movie_repo as Arc<_>),
&(profile_repo.clone() as Arc<_>),
&person_cmd,
&search_cmd,
&deps,
EnrichMovieCommand {
movie_id: movie_id.clone(),
profile,
@@ -96,8 +98,6 @@ impl domain::ports::PersonCommand for NoopPersonCommand {
async fn extracts_and_indexes_persons() {
let movie_repo = InMemoryMovieRepository::new();
let profile_repo = InMemoryMovieProfileRepository::new();
let search_cmd: Arc<dyn domain::ports::SearchCommand> = Arc::new(FakeSearchCommand);
let person_cmd: Arc<dyn domain::ports::PersonCommand> = Arc::new(NoopPersonCommand);
let movie = Movie::new(
None,
@@ -141,11 +141,15 @@ async fn extracts_and_indexes_persons() {
enriched_at: Utc::now(),
};
let deps = EnrichMovieDeps {
movie: movie_repo as Arc<_>,
movie_profile: Arc::clone(&profile_repo) as Arc<_>,
person_command: Arc::new(NoopPersonCommand),
search_command: Arc::new(FakeSearchCommand),
};
enrich_movie::execute(
&(movie_repo as Arc<_>),
&(profile_repo.clone() as Arc<_>),
&person_cmd,
&search_cmd,
&deps,
EnrichMovieCommand {
movie_id: movie_id.clone(),
profile,

View File

@@ -1,5 +1,3 @@
use std::sync::Arc;
use chrono::Utc;
use uuid::Uuid;
@@ -10,17 +8,14 @@ use domain::{
value_objects::MovieId,
};
use crate::{
movies::get_movie_profile::{self, GetMovieProfileQuery},
test_helpers::TestContextBuilder,
};
use crate::movies::get_movie_profile::{self, GetMovieProfileQuery};
#[tokio::test]
async fn returns_none_when_no_profile() {
let ctx = TestContextBuilder::new().build();
let movie_profile = InMemoryMovieProfileRepository::new();
let result = get_movie_profile::execute(
&ctx,
movie_profile,
GetMovieProfileQuery {
movie_id: Uuid::new_v4(),
},
@@ -69,12 +64,8 @@ async fn returns_profile_with_cast_and_crew() {
};
profile_repo.upsert(&profile).await.unwrap();
let ctx = TestContextBuilder::new()
.with_movie_profiles(Arc::clone(&profile_repo) as _)
.build();
let result = get_movie_profile::execute(
&ctx,
profile_repo.clone(),
GetMovieProfileQuery {
movie_id: movie_id.value(),
},

View File

@@ -1,14 +1,13 @@
use crate::{
movies::{get_movies, queries::GetMoviesQuery},
test_helpers::TestContextBuilder,
};
use domain::testing::InMemoryMovieRepository;
use crate::movies::{get_movies, queries::GetMoviesQuery};
#[tokio::test]
async fn returns_empty_when_no_movies() {
let ctx = TestContextBuilder::new().build();
let movie = InMemoryMovieRepository::new();
let result = get_movies::execute(
&ctx,
movie,
GetMoviesQuery {
limit: None,
offset: None,

View File

@@ -6,20 +6,36 @@ use domain::{
errors::DomainError,
models::Movie,
ports::{MetadataClient, MovieRepository},
testing::InMemoryMovieRepository,
testing::{
FakeSearchCommand, InMemoryMovieProfileRepository, InMemoryMovieRepository,
NoopEventPublisher, NoopObjectStorage,
},
value_objects::{ExternalMetadataId, MovieTitle, PosterUrl, ReleaseYear},
};
use crate::{
diary::commands::SyncPosterCommand, movies::sync_poster, test_helpers::TestContextBuilder,
diary::commands::SyncPosterCommand,
movies::{deps::SyncPosterDeps, sync_poster},
};
fn default_deps() -> SyncPosterDeps {
SyncPosterDeps {
movie: InMemoryMovieRepository::new(),
movie_profile: InMemoryMovieProfileRepository::new(),
metadata: Arc::new(domain::testing::FakeMetadataClient),
poster_fetcher: Arc::new(domain::testing::FakePosterFetcher),
object_storage: Arc::new(NoopObjectStorage),
event_publisher: NoopEventPublisher::new(),
search_command: Arc::new(FakeSearchCommand),
}
}
#[tokio::test]
async fn fails_when_movie_not_found() {
let ctx = TestContextBuilder::new().build();
let deps = default_deps();
let result = sync_poster::execute(
&ctx,
&deps,
SyncPosterCommand {
movie_id: Uuid::new_v4(),
},
@@ -42,11 +58,12 @@ async fn fails_when_no_external_id() {
let movie_id = movie.id().value();
movies.upsert_movie(&movie).await.unwrap();
let ctx = TestContextBuilder::new()
.with_movies(Arc::clone(&movies) as _)
.build();
let deps = SyncPosterDeps {
movie: Arc::clone(&movies) as _,
..default_deps()
};
let result = sync_poster::execute(&ctx, SyncPosterCommand { movie_id }).await;
let result = sync_poster::execute(&deps, SyncPosterCommand { movie_id }).await;
assert!(result.is_err());
}
@@ -85,12 +102,13 @@ async fn syncs_poster_for_movie_with_external_id() {
let movie_id = movie.id().value();
movies.upsert_movie(&movie).await.unwrap();
let ctx = TestContextBuilder::new()
.with_movies(Arc::clone(&movies) as _)
.with_metadata_client(Arc::new(FakeMetaWithPoster) as _)
.build();
let deps = SyncPosterDeps {
movie: Arc::clone(&movies) as _,
metadata: Arc::new(FakeMetaWithPoster) as _,
..default_deps()
};
sync_poster::execute(&ctx, SyncPosterCommand { movie_id })
sync_poster::execute(&deps, SyncPosterCommand { movie_id })
.await
.unwrap();

View File

@@ -0,0 +1,14 @@
use std::sync::Arc;
use domain::ports::{EventPublisher, PersonCommand, PersonEnrichmentClient, PersonQuery};
pub struct GetPersonDeps {
pub person_query: Arc<dyn PersonQuery>,
pub event_publisher: Arc<dyn EventPublisher>,
}
pub struct EnrichPersonDeps {
pub person_query: Arc<dyn PersonQuery>,
pub person_enrichment: Option<Arc<dyn PersonEnrichmentClient>>,
pub person_command: Arc<dyn PersonCommand>,
}

View File

@@ -1,15 +1,16 @@
use crate::context::AppContext;
use chrono::Utc;
use domain::{errors::DomainError, models::PersonId};
use super::deps::EnrichPersonDeps;
const STALENESS_DAYS: i64 = 90;
pub async fn execute(
ctx: &AppContext,
deps: &EnrichPersonDeps,
person_id: PersonId,
external_id: &str,
) -> Result<(), DomainError> {
if let Some(person) = ctx.repos.person_query.get_by_id(&person_id).await?
if let Some(person) = deps.person_query.get_by_id(&person_id).await?
&& let Some(at) = person.enriched_at()
&& (Utc::now() - at).num_days() < STALENESS_DAYS
{
@@ -17,7 +18,7 @@ pub async fn execute(
return Ok(());
}
let client = ctx.services.person_enrichment.as_ref().ok_or_else(|| {
let client = deps.person_enrichment.as_ref().ok_or_else(|| {
DomainError::InfrastructureError("person enrichment client not configured".into())
})?;
@@ -30,8 +31,7 @@ pub async fn execute(
Err(e) => return Err(e),
};
ctx.repos
.person_command
deps.person_command
.update_enrichment(&person_id, &data)
.await?;
tracing::info!(person_id = %person_id.value(), "person enriched");

View File

@@ -1,4 +1,3 @@
use crate::context::AppContext;
use chrono::Utc;
use domain::{
errors::DomainError,
@@ -6,15 +5,16 @@ use domain::{
models::{Person, PersonId},
};
use super::deps::GetPersonDeps;
const ENRICHMENT_TTL_DAYS: i64 = 90;
pub async fn execute(ctx: &AppContext, id: PersonId) -> Result<Option<Person>, DomainError> {
let person = ctx.repos.person_query.get_by_id(&id).await?;
pub async fn execute(deps: &GetPersonDeps, id: PersonId) -> Result<Option<Person>, DomainError> {
let person = deps.person_query.get_by_id(&id).await?;
if let Some(ref p) = person
&& should_enrich(p)
{
let _ = ctx
.services
let _ = deps
.event_publisher
.publish(&DomainEvent::PersonEnrichmentRequested {
person_id: id,

View File

@@ -1,4 +1,3 @@
use crate::context::AppContext;
use chrono::Utc;
use domain::{
errors::DomainError,
@@ -6,13 +5,14 @@ use domain::{
models::{Person, PersonCredits, PersonId},
};
use super::deps::GetPersonDeps;
const ENRICHMENT_TTL_DAYS: i64 = 90;
pub async fn execute(ctx: &AppContext, id: PersonId) -> Result<PersonCredits, DomainError> {
let credits = ctx.repos.person_query.get_credits(&id).await?;
pub async fn execute(deps: &GetPersonDeps, id: PersonId) -> Result<PersonCredits, DomainError> {
let credits = deps.person_query.get_credits(&id).await?;
if should_enrich(&credits.person) {
let _ = ctx
.services
let _ = deps
.event_publisher
.publish(&DomainEvent::PersonEnrichmentRequested {
person_id: id,

View File

@@ -1,3 +1,4 @@
pub mod deps;
pub mod enrich;
pub mod get;
pub mod get_credits;

View File

@@ -1,14 +1,18 @@
use domain::models::PersonId;
use domain::testing::{FakePersonQuery, NoopEventPublisher};
use std::sync::Arc;
use uuid::Uuid;
use crate::person::get;
use crate::test_helpers::TestContextBuilder;
use crate::person::{deps::GetPersonDeps, get};
#[tokio::test]
async fn returns_none_for_unknown_person() {
let ctx = TestContextBuilder::new().build();
let deps = GetPersonDeps {
person_query: Arc::new(FakePersonQuery),
event_publisher: NoopEventPublisher::new(),
};
let result = get::execute(&ctx, PersonId::from_uuid(Uuid::new_v4()))
let result = get::execute(&deps, PersonId::from_uuid(Uuid::new_v4()))
.await
.unwrap();

View File

@@ -1,14 +1,18 @@
use domain::models::PersonId;
use domain::testing::{FakePersonQuery, NoopEventPublisher};
use std::sync::Arc;
use uuid::Uuid;
use crate::person::get_credits;
use crate::test_helpers::TestContextBuilder;
use crate::person::{deps::GetPersonDeps, get_credits};
#[tokio::test]
async fn returns_empty_credits() {
let ctx = TestContextBuilder::new().build();
let deps = GetPersonDeps {
person_query: Arc::new(FakePersonQuery),
event_publisher: NoopEventPublisher::new(),
};
let result = get_credits::execute(&ctx, PersonId::from_uuid(Uuid::new_v4()))
let result = get_credits::execute(&deps, PersonId::from_uuid(Uuid::new_v4()))
.await
.unwrap();

View File

@@ -1,11 +1,15 @@
use crate::context::AppContext;
use domain::{
errors::DomainError,
models::{SearchQuery, SearchResults},
ports::SearchPort,
};
use std::sync::Arc;
pub async fn execute(ctx: &AppContext, query: SearchQuery) -> Result<SearchResults, DomainError> {
ctx.repos.search_port.search(&query).await
pub async fn execute(
search_port: Arc<dyn SearchPort>,
query: SearchQuery,
) -> Result<SearchResults, DomainError> {
search_port.search(&query).await
}
#[cfg(test)]

View File

@@ -5,9 +5,9 @@ use crate::test_helpers::TestContextBuilder;
#[tokio::test]
async fn returns_empty_results() {
let ctx = TestContextBuilder::new().build();
let b = TestContextBuilder::new();
let result = execute::execute(&ctx, SearchQuery::default())
let result = execute::execute(b.search_port.clone(), SearchQuery::default())
.await
.unwrap();

Some files were not shown because too many files have changed in this diff Show More